diff --git a/src/ngx_http_lua_socket_tcp.c b/src/ngx_http_lua_socket_tcp.c index 13655f100b..f73f912f3d 100644 --- a/src/ngx_http_lua_socket_tcp.c +++ b/src/ngx_http_lua_socket_tcp.c @@ -561,6 +561,7 @@ ngx_http_lua_socket_tcp_create_socket_pool(lua_State *L, ngx_http_request_t *r, ngx_queue_init(&sp->cache_connect_op); ngx_queue_init(&sp->wait_connect_op); + ngx_queue_init(&sp->resume_connect_op); ngx_queue_init(&sp->cache); ngx_queue_init(&sp->free); @@ -4424,6 +4425,10 @@ ngx_http_lua_socket_tcp_resume_conn_op(ngx_http_lua_socket_pool_t *spool) ngx_del_timer(&conn_op_ctx->event); } + /* keep posted operations out of the wait queue and track their cleanup */ + ngx_queue_remove(q); + ngx_queue_insert_tail(&spool->resume_connect_op, q); + conn_op_ctx->event.handler = ngx_http_lua_socket_tcp_conn_op_resume_handler; @@ -4452,7 +4457,6 @@ ngx_http_lua_socket_tcp_conn_op_ctx_cleanup(void *data) static void ngx_http_lua_socket_tcp_conn_op_resume_handler(ngx_event_t *ev) { - ngx_queue_t *q; ngx_connection_t *c; ngx_http_lua_ctx_t *ctx; ngx_http_request_t *r; @@ -4467,26 +4471,7 @@ ngx_http_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_http_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, - "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; - } - - q = ngx_queue_head(&spool->wait_connect_op); - ngx_queue_remove(q); + ngx_queue_remove(&conn_op_ctx->queue); coctx = u->write_co_ctx; coctx->cleanup = NULL; @@ -6111,6 +6096,11 @@ ngx_http_lua_socket_shutdown_pool_helper(ngx_http_lua_socket_pool_t *spool) ngx_http_lua_socket_tcp_free_conn_op_ctx(conn_op_ctx); } + if (!ngx_queue_empty(&spool->resume_connect_op)) { + ngx_queue_add(&spool->wait_connect_op, &spool->resume_connect_op); + ngx_queue_init(&spool->resume_connect_op); + } + while (!ngx_queue_empty(&spool->wait_connect_op)) { q = ngx_queue_head(&spool->wait_connect_op); ngx_queue_remove(q); @@ -6121,6 +6111,14 @@ ngx_http_lua_socket_shutdown_pool_helper(ngx_http_lua_socket_pool_t *spool) ngx_del_timer(&conn_op_ctx->event); } +#if (nginx_version >= 1007005) + if (conn_op_ctx->event.posted) { +#else + if (conn_op_ctx->event.prev) { +#endif + ngx_delete_posted_event((&conn_op_ctx->event)); + } + ngx_http_lua_socket_tcp_free_conn_op_ctx(conn_op_ctx); } diff --git a/src/ngx_http_lua_socket_tcp.h b/src/ngx_http_lua_socket_tcp.h index f47091ff05..4491f8ce03 100644 --- a/src/ngx_http_lua_socket_tcp.h +++ b/src/ngx_http_lua_socket_tcp.h @@ -67,6 +67,7 @@ typedef struct { ngx_int_t size; ngx_queue_t cache_connect_op; ngx_queue_t wait_connect_op; + ngx_queue_t resume_connect_op; /* connections == active connections + pending connect operations, * while active connections == out-of-pool reused connections diff --git a/t/068-socket-keepalive.t b/t/068-socket-keepalive.t index a6c8c42f4e..f658560932 100644 --- a/t/068-socket-keepalive.t +++ b/t/068-socket-keepalive.t @@ -3196,3 +3196,176 @@ qr/\Qbad argument #2 to 'setkeepalive' (number expected, got string)\E/ --- timeout: 4 --- curl_error eval qr{HTTP/3 stream 0 reset by server} + + + +=== TEST 56: conn queuing: slots freed in the same cycle each resume a queued connect +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + + content_by_lua_block { + local port = ngx.var.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) + } + } +--- request +GET /t +--- response_body +w1: connected +w2: connected +--- error_log +lua tcp socket queue connect operation for connection pool "same_cycle" +--- no_error_log +[error] +--- timeout: 4 + + + +=== TEST 57: conn queuing: close multiple connections before resumptions run +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + + content_by_lua_block { + local semaphore = require "ngx.semaphore" + local connected = semaphore.new() + local release = semaphore.new() + local port = ngx.var.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) + } + } +--- request +GET /t +--- response_body +w1: reused 0 +w2: reused 0 +--- error_log +lua tcp socket queue connect operation for connection pool "same_cycle_close" +--- no_error_log +[error] +--- timeout: 4 + + + +=== TEST 58: conn queuing: cancel a posted resumption and resume the next waiter +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + + content_by_lua_block { + local semaphore = require "ngx.semaphore" + local connected = semaphore.new() + local release = semaphore.new() + local port = ngx.var.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, ", ")) + } + } +--- request +GET /t +--- response_body +w2: reused 1 +w3: reused 1 +order: w2, w3 +--- error_log +lua tcp socket abort queueing +--- no_error_log +[error] +--- timeout: 4