diff --git a/src/ngx_stream_lua_socket_tcp.c b/src/ngx_stream_lua_socket_tcp.c index 561be338..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,11 +4954,14 @@ 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)) { + q = spool->next_connect_op; + if (q == ngx_queue_sentinel(&spool->wait_connect_op)) { return; } - q = ngx_queue_head(&spool->wait_connect_op); + /* 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); @@ -4988,7 +5003,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; @@ -5002,27 +5016,7 @@ 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; - } - - 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; @@ -6522,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); @@ -6552,6 +6548,10 @@ ngx_stream_lua_socket_shutdown_pool_helper( ngx_del_timer(&conn_op_ctx->event); } + 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); } @@ -6963,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 829be217..43da5d71 100644 --- a/t/068-socket-keepalive.t +++ b/t/068-socket-keepalive.t @@ -2492,3 +2492,257 @@ 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 + + + +=== 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