diff --git a/src/webapp/uvcpp_web_static.cpp b/src/webapp/uvcpp_web_static.cpp index aca9bfc1..c90cd0e5 100644 --- a/src/webapp/uvcpp_web_static.cpp +++ b/src/webapp/uvcpp_web_static.cpp @@ -252,6 +252,16 @@ enum class probe_status { * `OK` 走同一条大路，只在最后那个 if/else 链上分开。 */ STREAM, + /** + * 工作池名额没抢到 ⇒ 503。 + * + * 与其余几种的差别在于：它是**唯一一种在 worker 里才判出来的状态**。 + * 名额过去是在 `serve()` 里（投递之前）取的，于是"命中 LRU 缓存、一次 + * 磁盘读都不需要"的请求也要先抢名额，抢不到就 503 —— 而那正是绝大多数 + * 请求。现在名额只在**确实要读盘**的两条支路上取（见 `run_job`），所以 + * 它的判定点跟着搬到了 worker 里。 + */ + REJECTED, }; struct probe_result { @@ -490,6 +500,15 @@ struct uvcpp_web_static::Impl { bool need_data; ///< HEAD 不需要字节，只要元数据 std::vector probes; + /** + * @brief 本 job 当前**持有一个工作池名额**吗。 + * + * 名额从 `serve()` 搬进 `run_job()` 之后，"拿过"不再是一条必走的路 + * （命中缓存、HEAD、404 都不拿），所以归还侧必须有个凭据 —— 否则 + * `release()` 会把别人的名额还掉，闸门自己就成了漏洞。 + */ + bool slot_held; + // worker 产出 probe_status status; bool policy_rejected; @@ -509,6 +528,7 @@ struct uvcpp_web_static::Impl { next(), is_head(false), need_data(true), + slot_held(false), status(probe_status::NOT_FOUND), policy_rejected(false), served_from_cache(false), @@ -664,12 +684,36 @@ struct uvcpp_web_static::Impl { void run_job(job* j); ///< worker 线程 void finish_job(job* j); ///< loop 线程 + + /** + * @brief 为一次**确实要读盘的** job 取工作池名额。 + * + * @return true = 可以继续（要么没设闸门，要么名额已到手，`j->slot_held` 为真）； + * false = 没抢到，`j->status` 已写成 `REJECTED`，调用方立即返回。 + * + * 取的时机从 `serve()`（投递之前）搬到了这里 —— **这是本模块唯一被搬过 + * 位置的东西**，取名额这件事本身没变。原因见 `run_job()` 里缓存校验那一段。 + */ + bool acquire_slot(job* j); }; // ========================================================================= // worker 线程 // ========================================================================= +bool uvcpp_web_static::Impl::acquire_slot(job* j) { + if (!work_limit) return true; // 没设闸门 = 一律放行 + + if (!work_limit->acquire()) { + // `rejected` 的成员注释写着"只在 loop 线程加减"，而这里在 worker 上， + // 所以计数不在这里加 —— 交给 `finish_job()`（同样只写状态，不动共享计数）。 + j->status = probe_status::REJECTED; + return false; + } + j->slot_held = true; + return true; +} + void uvcpp_web_static::Impl::run_job(job* j) { probe_result pr; probe_status s = probe_with_index(j->loop, *j->cfg, j->url, pr); @@ -727,12 +771,21 @@ void uvcpp_web_static::Impl::run_job(job* j) { if (static_cast(pr.size) > static_cast(thr)) { // 一个字节都不读。元数据已经全部带在 job 上了（上面那一段赋过值）， // 所以 ETag/Last-Modified/条件请求/Range 在 loop 线程那侧照常判定。 + // + // **流式也要占名额**：它不整读，但确实在读盘 —— 闸门在这条路上的意义 + // 与整读路径完全一样（见 acquire_slot）。 + if (!acquire_slot(j)) return; j->status = probe_status::STREAM; return; } // 缓存校验：拿 loop 线程给的期望快照和刚 stat 到的真实值比。 // 对上了就**不读盘** —— 热文件的成本是一次 stat。 + // + // **命中路径不占名额**，这是本模块最关键的一条：命中时 worker 只做了一次 + // `stat`（几微秒），一个字节的磁盘读都没有，内存面也早就有 `max_cached_file_size` + // 封着。让这样一次请求去抢名额，代价是**并发一过闸门就大面积 503**，而收益 + // 是零 —— 实测并发 64 时 71.7% 的请求被拒，其中 99.97% 本来能命中缓存。 for (size_t i = 0; i < j->probes.size(); ++i) { const cache_probe& p = j->probes[i]; if (p.key == pr.final_url && p.size == pr.size && @@ -743,6 +796,9 @@ void uvcpp_web_static::Impl::run_job(job* j) { } } + // 走到这里只剩一条路：**把文件整读进内存**。这才是闸门要卡的那件事。 + if (!acquire_slot(j)) return; + std::shared_ptr buf(new std::string()); if (!read_file(j->loop, pr.fs_path, pr.size, *buf)) { UVCPP_LOG_ERROR(log_category::IO) @@ -781,6 +837,20 @@ void uvcpp_web_static::Impl::finish_job(job* j) { case probe_status::IO_ERROR: resp.server_error(); return; + case probe_status::REJECTED: + // 名额没抢到。形状与原先 `serve()` 里那一支**逐字一致** —— 只是判定的 + // 位置从"投递之前"挪到了"真要读盘之前"（见 `run_job`）。 + ++rejected; // 观测用计数，只在本（loop）线程上加 + UVCPP_LOG_WARN(log_category::STATIC) + << "工作池已满（在途 " << (work_limit ? work_limit->in_flight() : 0) + << " / " << (work_limit ? work_limit->limit() : 0) + << "），拒绝静态请求 " << j->final_url; + resp.service_unavailable(); + // `Retry-After` 是 503 的配套语义（RFC 9110 §10.2.3）：告诉对端"等 1 秒 + // 再来"。**必须给这个头** —— 没有它，客户端只能靠猜，实测最常见的反应 + // 是立刻重试，正好把已经饱和的池子压得更死。 + resp.set_header("retry-after", "1"); + return; case probe_status::OK: case probe_status::STREAM: break; // STREAM 只换产出 body 的方式，元数据与 OK 完全一样 @@ -1129,31 +1199,14 @@ void uvcpp_web_static::serve(uvcpp_web_request& req, uvcpp_web_response& resp, // ALLOW：照常往下走 } - // 工作池名额：**投递之前**拿，拿不到就一个任务都不投。 - // - // 位置在最后一条同步回绝（dotfile）之后、`new job` 之前 —— 顺序是有意的： - // - 放在前面等于"为了回一个 403/404 先占一个线程池名额"，纯浪费； - // - 放在 `queue_work` 之后就没有意义了，那时任务已经在队列里了。 + // **工作池名额不在这里取**（曾经在这里）。闸门想限制的是"并发的磁盘读"， + // 而它的计数单位过去是"请求" —— 这两者在缓存未命中时重合、在命中时不重合， + // 而命中是绝大多数。于是并发一过闸门，连"只做了一次 stat、一个字节都没读" + // 的请求也一起被 503 掉。取名额的时机因此搬到了 `run_job()` 里真正要读盘 + // 那一句之前，那里才是"确实要占一份 IO"的判定点。 // - // 拿不到名额时**不留 `next`**：响应当场填好，框架照常收尾并把它发出去， - // 与上面几条同步路径形状完全一致。 - // - // 这是个**准入**而不是排队 —— 队列在这里没有任何好处：请求已经在手上， - // 堆着不投只是把线程池的队列换成我们自己的队列，内存照爆，还多一层延迟。 - if (im->work_limit && !im->work_limit->acquire()) { - ++im->rejected; - UVCPP_LOG_WARN(log_category::STATIC) - << "工作池已满（在途 " << im->work_limit->in_flight() << " / " - << im->work_limit->limit() << "），拒绝静态请求 " << url; - resp.service_unavailable(); - // `Retry-After` 是 503 的配套语义（RFC 9110 §10.2.3）：告诉对端"等 1 秒 - // 再来"。**必须给这个头** —— 没有它，客户端只能靠猜，实测最常见的反应 - // 是立刻重试，正好把已经饱和的池子压得更死。 - resp.set_header("retry-after", "1"); - resp.end(); - return; - } - + // 随之改变的只有 503 的产出位置：从这里的同步回绝变成 `finish_job()` 里的 + // 一个 `probe_status::REJECTED` 分支，响应内容（503 + Retry-After）不变。 Impl::job* j = new Impl::job(); j->self = im; j->cfg = im->cfg; @@ -1178,10 +1231,14 @@ void uvcpp_web_static::serve(uvcpp_web_request& req, uvcpp_web_response& resp, // 这个回调**本身就是"worker 已经返回"**，所以名额在这里就该还 —— // 早还一拍，排在后面的请求就早一拍被受理。 // + // **只能还自己拿过的那一份**：名额搬进 worker 之后，"拿过"不再是必走 + // 的路（命中缓存 / HEAD / 404 都不拿），无条件 release 会把别人的名额 + // 还掉 —— 闸门自己就成了漏洞。凭据是 `j->slot_held`。 + // // 与 `serve()` 里的 `rc != 0` 分支**互斥**（libuv 的约定： // `uv_queue_work` 返回 0 才会调 after_work），而且两条路都会 `delete // j`，所以不存在还两次的可能。 - if (j->self->work_limit) j->self->work_limit->release(); + if (j->slot_held) j->self->work_limit->release(); // **绝对不能在回调里 `delete w`。** // @@ -1216,8 +1273,9 @@ void uvcpp_web_static::serve(uvcpp_web_request& req, uvcpp_web_response& resp, UVCPP_LOG_ERROR(log_category::STATIC) << "投递静态读盘任务失败（" << rc << "）"; delete work; - // 任务从来没进过池子，所以名额要还（与 after_work 回调互斥，见那里的注释）。 - if (im->work_limit) im->work_limit->release(); + // 任务从来没进过池子 —— 而名额现在是在 worker 里才取的（见 `run_job`）， + // 所以这一支**根本没有名额可还**。原先那句 `release()` 在今天会凭空把 + // 别人的名额还掉。 // 删掉 job 同时也就丢掉了 next 的那份副本，于是框架照常收尾、 // 把下面这个 500 发出去。 delete j; diff --git a/tests/functional/web_app_worklimit_func.cpp b/tests/functional/web_app_worklimit_func.cpp index 24a8c4e3..14a5690e 100644 --- a/tests/functional/web_app_worklimit_func.cpp +++ b/tests/functional/web_app_worklimit_func.cpp @@ -19,10 +19,23 @@ * `app.work_limit()->acquire()` 把那个唯一的名额拿走。服务端的静态请求于是 * 必然拿不到名额，503 是**必然**发生的，与线程调度无关。 * - * 代价是：这样测出来的 503 是"名额被占住时静态路径会回绝"，而"静态路径确实 - * 会 acquire"这一点由**同一条断言**保证（如果它根本不做准入，这一条会返回 - * 200 而不是 503）。而"确实会 release"由第 6 组（上限 1 连发 5 次全 200） - * 保证。三条合起来才是完整的：**会拿、会还、拿不到会拒**。 + * 代价是：这样测出来的 503 是"名额被占住时**要读盘的**静态路径会回绝"，而 + * "静态路径确实会 acquire"这一点由**同一条断言**保证（如果它根本不做准入， + * 这一条会返回 200 而不是 503）。而"确实会 release"由第 6 组（上限 1 连发 + * 5 个不同的冷文件全 200）保证。三条合起来才是完整的：**会拿、会还、拿不到 + * 会拒**。 + * + * ## 还有第四条：**命中缓存的那一支不占名额** + * + * 闸门卡的是"并发的磁盘读"，而它过去在 `serve()` 里、投递之前就取名额 —— + * 于是"命中 LRU 缓存、一次磁盘读都不需要"的请求也一起被 503 掉，而那是绝 + * 大多数。`static_integration` 里用**同一个被占住的名额**同时钉住两侧：冷文件 + * 必须 503、热文件必须 200。两条互为对照 —— 只测一边分不出"命中绕开闸门"和 + * "闸门整个失效"这两种截然不同的实现。 + * + * 与它配套的还有第 7 组 `stream_still_gated`：名额**只在"确实要读盘"的两条 + * 支路上取**（整读进内存 / 超过阈值走分片流式），覆盖集合里唯一被移出去的是 + * 缓存命中那一支 —— 流式那一支仍然占，与改动前一致。第六条钉住的就是这一点。 * * ## 为什么不用真慢任务 * @@ -97,6 +110,18 @@ const char* k_root = "uvcpp_worklimit_test_root"; const char* k_file = "uvcpp_worklimit_test_root/hello.txt"; const char* k_text = "WORK-LIMIT-OK"; +/// 一个**从没被请求过**的文件 —— 用它来保证"这次一定要读盘"。 +/// +/// 闸门从 `serve()` 搬到 worker 之后，"要不要占名额"分成两条路：命中 LRU +/// 缓存的一次磁盘读都不做（不该占），未命中要整读进内存（该占）。要辨出 +/// 这两条，用例就必须能**指定**是哪一条 —— 冷文件与热文件各一个，最直接。 +const char* k_cold_file = "uvcpp_worklimit_test_root/cold.txt"; +/// 还名额那一组专用的一批冷文件（每发一个都必须是未命中）。 +char k_seq_files[5][64] = { + "uvcpp_worklimit_test_root/seq0.txt", "uvcpp_worklimit_test_root/seq1.txt", + "uvcpp_worklimit_test_root/seq2.txt", "uvcpp_worklimit_test_root/seq3.txt", + "uvcpp_worklimit_test_root/seq4.txt"}; + #ifdef _WIN32 #include #define TEST_MKDIR(p) _mkdir(p) @@ -116,6 +141,8 @@ void write_file(const std::string& path, const std::string& text) { void make_root() { TEST_MKDIR(k_root); write_file(k_file, k_text); + write_file(k_cold_file, k_text); + for (int i = 0; i < 5; ++i) write_file(k_seq_files[i], k_text); } void cleanup_root() { @@ -322,13 +349,17 @@ void test_static_integration() { // // 这里是**直接占位**（不是投一个慢任务）：上限是 1，测试把这个名额拿走， // 服务端就必然拿不到。503 是确定的，不依赖任何调度时机。 + // + // 请求的是 `cold.txt` —— 一个从没被请求过的文件，所以这次**必然要读盘**。 + // 这一点不能省：闸门现在只卡"确实要读盘"那一支，换成已缓存的 URL，这条 + // 断言测的就不再是"闸门会不会拒"，而是"缓存会不会命中"。 check(app.work_limit()->acquire(), "饱和：测试应当能占住唯一的名额"); check_eq_i(static_cast(app.work_limit()->in_flight()), 1, "饱和：在途数应当是 1"); uvcpp_http_response r503; - check(get(port, "/s/hello.txt", r503), "饱和：请求应当拿到响应"); - check_eq_i(status_of(r503), 503, "饱和：名额满时应当是 503"); + check(get(port, "/s/cold.txt", r503), "饱和：请求应当拿到响应"); + check_eq_i(status_of(r503), 503, "饱和：名额满且要读盘时应当是 503"); // `Retry-After` 是 503 的配套语义（RFC 9110 §10.2.3）。**它必须存在** —— // 没有它，客户端只能靠猜，最常见的反应是立刻重试，正好把已经饱和的池子 @@ -343,17 +374,36 @@ void test_static_integration() { check(body_of(r503).find(k_text) == std::string::npos, "饱和：503 的 body 里不该出现文件内容"); + // ---- 同一时刻、同一个饱和的闸门：**缓存命中必须照常 200** ---- + // + // 这一条就是本次改动的判据，而且它与上面那条 503 **互为对照** —— 同一个 + // 服务、同一个上限、同一个被占住的名额，差别只在"这个 URL 有没有进过 + // 缓存"。改之前两条都是 503（闸门在 `serve()` 里投递前就抢名额，而缓存 + // 查找在 worker 里、根本轮不到）。只测其中一条分不出"命中绕开闸门"和 + // "闸门整个失效"这两种截然不同的实现。 + uvcpp_http_response rhit; + check(get(port, "/s/hello.txt", rhit), "饱和-命中：请求应当拿到响应"); + check_eq_i(status_of(rhit), 200, + "饱和-命中：命中缓存不读盘、不占名额，必须是 200"); + check(body_of(rhit) == k_text, "饱和-命中：内容应当正常"); + check_eq_i(static_cast(st->rejected_count()), 1, + "饱和-命中：命中不该被回绝（回绝计数仍为 1）"); + // ---- 还回名额 → 立刻恢复 ---- app.work_limit()->release(); check_eq_i(static_cast(app.work_limit()->in_flight()), 0, "恢复：还回之后在途数应当是 0"); uvcpp_http_response r2; - check(get(port, "/s/hello.txt", r2), "恢复：还回名额后请求应当成功"); + check(get(port, "/s/cold.txt", r2), "恢复：还回名额后请求应当成功"); check_eq_i(status_of(r2), 200, "恢复：还回名额之后应当回到 200"); check(body_of(r2) == k_text, "恢复：内容应当正常"); - // ---- 名额必须**每次都还**：上限 1 连发 5 次 ---- + // ---- 名额必须**每次都还**：上限 1 连发 5 个不同的冷文件 ---- + // + // 五个文件必须**互不相同、且都还没进过缓存**：连发同一个的话，第 1 次 + // 读过之后后面全走缓存命中那一支，`release` 一次都不需要 —— 于是这条 + // 用例会在"漏还名额"的实现上照样全绿，判据就没了。 // // **这是"会 release"唯一的判据。** 只 acquire 不 release 的实现，第一次 // 请求完全正常，从第二次起全部 503 —— 只跑一次的话根本看不出来。上限压到 @@ -361,7 +411,10 @@ void test_static_integration() { int ok200 = 0; for (int i = 0; i < 5; ++i) { uvcpp_http_response rr; - if (get(port, "/s/hello.txt", rr) && status_of(rr) == 200 && + if (get(port, std::string("/s/") + (k_seq_files[i] + + std::strlen(k_root) + 1), + rr) && + status_of(rr) == 200 && body_of(rr) == k_text) { ++ok200; } @@ -643,6 +696,64 @@ void test_wakeup_reentrant_registration_not_lost() { } // namespace +// ========================================================================= +// 7. 流式（大文件）那一支**仍然**占名额 +// ========================================================================= +// +// 这是与 6 组配套的另一半。名额现在只在"确实要读盘"的两条支路上取：整读进 +// 内存的那一支，和超过阈值改走分片流式的那一支。前者由 6 组钉住，后者由这一 +// 组钉住 —— 两条都占，唯独"命中缓存、一次磁盘读都不做"不占。 +// +// **为什么流式也要占**：它不整读，但确实在读盘，而闸门从一开始（`serve()` +// 时代）就覆盖着它。这次只是把判定点从"投递之前"挪到"读盘之前"，覆盖集合 +// 里唯一被移出去的成员是缓存命中那一支 —— 其余与改动前**逐一对齐**。要是 +// 顺手把流式也放出去，那就不只是修 bug，而是放宽了闸门的语义，得单独讨论。 +// +// 怎么让一个小文件走流式：`max_cached_file_size = 0` 就是"一律走流式" +// （见该选项的说明），所以这里不需要真造一个大文件。 +void test_stream_path_still_gated() { + make_root(); + + uvcpp_web_app app; + configure_for_test(app); + app.set_work_limit(1); + + uvcpp_web_static_options o; + o.max_cached_file_size = 0; // 0 = 一律走分片流式 + std::shared_ptr st2 = app.serve_static("/raw", k_root, o); + check(st2->work_limit().get() == app.work_limit().get(), + "流式：接线，静态服务装上的应当是 App 那个闸门"); + + check(app.start_background() == 0, "流式：服务启动"); + const int port = app.bound_port(); + + uvcpp_http_response r; + check(get(port, "/raw/hello.txt", r), "流式-对照：请求应当成功"); + check_eq_i(status_of(r), 200, "流式-对照：名额空着应当是 200"); + check(body_of(r) == k_text, "流式-对照：内容应当正常"); + + // 占住唯一的名额 → 走流式的请求**也**必须被回绝。 + check(app.work_limit()->acquire(), "流式：占住唯一的名额"); + check_eq_i(static_cast(app.work_limit()->in_flight()), 1, + "流式：在途数应当是 1"); + + uvcpp_http_response rs; + check(get(port, "/raw/hello.txt", rs), "流式-饱和：请求应当拿到响应"); + check_eq_i(status_of(rs), 503, "流式-饱和：名额满时流式路径也应当是 503"); + check(!header_of(rs, "retry-after").empty(), + "流式-饱和：503 必须带 Retry-After"); + + app.work_limit()->release(); + uvcpp_http_response r2; + check(get(port, "/raw/hello.txt", r2), "流式-恢复：还回名额后请求应当成功"); + check_eq_i(status_of(r2), 200, "流式-恢复：还回名额之后应当回到 200"); + check(body_of(r2) == k_text, "流式-恢复：内容应当正常"); + + app.stop(); + app.join(); + cleanup_root(); +} + int main(int argc, char** argv) { std::cout << std::unitbuf; const std::string only = (argc > 1) ? argv[1] : std::string(); @@ -656,6 +767,7 @@ int main(int argc, char** argv) { {"unlimited", test_unlimited}, {"default_limit", test_default_limit}, {"static_integration", test_static_integration}, + {"stream_still_gated", test_stream_path_still_gated}, {"does_not_queue", test_does_not_queue}, {"wakeup_fires_on_release", test_wakeup_fires_on_release}, {"wakeup_one_shot", test_wakeup_one_shot}, diff --git a/tests/tools/stream_mutation.py b/tests/tools/stream_mutation.py index 5c23a5aa..2a00ef28 100644 --- a/tests/tools/stream_mutation.py +++ b/tests/tools/stream_mutation.py @@ -108,15 +108,31 @@ " /* MUTATION: 上限永不拒绝 */", "accounting / does_not_queue"), + # 名额的取/还在 `serve()` 里时是 S12/S13 那两条；搬进 worker 之后锚点跟着 + # 换成了下面这三条（S12/S13 保留语义，S14 是这次搬动的判别性变异）。 ("S12 static_no_admission", STATIC, - " if (im->work_limit && !im->work_limit->acquire()) {", + " if (!work_limit->acquire()) {", " if (false) { /* MUTATION: 静态路径不做准入 */", "static_integration(503)"), ("S13 slot_never_released", STATIC, - " if (j->self->work_limit) j->self->work_limit->release();", + " if (j->slot_held) j->self->work_limit->release();", " /* MUTATION: 名额只拿不还 */", - "static_integration(连发 5 次)"), + "static_integration(连发 5 个冷文件)"), + + # 这次改动的判据：命中缓存的那一支本来不该占名额。把它改回"命中也要占" + # 就等于还原了改动前的行为 —— 饱和时热文件会跟着冷文件一起被 503 掉。 + ("S14 cache_hit_takes_slot", STATIC, + " j->data = p.data;\n j->served_from_cache = true;\n return;", + " if (!acquire_slot(j)) return; /* MUTATION: 命中也要占名额 */\n" + " j->data = p.data;\n j->served_from_cache = true;\n return;", + "static_integration(饱和-命中)"), + + # 流式那一支**仍然**占名额（覆盖集合与改动前对齐）。放出去就红了。 + ("S15 stream_not_gated", STATIC, + " if (!acquire_slot(j)) return;\n j->status = probe_status::STREAM;", + " /* MUTATION: 流式不占名额 */\n j->status = probe_status::STREAM;", + "stream_still_gated"), ]