From 27d1a974082232825f0451603161bea2f7d0c7f8 Mon Sep 17 00:00:00 2001 From: perfgao Date: Fri, 30 Oct 2026 22:07:17 +0800 Subject: [PATCH 1/2] bugfix: queued connects were not resumed when several pool slots were freed in the same cycle. When a connection pool with a backlog frees a slot, resume_conn_op() posts the event of the operation at the head of wait_connect_op. That operation only leaves the queue when its event runs, so when two slots are freed in the same cycle, both calls post the same event and ngx_post_event() ignores the second one. The second slot is handed to no waiter, but the pool still counts it as busy, so new connects queue behind it too. Each lost resumption shrinks the pool until the queue drains, and under a burst, queued connects time out while slots are free. Resume the first queued operation whose event is not posted yet, and let the resume handler remove its own operation instead of the queue head. As several operations can now be posted at once, also delete a posted event before freeing its operation when the pool is destroyed. --- src/ngx_stream_lua_socket_tcp.c | 40 +++++++++++++++++++++----- t/068-socket-keepalive.t | 50 +++++++++++++++++++++++++++++++++ 2 files changed, 83 insertions(+), 7 deletions(-) diff --git a/src/ngx_stream_lua_socket_tcp.c b/src/ngx_stream_lua_socket_tcp.c index 561be338..d9a3934d 100644 --- a/src/ngx_stream_lua_socket_tcp.c +++ b/src/ngx_stream_lua_socket_tcp.c @@ -4946,10 +4946,30 @@ ngx_stream_lua_socket_tcp_resume_conn_op( return; } - q = ngx_queue_head(&spool->wait_connect_op); - conn_op_ctx = ngx_queue_data(q, - ngx_stream_lua_socket_tcp_conn_op_ctx_t, - queue); + /* each freed slot must resume its own queued connect operation. + * A resumed operation stays in wait_connect_op until its posted event + * runs, so when several slots are freed in the same cycle, picking the + * head every time re-posts the same event, which ngx_post_event() + * ignores, and the extra slots are lost while the queue still counts + * them as busy. Resume the first operation not posted yet instead. + */ + for (q = ngx_queue_head(&spool->wait_connect_op); + q != ngx_queue_sentinel(&spool->wait_connect_op); + q = ngx_queue_next(q)) + { + conn_op_ctx = ngx_queue_data(q, + ngx_stream_lua_socket_tcp_conn_op_ctx_t, + queue); + + if (!conn_op_ctx->event.posted) { + break; + } + } + + if (q == ngx_queue_sentinel(&spool->wait_connect_op)) { + return; + } + ngx_log_debug4(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, "stream lua tcp socket post connect operation " "resumption u: %p, ctx: %p for connection pool \"%s\", " @@ -4988,7 +5008,6 @@ ngx_stream_lua_socket_tcp_conn_op_ctx_cleanup(void *data) static void ngx_stream_lua_socket_tcp_conn_op_resume_handler(ngx_event_t *ev) { - ngx_queue_t *q; ngx_stream_lua_request_t *r; ngx_stream_lua_cleanup_t *cln; ngx_stream_lua_ctx_t *ctx; @@ -5021,8 +5040,9 @@ ngx_stream_lua_socket_tcp_conn_op_resume_handler(ngx_event_t *ev) return; } - q = ngx_queue_head(&spool->wait_connect_op); - ngx_queue_remove(q); + /* several operations can be posted at once, so remove this one + * rather than the queue head */ + ngx_queue_remove(&conn_op_ctx->queue); coctx = u->write_co_ctx; coctx->cleanup = NULL; @@ -6552,6 +6572,12 @@ ngx_stream_lua_socket_shutdown_pool_helper( ngx_del_timer(&conn_op_ctx->event); } + /* more than one queued operation can be posted for + * resumption now, do not free one with its event still posted */ + if (conn_op_ctx->event.posted) { + ngx_delete_posted_event(&conn_op_ctx->event); + } + ngx_stream_lua_socket_tcp_free_conn_op_ctx(conn_op_ctx); } diff --git a/t/068-socket-keepalive.t b/t/068-socket-keepalive.t index 829be217..48e72f33 100644 --- a/t/068-socket-keepalive.t +++ b/t/068-socket-keepalive.t @@ -2492,3 +2492,53 @@ qr/\Qbad argument #2 to 'setkeepalive' (number expected, got string)\E/ --- no_error_log [crit] --- timeout: 4 + + + +=== TEST 55: conn queuing: slots freed in the same cycle each resume a queued connect +--- stream_server_config + content_by_lua_block { + local port = $TEST_NGINX_MEMCACHED_PORT + local opts = {pool = "same_cycle", pool_size = 2, backlog = 2} + + local sock1 = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local sock2 = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local function waiter(name) + return ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + sock:settimeouts(500, 500, 500) + + local ok, err = sock:connect("127.0.0.1", port, opts) + if not ok then + return name .. ": " .. err + end + + -- keep the slot past the other waiter's connect timeout, + -- so only a resumption can let it connect + ngx.sleep(1) + sock:close() + return name .. ": connected" + end) + end + + local w1 = waiter("w1") + local w2 = waiter("w2") + + -- release both connections in the same cycle + assert(sock1:setkeepalive()) + assert(sock2:setkeepalive()) + + local _, res1 = ngx.thread.wait(w1) + local _, res2 = ngx.thread.wait(w2) + ngx.say(res1) + ngx.say(res2) + } +--- stream_response +w1: connected +w2: connected +--- error_log +stream lua tcp socket queue connect operation for connection pool "same_cycle" +--- no_error_log +[error] +--- timeout: 4 From 4ad8222cfdd86e4107d5eef9fb95202b63bb5ea1 Mon Sep 17 00:00:00 2001 From: lijunlong Date: Mon, 2 Nov 2026 22:05:59 +0800 Subject: [PATCH 2/2] optimize: resume queued connects in constant time. Replace the scan over posted connect events with a cursor pointing to the next unposted operation. Advance it when scheduling a waiter and when its waiter times out or is cancelled, and reset it on pool shutdown. Keep operations on the existing wait queue until resumed. Place the cursor after the pool's hot fields so their offsets stay unchanged, adding eight bytes per pool on 64-bit builds. Add regressions for closing several connections in one cycle, cancelling a posted operation, timing out the next waiter, and appending waiters after all existing operations have been posted. Keepalive, preread, timeout, and HUP checks pass 454 assertions. An isolated ASan probe verifies shutdown cleanup. Short fixed-CPU load tests show no throughput regression against the previous implementation. Template source: openresty/meta-lua-nginx-module@6b1be53. --- src/ngx_stream_lua_socket_tcp.c | 74 ++++-------- src/ngx_stream_lua_socket_tcp.h | 2 + t/068-socket-keepalive.t | 204 ++++++++++++++++++++++++++++++++ 3 files changed, 232 insertions(+), 48 deletions(-) diff --git a/src/ngx_stream_lua_socket_tcp.c b/src/ngx_stream_lua_socket_tcp.c index d9a3934d..32a0fd8e 100644 --- a/src/ngx_stream_lua_socket_tcp.c +++ b/src/ngx_stream_lua_socket_tcp.c @@ -607,6 +607,7 @@ ngx_stream_lua_socket_tcp_create_socket_pool(lua_State *L, ngx_queue_init(&sp->cache_connect_op); ngx_queue_init(&sp->wait_connect_op); + sp->next_connect_op = ngx_queue_sentinel(&sp->wait_connect_op); ngx_queue_init(&sp->cache); ngx_queue_init(&sp->free); @@ -749,6 +750,12 @@ ngx_stream_lua_socket_tcp_connect_helper(lua_State *L, ngx_queue_insert_tail(&spool->wait_connect_op, &conn_op_ctx->queue); + if (spool->next_connect_op + == ngx_queue_sentinel(&spool->wait_connect_op)) + { + spool->next_connect_op = &conn_op_ctx->queue; + } + ngx_log_debug3(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, "stream lua tcp socket queued connect " "operation for %d(ms), u: %p, ctx: %p", @@ -4854,9 +4861,14 @@ ngx_stream_lua_socket_tcp_conn_op_timeout_handler(ngx_event_t *ev) ngx_stream_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx; conn_op_ctx = ev->data; + u = conn_op_ctx->u; + + if (u->socket_pool->next_connect_op == &conn_op_ctx->queue) { + u->socket_pool->next_connect_op = ngx_queue_next(&conn_op_ctx->queue); + } + ngx_queue_remove(&conn_op_ctx->queue); - u = conn_op_ctx->u; r = u->request; coctx = u->write_co_ctx; @@ -4942,34 +4954,17 @@ ngx_stream_lua_socket_tcp_resume_conn_op( * operation resumption, so that there is no resumption happens when Nginx * is exiting. */ - if (ngx_queue_empty(&spool->wait_connect_op)) { - return; - } - - /* each freed slot must resume its own queued connect operation. - * A resumed operation stays in wait_connect_op until its posted event - * runs, so when several slots are freed in the same cycle, picking the - * head every time re-posts the same event, which ngx_post_event() - * ignores, and the extra slots are lost while the queue still counts - * them as busy. Resume the first operation not posted yet instead. - */ - for (q = ngx_queue_head(&spool->wait_connect_op); - q != ngx_queue_sentinel(&spool->wait_connect_op); - q = ngx_queue_next(q)) - { - conn_op_ctx = ngx_queue_data(q, - ngx_stream_lua_socket_tcp_conn_op_ctx_t, - queue); - - if (!conn_op_ctx->event.posted) { - break; - } - } - + q = spool->next_connect_op; if (q == ngx_queue_sentinel(&spool->wait_connect_op)) { return; } + /* advance before posting so each slot resumes a distinct waiter */ + spool->next_connect_op = ngx_queue_next(q); + + conn_op_ctx = ngx_queue_data(q, + ngx_stream_lua_socket_tcp_conn_op_ctx_t, + queue); ngx_log_debug4(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, "stream lua tcp socket post connect operation " "resumption u: %p, ctx: %p for connection pool \"%s\", " @@ -5021,27 +5016,6 @@ ngx_stream_lua_socket_tcp_conn_op_resume_handler(ngx_event_t *ev) r = u->request; spool = u->socket_pool; - if (ngx_queue_empty(&spool->wait_connect_op)) { -#if (NGX_DEBUG) - ngx_stream_lua_assert(!(spool->backlog >= 0 - && spool->connections > spool->size)); - -#else - if (spool->backlog >= 0 && spool->connections > spool->size) { - ngx_log_error(NGX_LOG_ERR, ngx_cycle->log, 0, - "stream lua tcp socket connections count " - "mismatched for connection pool \"%s\", connections: " - "%i, size: %i", - spool->key, spool->connections, spool->size); - spool->connections = spool->size; - } -#endif - - return; - } - - /* several operations can be posted at once, so remove this one - * rather than the queue head */ ngx_queue_remove(&conn_op_ctx->queue); coctx = u->write_co_ctx; @@ -6542,6 +6516,8 @@ ngx_stream_lua_socket_shutdown_pool_helper( ngx_stream_lua_socket_pool_item_t *item; ngx_stream_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx; + spool->next_connect_op = ngx_queue_sentinel(&spool->wait_connect_op); + while (!ngx_queue_empty(&spool->cache)) { q = ngx_queue_head(&spool->cache); @@ -6572,8 +6548,6 @@ ngx_stream_lua_socket_shutdown_pool_helper( ngx_del_timer(&conn_op_ctx->event); } - /* more than one queued operation can be posted for - * resumption now, do not free one with its event still posted */ if (conn_op_ctx->event.posted) { ngx_delete_posted_event(&conn_op_ctx->event); } @@ -6989,6 +6963,10 @@ ngx_stream_lua_tcp_queue_conn_op_cleanup(void *data) ngx_del_timer(&conn_op_ctx->event); } + if (u->socket_pool->next_connect_op == &conn_op_ctx->queue) { + u->socket_pool->next_connect_op = ngx_queue_next(&conn_op_ctx->queue); + } + ngx_queue_remove(&conn_op_ctx->queue); ngx_queue_insert_head(&u->socket_pool->cache_connect_op, &conn_op_ctx->queue); diff --git a/src/ngx_stream_lua_socket_tcp.h b/src/ngx_stream_lua_socket_tcp.h index d5c4c76d..f44bf9bd 100644 --- a/src/ngx_stream_lua_socket_tcp.h +++ b/src/ngx_stream_lua_socket_tcp.h @@ -87,6 +87,8 @@ typedef struct { ngx_int_t backlog; + ngx_queue_t *next_connect_op; + u_char key[1]; } ngx_stream_lua_socket_pool_t; diff --git a/t/068-socket-keepalive.t b/t/068-socket-keepalive.t index 48e72f33..43da5d71 100644 --- a/t/068-socket-keepalive.t +++ b/t/068-socket-keepalive.t @@ -2542,3 +2542,207 @@ stream lua tcp socket queue connect operation for connection pool "same_cycle" --- no_error_log [error] --- timeout: 4 + + + +=== TEST 56: conn queuing: close multiple connections before resumptions run +--- stream_server_config + content_by_lua_block { + local semaphore = require "ngx.semaphore" + local connected = semaphore.new() + local release = semaphore.new() + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "same_cycle_close", pool_size = 2, backlog = 2} + + local sock1 = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local sock2 = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local function waiter(name) + local sock = ngx.socket.tcp() + sock:settimeout(500) + assert(sock:connect("127.0.0.1", port, opts)) + local reused = sock:getreusedtimes() + connected:post() + assert(release:wait(2)) + sock:close() + return name .. ": reused " .. reused + end + + local w1 = ngx.thread.spawn(waiter, "w1") + local w2 = ngx.thread.spawn(waiter, "w2") + + assert(sock1:close()) + assert(sock2:close()) + + -- neither waiter may free its slot before both have connected + assert(connected:wait(2)) + assert(connected:wait(2)) + release:post(2) + + local _, res1 = ngx.thread.wait(w1) + local _, res2 = ngx.thread.wait(w2) + ngx.say(res1) + ngx.say(res2) + } +--- stream_response +w1: reused 0 +w2: reused 0 +--- error_log +stream lua tcp socket queue connect operation for connection pool "same_cycle_close" +--- no_error_log +[error] +--- timeout: 4 + + + +=== TEST 57: conn queuing: cancel a posted resumption and resume the next waiter +--- stream_server_config + content_by_lua_block { + local semaphore = require "ngx.semaphore" + local connected = semaphore.new() + local release = semaphore.new() + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "cancel_posted", pool_size = 2, backlog = 3} + local order = {} + + local sock1 = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local sock2 = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local function waiter(name) + local sock = ngx.socket.tcp() + sock:settimeout(500) + assert(sock:connect("127.0.0.1", port, opts)) + order[#order + 1] = name + local reused = sock:getreusedtimes() + connected:post() + assert(release:wait(2)) + sock:close() + return name .. ": reused " .. reused + end + + local w1 = ngx.thread.spawn(waiter, "w1") + local w2 = ngx.thread.spawn(waiter, "w2") + local w3 = ngx.thread.spawn(waiter, "w3") + + assert(sock1:setkeepalive()) + assert(sock2:setkeepalive()) + assert(ngx.thread.kill(w1)) + + assert(connected:wait(2)) + assert(connected:wait(2)) + release:post(2) + + local _, res2 = ngx.thread.wait(w2) + local _, res3 = ngx.thread.wait(w3) + ngx.say(res2) + ngx.say(res3) + ngx.say("order: ", table.concat(order, ", ")) + } +--- stream_response +w2: reused 1 +w3: reused 1 +order: w2, w3 +--- error_log +stream lua tcp socket abort queueing +--- no_error_log +[error] +--- timeout: 4 + + + +=== TEST 58: conn queuing: advance past a timed out waiter +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local semaphore = require "ngx.semaphore" + local expired = semaphore.new() + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "expired_head", pool_size = 1, backlog = 2} + local sock = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local function waiter1() + local s = ngx.socket.tcp() + s:settimeout(10) + local ok, err = s:connect("127.0.0.1", port, opts) + expired:post() + return "w1: " .. (err or "connected") + end + + local function waiter2() + local s = ngx.socket.tcp() + s:settimeout(500) + assert(s:connect("127.0.0.1", port, opts)) + local reused = s:getreusedtimes() + s:close() + return "w2: reused " .. reused + end + + local w1 = ngx.thread.spawn(waiter1) + local w2 = ngx.thread.spawn(waiter2) + assert(expired:wait(2)) + assert(sock:setkeepalive()) + + local _, res1 = ngx.thread.wait(w1) + local _, res2 = ngx.thread.wait(w2) + ngx.say(res1) + ngx.say(res2) + } +--- stream_response +w1: timeout +w2: reused 1 +--- error_log +stream lua tcp socket queue connect operation for connection pool "expired_head" +--- no_error_log +[error] +--- timeout: 4 + + + +=== TEST 59: conn queuing: append after all existing waiters have been posted +--- stream_server_config + content_by_lua_block { + local semaphore = require "ngx.semaphore" + local connected = semaphore.new() + local release = semaphore.new() + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "append_posted", pool_size = 1, backlog = 2} + local sock = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local function waiter1() + local s = ngx.socket.tcp() + s:settimeout(500) + assert(s:connect("127.0.0.1", port, opts)) + connected:post() + assert(release:wait(2)) + s:close() + return "w1: connected" + end + + local function waiter2() + local s = ngx.socket.tcp() + s:settimeout(500) + assert(s:connect("127.0.0.1", port, opts)) + s:close() + return "w2: connected" + end + + local w1 = ngx.thread.spawn(waiter1) + assert(sock:close()) + local w2 = ngx.thread.spawn(waiter2) + + assert(connected:wait(2)) + release:post() + local _, res1 = ngx.thread.wait(w1) + local _, res2 = ngx.thread.wait(w2) + ngx.say(res1) + ngx.say(res2) + } +--- stream_response +w1: connected +w2: connected +--- error_log +stream lua tcp socket queue connect operation for connection pool "append_posted" +--- no_error_log +[error] +--- timeout: 4