From c2444b1011379eed0e676a634b37b0e29db6e5c2 Mon Sep 17 00:00:00 2001 From: AxisNow-Dev Date: Wed, 4 Nov 2026 13:19:44 +0800 Subject: [PATCH 1/2] bugfix: preserve cosocket pool slots across queued connects. Hand idle connections and free pool slots directly to queued connect operations. Track waiting and posted operations separately so new connects cannot take a connection assigned to a waiter or exceed the configured pool size. Reuse cached peers before opening a connection when a resumed operation owns an empty slot. Release or transfer each slot once on cancellation, timeout, connection failure, session cleanup, and pool shutdown. Keep queued sockets busy until their connect completes and preserve their pool on rejected concurrent operations. Add regression and stress coverage for handoff, spare capacity, failed connects, coroutine cleanup, and shutdown in preread and content phases. --- src/ngx_stream_lua_socket_tcp.c | 659 ++++++++------ src/ngx_stream_lua_socket_tcp.h | 23 +- t/068-socket-keepalive.t | 1486 ++++++++++++++++++++++++++++++- t/196-socket-keepalive-queue.t | 381 ++++++++ 4 files changed, 2275 insertions(+), 274 deletions(-) create mode 100644 t/196-socket-keepalive-queue.t diff --git a/src/ngx_stream_lua_socket_tcp.c b/src/ngx_stream_lua_socket_tcp.c index 32a0fd8e..303e472f 100644 --- a/src/ngx_stream_lua_socket_tcp.c +++ b/src/ngx_stream_lua_socket_tcp.c @@ -123,6 +123,9 @@ static void ngx_stream_lua_socket_tcp_create_socket_pool(lua_State *L, ngx_int_t backlog, ngx_stream_lua_socket_pool_t **spool); static ngx_int_t ngx_stream_lua_get_keepalive_peer(ngx_stream_lua_request_t *r, ngx_stream_lua_socket_tcp_upstream_t *u); +static ngx_int_t ngx_stream_lua_socket_tcp_reuse_peer( + ngx_stream_lua_request_t *r, ngx_stream_lua_socket_tcp_upstream_t *u, + ngx_stream_lua_socket_pool_item_t *item); static void ngx_stream_lua_socket_keepalive_dummy_handler(ngx_event_t *ev); static int ngx_stream_lua_socket_tcp_connect_helper(lua_State *L, ngx_stream_lua_socket_tcp_upstream_t *u, ngx_stream_lua_request_t *r, @@ -133,8 +136,16 @@ static void ngx_stream_lua_socket_tcp_conn_op_timeout_handler( static int ngx_stream_lua_socket_tcp_conn_op_timeout_retval_handler( ngx_stream_lua_request_t *r, ngx_stream_lua_socket_tcp_upstream_t *u, lua_State *L); -static void ngx_stream_lua_socket_tcp_resume_conn_op( - ngx_stream_lua_socket_pool_t *spool); +static void ngx_stream_lua_socket_tcp_release_slot( + ngx_stream_lua_socket_pool_t *spool, + ngx_stream_lua_socket_pool_item_t *item); +static void ngx_stream_lua_socket_tcp_cleanup_failed_connect( + ngx_stream_lua_request_t *r, ngx_stream_lua_socket_tcp_upstream_t *u); +static ngx_int_t ngx_stream_lua_socket_tcp_resume_conn_op( + ngx_stream_lua_socket_pool_t *spool, + ngx_stream_lua_socket_pool_item_t *item); +static void ngx_stream_lua_socket_tcp_cancel_conn_op( + ngx_stream_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx); static void 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); static ngx_int_t ngx_stream_lua_socket_keepalive_close_handler(ngx_event_t *ev); @@ -603,11 +614,12 @@ ngx_stream_lua_socket_tcp_create_socket_pool(lua_State *L, sp->backlog = backlog; sp->size = pool_size; sp->connections = 0; + sp->waiting = 0; sp->lua_vm = ngx_stream_lua_get_lua_vm(r, NULL); 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->posted_connect_op); ngx_queue_init(&sp->cache); ngx_queue_init(&sp->free); @@ -653,7 +665,7 @@ ngx_stream_lua_socket_tcp_connect_helper(lua_State *L, ngx_stream_core_srv_conf_t *clcf; spool = u->socket_pool; - if (spool != NULL) { + if (spool != NULL && !resuming) { rc = ngx_stream_lua_get_keepalive_peer(r, u); if (rc == NGX_OK) { @@ -661,115 +673,121 @@ ngx_stream_lua_socket_tcp_connect_helper(lua_State *L, return 1; } - /* rc == NGX_DECLINED */ + if (rc == NGX_ERROR) { + ngx_stream_lua_socket_tcp_finalize(r, u); + lua_pushnil(L); + lua_pushliteral(L, "no memory"); + return 2; + } - spool->connections++; + /* rc == NGX_DECLINED */ - /* check if backlog is enabled and - * don't queue resuming connection operation */ - if (spool->backlog >= 0 && !resuming) { + if (spool->backlog >= 0 && spool->connections >= spool->size) { dd("lua tcp socket %s connections %ld", spool->key, spool->connections); - if (spool->connections > spool->size + spool->backlog) { - spool->connections--; + if (spool->waiting >= spool->backlog) { lua_pushnil(L); lua_pushliteral(L, "too many waiting connect operations"); return 2; } - if (spool->connections > spool->size) { - ngx_log_debug2(NGX_LOG_DEBUG_STREAM, u->peer.log, 0, - "stream lua tcp socket queue connect " - "operation for connection pool \"%s\", " - "connections: %i", - spool->key, spool->connections); - - host_size = sizeof(u_char) * - (ngx_max(host_len, NGX_INET_ADDRSTRLEN) + 1); - - if (!ngx_queue_empty(&spool->cache_connect_op)) { - q = ngx_queue_last(&spool->cache_connect_op); - ngx_queue_remove(q); - conn_op_ctx = ngx_queue_data( - q, ngx_stream_lua_socket_tcp_conn_op_ctx_t, queue); - - conn_op_host = &conn_op_ctx->host; - if (host_len > conn_op_host->len - && host_len > NGX_INET_ADDRSTRLEN) - { - ngx_free(conn_op_host->data); - conn_op_host->data = ngx_alloc(host_size, - ngx_cycle->log); - if (conn_op_host->data == NULL) { - ngx_free(conn_op_ctx); - goto no_memory_and_not_resuming; - } - } + ngx_log_debug2(NGX_LOG_DEBUG_STREAM, u->peer.log, 0, + "stream lua tcp socket queue connect " + "operation for connection pool \"%s\", " + "connections: %i", + spool->key, spool->connections); - } else { - conn_op_ctx = ngx_alloc( - sizeof(ngx_stream_lua_socket_tcp_conn_op_ctx_t), - ngx_cycle->log); - if (conn_op_ctx == NULL) { - goto no_memory_and_not_resuming; - } + host_size = sizeof(u_char) * + (ngx_max(host_len, NGX_INET_ADDRSTRLEN) + 1); + + if (!ngx_queue_empty(&spool->cache_connect_op)) { + q = ngx_queue_last(&spool->cache_connect_op); + ngx_queue_remove(q); + conn_op_ctx = ngx_queue_data( + q, ngx_stream_lua_socket_tcp_conn_op_ctx_t, queue); - conn_op_host = &conn_op_ctx->host; - conn_op_host->data = ngx_alloc(host_size, ngx_cycle->log); + conn_op_host = &conn_op_ctx->host; + if (host_len > conn_op_host->len + && host_len > NGX_INET_ADDRSTRLEN) + { + ngx_free(conn_op_host->data); + conn_op_host->data = ngx_alloc(host_size, + ngx_cycle->log); if (conn_op_host->data == NULL) { ngx_free(conn_op_ctx); - goto no_memory_and_not_resuming; + return luaL_error(L, "no memory"); } } - conn_op_ctx->cleanup = NULL; + } else { + conn_op_ctx = ngx_alloc( + sizeof(ngx_stream_lua_socket_tcp_conn_op_ctx_t), + ngx_cycle->log); + if (conn_op_ctx == NULL) { + return luaL_error(L, "no memory"); + } - ngx_memcpy(conn_op_host->data, host_ref, host_len); - conn_op_host->data[host_len] = '\0'; - conn_op_host->len = host_len; + conn_op_host = &conn_op_ctx->host; + conn_op_host->data = ngx_alloc(host_size, ngx_cycle->log); + if (conn_op_host->data == NULL) { + ngx_free(conn_op_ctx); + return luaL_error(L, "no memory"); + } + } - conn_op_ctx->port = port; + conn_op_ctx->cleanup = NULL; - u->write_co_ctx = ctx->cur_co_ctx; + ngx_memcpy(conn_op_host->data, host_ref, host_len); + conn_op_host->data[host_len] = '\0'; + conn_op_host->len = host_len; - conn_op_ctx->u = u; - ctx->cur_co_ctx->cleanup = - ngx_stream_lua_tcp_queue_conn_op_cleanup; - ctx->cur_co_ctx->data = conn_op_ctx; + conn_op_ctx->port = port; + conn_op_ctx->item = NULL; + conn_op_ctx->queued = 1; - ngx_memzero(&conn_op_ctx->event, sizeof(ngx_event_t)); - conn_op_ctx->event.handler = - ngx_stream_lua_socket_tcp_conn_op_timeout_handler; - conn_op_ctx->event.data = conn_op_ctx; - conn_op_ctx->event.log = ngx_cycle->log; + u->conn_waiting = 1; + u->write_co_ctx = ctx->cur_co_ctx; - ngx_add_timer(&conn_op_ctx->event, u->connect_timeout); + conn_op_ctx->u = u; + ctx->cur_co_ctx->cleanup = + ngx_stream_lua_tcp_queue_conn_op_cleanup; + ctx->cur_co_ctx->data = conn_op_ctx; - ngx_queue_insert_tail(&spool->wait_connect_op, - &conn_op_ctx->queue); + ngx_memzero(&conn_op_ctx->event, sizeof(ngx_event_t)); + conn_op_ctx->event.handler = + ngx_stream_lua_socket_tcp_conn_op_timeout_handler; + conn_op_ctx->event.data = conn_op_ctx; + conn_op_ctx->event.log = ngx_cycle->log; - if (spool->next_connect_op - == ngx_queue_sentinel(&spool->wait_connect_op)) - { - spool->next_connect_op = &conn_op_ctx->queue; - } + ngx_add_timer(&conn_op_ctx->event, u->connect_timeout); - 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", - u->connect_timeout, conn_op_ctx->u, conn_op_ctx); + ngx_queue_insert_tail(&spool->wait_connect_op, + &conn_op_ctx->queue); - return lua_yield(L, 0); - } + spool->waiting++; + + 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", + u->connect_timeout, conn_op_ctx->u, conn_op_ctx); + + return lua_yield(L, 0); } + spool->connections++; } /* end spool != NULL */ host.data = ngx_palloc(r->pool, host_len + 1); if (host.data == NULL) { - return luaL_error(L, "no memory"); + if (resuming) { + lua_pushnil(L); + lua_pushliteral(L, "no memory"); + goto failed; + } + + goto no_memory_and_not_resuming; } host.len = host_len; @@ -956,19 +974,13 @@ ngx_stream_lua_socket_tcp_connect_helper(lua_State *L, failed: - if (spool != NULL) { - spool->connections--; - ngx_stream_lua_socket_tcp_resume_conn_op(spool); - } + ngx_stream_lua_socket_tcp_cleanup_failed_connect(r, u); return 2; no_memory_and_not_resuming: - if (spool != NULL) { - spool->connections--; - ngx_stream_lua_socket_tcp_resume_conn_op(spool); - } + ngx_stream_lua_socket_tcp_cleanup_failed_connect(r, u); return luaL_error(L, "no memory"); } @@ -1124,10 +1136,6 @@ ngx_stream_lua_socket_tcp_connect(lua_State *L) case LUA_TSTRING: custom_pool = 1; - - lua_pushvalue(L, -1); - lua_rawseti(L, 1, SOCKET_KEY_INDEX); - key_index = n + 1; break; @@ -1173,13 +1181,6 @@ ngx_stream_lua_socket_tcp_connect(lua_State *L) port = 0; } - if (!custom_pool) { - /* the key's index is 2 */ - - lua_pushvalue(L, 2); - lua_rawseti(L, 1, SOCKET_KEY_INDEX); - } - lua_rawgeti(L, 1, SOCKET_CTX_INDEX); u = lua_touserdata(L, -1); lua_pop(L, 1); @@ -1224,6 +1225,10 @@ ngx_stream_lua_socket_tcp_connect(lua_State *L) lua_rawseti(L, 1, SOCKET_CTX_INDEX); } + /* A rejected reconnect must not change the pending operation's pool. */ + lua_pushvalue(L, key_index); + lua_rawseti(L, 1, SOCKET_KEY_INDEX); + ngx_memzero(u, sizeof(ngx_stream_lua_socket_tcp_upstream_t)); u->request = r; /* set the controlling request */ @@ -4863,11 +4868,8 @@ ngx_stream_lua_socket_tcp_conn_op_timeout_handler(ngx_event_t *ev) 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->socket_pool->waiting--; r = u->request; @@ -4876,6 +4878,7 @@ ngx_stream_lua_socket_tcp_conn_op_timeout_handler(ngx_event_t *ev) /* note that we store conn_op_ctx in coctx->data instead of u */ coctx->data = conn_op_ctx; u->write_co_ctx = NULL; + u->conn_waiting = 0; llcf = ngx_stream_lua_get_module_loc_conf(r, ngx_stream_lua_module); @@ -4888,7 +4891,6 @@ ngx_stream_lua_socket_tcp_conn_op_timeout_handler(ngx_event_t *ev) ngx_queue_insert_head(&u->socket_pool->cache_connect_op, &conn_op_ctx->queue); - u->socket_pool->connections--; ctx = ngx_stream_lua_get_module_ctx(r, ngx_stream_lua_module); if (ctx == NULL) { @@ -4929,42 +4931,35 @@ ngx_stream_lua_socket_tcp_conn_op_timeout_retval_handler( } -static void -ngx_stream_lua_socket_tcp_resume_conn_op( - ngx_stream_lua_socket_pool_t *spool) +static ngx_int_t +ngx_stream_lua_socket_tcp_resume_conn_op(ngx_stream_lua_socket_pool_t *spool, + ngx_stream_lua_socket_pool_item_t *item) { ngx_queue_t *q; ngx_stream_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx; -#if (NGX_DEBUG) - ngx_stream_lua_assert(spool->connections >= 0); - -#else - if (spool->connections < 0) { - 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 = 0; - } -#endif + ngx_stream_lua_assert(spool->connections > 0); + ngx_stream_lua_assert(spool->waiting >= 0); + ngx_stream_lua_assert((spool->waiting == 0) + == ngx_queue_empty(&spool->wait_connect_op)); - /* we manually destroy wait_connect_op before triggering connect - * operation resumption, so that there is no resumption happens when Nginx - * is exiting. - */ - q = spool->next_connect_op; - if (q == ngx_queue_sentinel(&spool->wait_connect_op)) { - return; + if (ngx_queue_empty(&spool->wait_connect_op)) { + return NGX_DECLINED; } - /* advance before posting so each slot resumes a distinct waiter */ - spool->next_connect_op = ngx_queue_next(q); + q = ngx_queue_head(&spool->wait_connect_op); + ngx_queue_remove(q); + spool->waiting--; + ngx_queue_insert_tail(&spool->posted_connect_op, q); - conn_op_ctx = ngx_queue_data(q, - ngx_stream_lua_socket_tcp_conn_op_ctx_t, + conn_op_ctx = ngx_queue_data(q, ngx_stream_lua_socket_tcp_conn_op_ctx_t, queue); + conn_op_ctx->queued = 0; + conn_op_ctx->item = item; + if (item != NULL) { + item->owner = conn_op_ctx; + } + 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\", " @@ -4979,24 +4974,99 @@ ngx_stream_lua_socket_tcp_resume_conn_op( ngx_stream_lua_socket_tcp_conn_op_resume_handler; ngx_post_event((&conn_op_ctx->event), &ngx_posted_events); + + return NGX_OK; } static void -ngx_stream_lua_socket_tcp_conn_op_ctx_cleanup(void *data) +ngx_stream_lua_socket_tcp_release_slot(ngx_stream_lua_socket_pool_t *spool, + ngx_stream_lua_socket_pool_item_t *item) { - ngx_stream_lua_socket_tcp_upstream_t *u; - ngx_stream_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx = data; + if (ngx_stream_lua_socket_tcp_resume_conn_op(spool, item) == NGX_OK) { + return; + } - u = conn_op_ctx->u; + if (item != NULL) { + item->owner = NULL; + ngx_queue_insert_head(&spool->cache, &item->queue); + return; + } - ngx_log_debug2(NGX_LOG_DEBUG_STREAM, u->request->connection->log, 0, - "stream cleanup lua tcp socket " - "conn_op_ctx: %p, u: %p", - conn_op_ctx, u); + spool->connections--; + if (spool->connections == 0) { + ngx_stream_lua_socket_free_pool(ngx_cycle->log, spool); + } +} - ngx_queue_insert_head(&u->socket_pool->cache_connect_op, - &conn_op_ctx->queue); + +static void +ngx_stream_lua_socket_tcp_cleanup_failed_connect(ngx_stream_lua_request_t *r, + ngx_stream_lua_socket_tcp_upstream_t *u) +{ + if (u->conn_closed) { + return; + } + + ngx_stream_lua_socket_tcp_finalize(r, u); + + if (!u->conn_closed) { + /* No peer was created, so finalize() did not release the slot. */ + u->conn_closed = 1; + + if (u->socket_pool != NULL) { + ngx_stream_lua_socket_tcp_release_slot(u->socket_pool, NULL); + } + } +} + + +static void +ngx_stream_lua_socket_tcp_cancel_conn_op( + ngx_stream_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx) +{ + ngx_uint_t queued; + ngx_stream_lua_socket_pool_t *spool; + ngx_stream_lua_socket_pool_item_t *item; + + spool = conn_op_ctx->u->socket_pool; + conn_op_ctx->u->conn_waiting = 0; + conn_op_ctx->u->write_co_ctx = NULL; + queued = conn_op_ctx->queued; + item = conn_op_ctx->item; + conn_op_ctx->item = NULL; + if (item != NULL) { + ngx_stream_lua_assert(item->owner == conn_op_ctx); + item->owner = NULL; + } + + if (conn_op_ctx->event.posted) { + ngx_delete_posted_event((&conn_op_ctx->event)); + } + + if (conn_op_ctx->event.timer_set) { + ngx_del_timer(&conn_op_ctx->event); + } + + ngx_queue_remove(&conn_op_ctx->queue); + ngx_queue_insert_head(&spool->cache_connect_op, &conn_op_ctx->queue); + + if (queued) { + spool->waiting--; + + } else { + ngx_stream_lua_socket_tcp_release_slot(spool, item); + } +} + + +static void +ngx_stream_lua_socket_tcp_conn_op_ctx_cleanup(void *data) +{ + ngx_stream_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx = data; + + conn_op_ctx->cleanup = NULL; + ngx_stream_lua_socket_tcp_cancel_conn_op(conn_op_ctx); } @@ -5007,16 +5077,12 @@ ngx_stream_lua_socket_tcp_conn_op_resume_handler(ngx_event_t *ev) ngx_stream_lua_cleanup_t *cln; ngx_stream_lua_ctx_t *ctx; ngx_stream_lua_co_ctx_t *coctx; - ngx_stream_lua_socket_pool_t *spool; ngx_stream_lua_socket_tcp_upstream_t *u; ngx_stream_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx; conn_op_ctx = ev->data; u = conn_op_ctx->u; r = u->request; - spool = u->socket_pool; - - ngx_queue_remove(&conn_op_ctx->queue); coctx = u->write_co_ctx; coctx->cleanup = NULL; @@ -5027,8 +5093,7 @@ ngx_stream_lua_socket_tcp_conn_op_resume_handler(ngx_event_t *ev) ctx = ngx_stream_lua_get_module_ctx(r, ngx_stream_lua_module); if (ctx == NULL) { - ngx_queue_insert_head(&spool->cache_connect_op, - &conn_op_ctx->queue); + ngx_stream_lua_socket_tcp_cancel_conn_op(conn_op_ctx); return; } @@ -5049,16 +5114,19 @@ ngx_stream_lua_socket_tcp_conn_op_resume_handler(ngx_event_t *ev) } else { cln = ngx_stream_lua_cleanup_add(r, 0); - if (cln != NULL) { - cln->handler = ngx_stream_lua_socket_tcp_conn_op_ctx_cleanup; - cln->data = conn_op_ctx; - conn_op_ctx->cleanup = &cln->handler; + if (cln == NULL) { + ngx_stream_lua_socket_tcp_cancel_conn_op(conn_op_ctx); + ngx_stream_lua_finalize_request(r, NGX_ERROR); + return; } + cln->handler = ngx_stream_lua_socket_tcp_conn_op_ctx_cleanup; + cln->data = conn_op_ctx; + conn_op_ctx->cleanup = &cln->handler; + ctx->resume_handler = ngx_stream_lua_socket_tcp_conn_op_resume; ngx_stream_lua_core_run_phases(r); } - } @@ -5068,8 +5136,11 @@ ngx_stream_lua_socket_tcp_conn_op_resume_retval_handler( ngx_stream_lua_socket_tcp_upstream_t *u, lua_State *L) { int nret; + ngx_int_t rc; ngx_stream_lua_ctx_t *ctx; ngx_stream_lua_co_ctx_t *coctx; + ngx_stream_lua_socket_pool_t *spool; + ngx_stream_lua_socket_pool_item_t *item; ngx_stream_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx; ctx = ngx_stream_lua_get_module_ctx(r, ngx_stream_lua_module); @@ -5086,15 +5157,65 @@ ngx_stream_lua_socket_tcp_conn_op_resume_retval_handler( conn_op_ctx->cleanup = NULL; } - /* decrease pending connect operation counter */ - u->socket_pool->connections--; + u->conn_waiting = 0; + ngx_queue_remove(&conn_op_ctx->queue); + spool = u->socket_pool; + item = conn_op_ctx->item; + conn_op_ctx->item = NULL; + + if (item != NULL) { + ngx_stream_lua_assert(item->owner == conn_op_ctx); + item->owner = NULL; + rc = ngx_stream_lua_socket_tcp_reuse_peer(r, u, item); + ngx_queue_insert_head(&spool->cache_connect_op, &conn_op_ctx->queue); + if (rc != NGX_OK) { + ngx_stream_lua_socket_tcp_finalize(r, u); + lua_pushnil(L); + lua_pushliteral(L, "no memory"); + return 2; + } + + lua_pushinteger(L, 1); + return 1; + } + + /* An idle connection may have been cached after this operation was + * resumed with an empty slot: reuse it rather than open a new one, and + * give the spare slot back. */ + + rc = ngx_stream_lua_get_keepalive_peer(r, u); + if (rc != NGX_DECLINED) { + ngx_queue_insert_head(&spool->cache_connect_op, &conn_op_ctx->queue); + + if (rc == NGX_OK) { + ngx_log_debug0(NGX_LOG_DEBUG_STREAM, r->connection->log, 0, + "stream lua tcp socket resumed connect operation " + "reuses an idle connection, releasing its spare " + "slot"); + + ngx_stream_lua_socket_tcp_release_slot(spool, NULL); + lua_pushinteger(L, 1); + return 1; + } + + /* rc == NGX_ERROR: finalize() closes the idle connection, now + * attached to u, and releases its slot; the spare slot of this + * operation must be released as well */ + + ngx_stream_lua_socket_tcp_finalize(r, u); + ngx_stream_lua_socket_tcp_release_slot(spool, NULL); + lua_pushnil(L); + lua_pushliteral(L, "no memory"); + return 2; + } + + /* The posted operation already owns the slot used by connect_helper(). */ nret = ngx_stream_lua_socket_tcp_connect_helper(L, u, r, ctx, conn_op_ctx->host.data, conn_op_ctx->host.len, conn_op_ctx->port, 1); - ngx_queue_insert_head(&u->socket_pool->cache_connect_op, - &conn_op_ctx->queue); + ngx_queue_insert_head(&spool->cache_connect_op, &conn_op_ctx->queue); return nret; } @@ -5160,14 +5281,7 @@ ngx_stream_lua_socket_tcp_finalize(ngx_stream_lua_request_t *r, return; } - spool->connections--; - - if (spool->connections == 0) { - ngx_stream_lua_socket_free_pool(r->connection->log, spool); - return; - } - - ngx_stream_lua_socket_tcp_resume_conn_op(spool); + ngx_stream_lua_socket_tcp_release_slot(spool, NULL); } } @@ -6199,7 +6313,18 @@ ngx_stream_lua_socket_tcp_setkeepalive(lua_State *L) &spool); } - if (ngx_queue_empty(&spool->free)) { + /* Assigned connections belong to posted operations and cannot be evicted. + * An uncounted pre-pool connection may only replace an unassigned idle + * connection when all slots have already been allocated. */ + if (ngx_queue_empty(&spool->free) + || (u->socket_pool == NULL && spool->backlog >= 0 + && spool->connections >= spool->size)) + { + if (ngx_queue_empty(&spool->cache)) { + ngx_stream_lua_socket_tcp_finalize(r, u); + lua_pushinteger(L, 1); + return 1; + } q = ngx_queue_last(&spool->cache); ngx_queue_remove(q); @@ -6230,6 +6355,7 @@ ngx_stream_lua_socket_tcp_setkeepalive(lua_State *L) } item->connection = c; + item->owner = NULL; ngx_queue_insert_head(&spool->cache, q); ngx_log_debug0(NGX_LOG_DEBUG_STREAM, pc->log, 0, @@ -6296,12 +6422,9 @@ ngx_stream_lua_socket_tcp_setkeepalive(lua_State *L) ngx_stream_lua_socket_tcp_finalize(r, u); #endif - /* since we set u->peer->connection to NULL previously, the connect - * operation won't be resumed in the - * ngx_stream_lua_socket_tcp_finalize. - * Therefore we need to resume it here. - */ - ngx_stream_lua_socket_tcp_resume_conn_op(spool); + /* Transfer this connection and its slot together. */ + ngx_queue_remove(q); + ngx_stream_lua_socket_tcp_release_slot(spool, item); lua_pushinteger(L, 1); return 1; @@ -6312,91 +6435,101 @@ static ngx_int_t ngx_stream_lua_get_keepalive_peer(ngx_stream_lua_request_t *r, ngx_stream_lua_socket_tcp_upstream_t *u) { - ngx_stream_lua_socket_pool_item_t *item; + ngx_queue_t *q; ngx_stream_lua_socket_pool_t *spool; - - ngx_stream_lua_cleanup_t *cln; - ngx_queue_t *q; - ngx_peer_connection_t *pc; - ngx_connection_t *c; + ngx_stream_lua_socket_pool_item_t *item; ngx_log_debug0(NGX_LOG_DEBUG_STREAM, r->connection->log, 0, "stream lua tcp socket pool get keepalive peer"); - pc = &u->peer; - spool = u->socket_pool; - if (!ngx_queue_empty(&spool->cache)) { - q = ngx_queue_head(&spool->cache); + if (ngx_queue_empty(&spool->cache)) { + ngx_log_debug0(NGX_LOG_DEBUG_STREAM, u->peer.log, 0, + "stream lua tcp socket keepalive: connection pool " + "empty"); + return NGX_DECLINED; + } - item = ngx_queue_data(q, ngx_stream_lua_socket_pool_item_t, queue); - c = item->connection; + q = ngx_queue_head(&spool->cache); + ngx_queue_remove(q); + item = ngx_queue_data(q, ngx_stream_lua_socket_pool_item_t, queue); - ngx_queue_remove(q); - ngx_queue_insert_head(&spool->free, q); + return ngx_stream_lua_socket_tcp_reuse_peer(r, u, item); +} + + +static ngx_int_t +ngx_stream_lua_socket_tcp_reuse_peer(ngx_stream_lua_request_t *r, + ngx_stream_lua_socket_tcp_upstream_t *u, + ngx_stream_lua_socket_pool_item_t *item) +{ + ngx_stream_lua_cleanup_t *cln; + ngx_peer_connection_t *pc; + ngx_connection_t *c; + + ngx_stream_lua_assert(item->owner == NULL); + + pc = &u->peer; + c = item->connection; - ngx_log_debug2(NGX_LOG_DEBUG_STREAM, pc->log, 0, - "stream lua tcp socket get keepalive peer: " - "using connection %p, fd:%d", c, c->fd); + ngx_queue_insert_head(&u->socket_pool->free, &item->queue); - pc->socklen = item->socklen; - ngx_memcpy(pc->sockaddr, &item->sockaddr, item->socklen); - ngx_memcpy(u->host, item->host, sizeof(item->host)); + ngx_log_debug2(NGX_LOG_DEBUG_STREAM, pc->log, 0, + "stream lua tcp socket get keepalive peer: " + "using connection %p, fd:%d", c, c->fd); - u->log.handler = ngx_stream_lua_socket_tcp_log_error; - u->log.data = u; + pc->socklen = item->socklen; + ngx_memcpy(pc->sockaddr, &item->sockaddr, item->socklen); + ngx_memcpy(u->host, item->host, sizeof(item->host)); - c->idle = 0; - c->log = pc->log; - c->pool->log = pc->log; - c->read->log = pc->log; - c->write->log = pc->log; - c->data = u; + u->log.handler = ngx_stream_lua_socket_tcp_log_error; + u->log.data = u; + + c->idle = 0; + c->log = pc->log; + c->pool->log = pc->log; + c->read->log = pc->log; + c->write->log = pc->log; + c->data = u; #if 1 - c->write->handler = ngx_stream_lua_socket_tcp_handler; - c->read->handler = ngx_stream_lua_socket_tcp_handler; + c->write->handler = ngx_stream_lua_socket_tcp_handler; + c->read->handler = ngx_stream_lua_socket_tcp_handler; #endif - if (c->read->timer_set) { - ngx_del_timer(c->read); - } + if (c->read->timer_set) { + ngx_del_timer(c->read); + } - pc->connection = c; - pc->cached = 1; - pc->socklen = item->socklen; - ngx_memcpy(&u->sockaddr, &item->sockaddr, item->socklen); - ngx_memcpy(u->host, item->host, sizeof(item->host)); + pc->connection = c; + pc->cached = 1; + pc->socklen = item->socklen; + ngx_memcpy(&u->sockaddr, &item->sockaddr, item->socklen); + ngx_memcpy(u->host, item->host, sizeof(item->host)); - u->log.handler = ngx_stream_lua_socket_tcp_log_error; - u->log.data = u; - u->reused = item->reused + 1; + u->log.handler = ngx_stream_lua_socket_tcp_log_error; + u->log.data = u; + u->reused = item->reused + 1; #if 1 - u->write_event_handler = ngx_stream_lua_socket_dummy_handler; - u->read_event_handler = ngx_stream_lua_socket_dummy_handler; + u->write_event_handler = ngx_stream_lua_socket_dummy_handler; + u->read_event_handler = ngx_stream_lua_socket_dummy_handler; #endif - if (u->cleanup == NULL) { - cln = ngx_stream_lua_cleanup_add(r, 0); - if (cln == NULL) { - u->ft_type |= NGX_STREAM_LUA_SOCKET_FT_ERROR; - return NGX_ERROR; - } - - cln->handler = ngx_stream_lua_socket_tcp_cleanup; - cln->data = u; - u->cleanup = &cln->handler; + if (u->cleanup == NULL) { + cln = ngx_stream_lua_cleanup_add(r, 0); + if (cln == NULL) { + u->ft_type |= NGX_STREAM_LUA_SOCKET_FT_ERROR; + return NGX_ERROR; } - return NGX_OK; + cln->handler = ngx_stream_lua_socket_tcp_cleanup; + cln->data = u; + u->cleanup = &cln->handler; } - ngx_log_debug0(NGX_LOG_DEBUG_STREAM, pc->log, 0, - "stream lua tcp socket keepalive: connection pool empty"); - - return NGX_DECLINED; + return NGX_OK; } @@ -6420,6 +6553,7 @@ ngx_stream_lua_socket_keepalive_close_handler(ngx_event_t *ev) { ngx_stream_lua_socket_pool_item_t *item; ngx_stream_lua_socket_pool_t *spool; + ngx_stream_lua_socket_tcp_conn_op_ctx_t *owner; int n; unsigned char buf[1]; @@ -6465,20 +6599,23 @@ ngx_stream_lua_socket_keepalive_close_handler(ngx_event_t *ev) item = c->data; spool = item->socket_pool; + owner = item->owner; + item->owner = NULL; ngx_stream_lua_socket_tcp_close_connection(c); - ngx_queue_remove(&item->queue); - ngx_queue_insert_head(&spool->free, &item->queue); - spool->connections--; + if (owner != NULL) { + ngx_stream_lua_assert(owner->item == item); + owner->item = NULL; - dd("keepalive: connections: %u", (unsigned) spool->connections); + } else { + ngx_queue_remove(&item->queue); + } - if (spool->connections == 0) { - ngx_stream_lua_socket_free_pool(ev->log, spool); + ngx_queue_insert_head(&spool->free, &item->queue); - } else { - ngx_stream_lua_socket_tcp_resume_conn_op(spool); + if (owner == NULL) { + ngx_stream_lua_socket_tcp_release_slot(spool, NULL); } return NGX_DECLINED; @@ -6516,13 +6653,19 @@ 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); + if (!ngx_queue_empty(&spool->posted_connect_op)) { + ngx_queue_add(&spool->wait_connect_op, &spool->posted_connect_op); + ngx_queue_init(&spool->posted_connect_op); + } + + spool->waiting = 0; while (!ngx_queue_empty(&spool->cache)) { q = ngx_queue_head(&spool->cache); item = ngx_queue_data(q, ngx_stream_lua_socket_pool_item_t, queue); c = item->connection; + item->owner = NULL; ngx_stream_lua_socket_tcp_close_connection(c); @@ -6544,6 +6687,14 @@ ngx_stream_lua_socket_shutdown_pool_helper( conn_op_ctx = ngx_queue_data(q, ngx_stream_lua_socket_tcp_conn_op_ctx_t, queue); + item = conn_op_ctx->item; + if (item != NULL) { + conn_op_ctx->item = NULL; + item->owner = NULL; + ngx_stream_lua_socket_tcp_close_connection(item->connection); + ngx_queue_insert_head(&spool->free, &item->queue); + } + if (conn_op_ctx->event.timer_set) { ngx_del_timer(&conn_op_ctx->event); } @@ -6901,13 +7052,8 @@ ngx_stream_lua_socket_tcp_resume_helper(ngx_stream_lua_request_t *r, "u:%p", prepare_retvals, u); nret = prepare_retvals(r, u, ctx->cur_co_ctx->co); - if (socket_op == SOCKET_OP_CONNECT - && nret > 1 - && !u->conn_closed - && u->socket_pool != NULL) - { - u->socket_pool->connections--; - ngx_stream_lua_socket_tcp_resume_conn_op(u->socket_pool); + if (socket_op == SOCKET_OP_CONNECT && nret > 1) { + ngx_stream_lua_socket_tcp_cleanup_failed_connect(r, u); } if (nret == NGX_AGAIN) { @@ -6945,34 +7091,16 @@ static void ngx_stream_lua_tcp_queue_conn_op_cleanup(void *data) { ngx_stream_lua_co_ctx_t *coctx = data; - ngx_stream_lua_socket_tcp_upstream_t *u; ngx_stream_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx; conn_op_ctx = coctx->data; - u = conn_op_ctx->u; ngx_log_debug2(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, "stream lua tcp socket abort queueing, " "conn_op_ctx: %p, u: %p", - conn_op_ctx, u); - - if (conn_op_ctx->event.posted) { - ngx_delete_posted_event(&conn_op_ctx->event); - - } else if (conn_op_ctx->event.timer_set) { - 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); + conn_op_ctx, conn_op_ctx->u); - u->socket_pool->connections--; - ngx_stream_lua_socket_tcp_resume_conn_op(u->socket_pool); + ngx_stream_lua_socket_tcp_cancel_conn_op(conn_op_ctx); } @@ -6992,8 +7120,7 @@ ngx_stream_lua_tcp_resolve_cleanup(void *data) } if (u->socket_pool != NULL) { - u->socket_pool->connections--; - ngx_stream_lua_socket_tcp_resume_conn_op(u->socket_pool); + ngx_stream_lua_socket_tcp_release_slot(u->socket_pool, NULL); } rctx = u->resolved->ctx; diff --git a/src/ngx_stream_lua_socket_tcp.h b/src/ngx_stream_lua_socket_tcp.h index f44bf9bd..0a5ee2b5 100644 --- a/src/ngx_stream_lua_socket_tcp.h +++ b/src/ngx_stream_lua_socket_tcp.h @@ -45,6 +45,10 @@ typedef struct ngx_stream_lua_socket_tcp_upstream_s ngx_stream_lua_socket_tcp_upstream_t; +typedef struct ngx_stream_lua_socket_pool_item_s + ngx_stream_lua_socket_pool_item_t; + + typedef int (*ngx_stream_lua_socket_tcp_retval_handler)(ngx_stream_lua_request_t *r, ngx_stream_lua_socket_tcp_upstream_t *u, lua_State *L); @@ -60,7 +64,11 @@ typedef struct { ngx_str_t host; ngx_stream_lua_cleanup_pt *cleanup; ngx_stream_lua_socket_tcp_upstream_t *u; + ngx_stream_lua_socket_pool_item_t *item; in_port_t port; + + /* waiting for a slot; otherwise this operation owns one */ + unsigned queued:1; } ngx_stream_lua_socket_tcp_conn_op_ctx_t; @@ -75,20 +83,20 @@ typedef struct { ngx_int_t size; ngx_queue_t cache_connect_op; ngx_queue_t wait_connect_op; + ngx_queue_t posted_connect_op; - /* connections == active connections + pending connect operations, - * while active connections == out-of-pool reused connections - * + in-pool connections */ + /* slots held by active sockets, cached connections, and posted connects */ ngx_int_t connections; + /* connect operations still waiting for a slot */ + ngx_int_t waiting; + /* queues of ngx_stream_lua_socket_pool_item_t: */ ngx_queue_t cache; ngx_queue_t free; ngx_int_t backlog; - ngx_queue_t *next_connect_op; - u_char key[1]; } ngx_stream_lua_socket_pool_t; @@ -183,7 +191,7 @@ typedef struct { } ngx_stream_lua_socket_compiled_pattern_t; -typedef struct { +struct ngx_stream_lua_socket_pool_item_s { ngx_stream_lua_socket_pool_t *socket_pool; ngx_queue_t queue; @@ -195,7 +203,8 @@ typedef struct { ngx_uint_t reused; -} ngx_stream_lua_socket_pool_item_t; + ngx_stream_lua_socket_tcp_conn_op_ctx_t *owner; +}; void ngx_stream_lua_inject_socket_tcp_api(ngx_log_t *log, lua_State *L); diff --git a/t/068-socket-keepalive.t b/t/068-socket-keepalive.t index 43da5d71..397e72a6 100644 --- a/t/068-socket-keepalive.t +++ b/t/068-socket-keepalive.t @@ -2,7 +2,7 @@ use Test::Nginx::Socket::Lua::Stream; -plan tests => repeat_each() * (blocks() * 4 + 32); +plan tests => repeat_each() * (blocks() * 4 + 33); our $HtmlDir = html_dir; @@ -2746,3 +2746,1487 @@ stream lua tcp socket queue connect operation for connection pool "append_posted --- no_error_log [error] --- timeout: 4 + + + +=== TEST 60: conn queuing: a new connect does not take the idle connection handed to a queued one +--- stream_server_config + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "queue_jump", pool_size = 1, backlog = 2} + local sock1 = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + -- queued behind sock1, the only slot + local waiter = ngx.thread.spawn(function() + local s = ngx.socket.tcp() + s:settimeout(1000) + assert(s:connect("127.0.0.1", port, opts)) + local reused = s:getreusedtimes() + assert(s:setkeepalive()) + return "waiter: reused " .. reused + end) + + -- hand the slot to the waiter, whose resumption is only posted + assert(sock1:setkeepalive()) + + -- a connect in the same cycle must queue behind the waiter: taking + -- the idle connection would make the waiter open a second one + local sock3 = ngx.socket.tcp() + sock3:settimeout(1000) + assert(sock3:connect("127.0.0.1", port, opts)) + local reused = sock3:getreusedtimes() + sock3:close() + + local _, res = ngx.thread.wait(waiter) + ngx.say(res) + ngx.say("newcomer: reused ", reused) + } +--- stream_response +waiter: reused 1 +newcomer: reused 2 +--- error_log +stream lua tcp socket queue connect operation for connection pool "queue_jump" +--- no_error_log +[error] + + + +=== TEST 61: conn queuing: a new connect reuses an idle connection no queued one was resumed for +--- stream_server_config + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "spare_idle", 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)) + + -- queued behind sock1 and sock2 + local waiter = ngx.thread.spawn(function() + local s = ngx.socket.tcp() + s:settimeout(1000) + assert(s:connect("127.0.0.1", port, opts)) + local reused = s:getreusedtimes() + -- hold the connection past the newcomer's connect timeout + ngx.sleep(0.5) + assert(s:setkeepalive()) + return "waiter: reused " .. reused + end) + + -- two connections come back, one of them is handed to the waiter + assert(sock1:setkeepalive()) + assert(sock2:setkeepalive()) + + -- the other one is free to reuse: queuing behind the waiter would + -- time out while it sits idle + local sock3 = ngx.socket.tcp() + sock3:settimeout(200) + local ok, err = sock3:connect("127.0.0.1", port, opts) + local res = ok and "newcomer: reused " .. sock3:getreusedtimes() + or "newcomer: " .. err + if ok then + sock3:close() + end + + local _, wres = ngx.thread.wait(waiter) + ngx.say(wres) + ngx.say(res) + } +--- stream_response +waiter: reused 1 +newcomer: reused 1 +--- error_log +stream lua tcp socket queue connect operation for connection pool "spare_idle" +--- no_error_log +[error] +--- timeout: 4 + + + +=== TEST 62: conn queuing: a cancelled resumed connect passes its idle connection to the next one +--- stream_server_config + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "pass_reserved", pool_size = 1, backlog = 3} + local sock1 = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local function waiter(name) + return ngx.thread.spawn(function() + local s = ngx.socket.tcp() + s:settimeout(1000) + assert(s:connect("127.0.0.1", port, opts)) + local reused = s:getreusedtimes() + assert(s:setkeepalive()) + return name .. ": reused " .. reused + end) + end + + local w1 = waiter("w1") + local w2 = waiter("w2") + + -- hand the only connection to w1, then cancel w1 before it runs + assert(sock1:setkeepalive()) + assert(ngx.thread.kill(w1)) + + -- the connection now belongs to w2, not to a new connect + local sock3 = ngx.socket.tcp() + sock3:settimeout(1000) + assert(sock3:connect("127.0.0.1", port, opts)) + local reused = sock3:getreusedtimes() + sock3:close() + + local _, res = ngx.thread.wait(w2) + ngx.say(res) + ngx.say("newcomer: reused ", reused) + } +--- stream_response +w2: reused 1 +newcomer: reused 2 +--- error_log +stream lua tcp socket queue connect operation for connection pool "pass_reserved" +--- no_error_log +[error] + + + +=== TEST 63: conn queuing: cancelling a queued connect frees no slot +--- stream_server_config + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "cancel_queued", pool_size = 1, backlog = 2} + local sock1 = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local function waiter(name) + return ngx.thread.spawn(function() + local s = ngx.socket.tcp() + s:settimeout(300) + local ok, err = s:connect("127.0.0.1", port, opts) + if not ok then + return name .. ": " .. err + end + local reused = s:getreusedtimes() + s:close() + return name .. ": reused " .. reused + end) + end + + local w1 = waiter("w1") + local w2 = waiter("w2") + + -- w1 never got a slot: cancelling it must not let w2 connect while + -- sock1 still holds the only one + assert(ngx.thread.kill(w1)) + + local _, res = ngx.thread.wait(w2) + ngx.say(res) + sock1:close() + } +--- stream_response +w2: timeout +--- error_log +stream lua tcp socket queue connect operation for connection pool "cancel_queued" +--- no_error_log +[alert] + + + +=== TEST 64: conn queuing: a resumed connect cancelled after its idle connection closed hands the slot on +--- stream_server_config + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "reserved_closed", pool_size = 1, backlog = 2} + local sock1 = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local function waiter(name) + return ngx.thread.spawn(function() + local s = ngx.socket.tcp() + s:settimeout(500) + local ok, err = s:connect("127.0.0.1", port, opts) + if not ok then + return name .. ": " .. err + end + local reused = s:getreusedtimes() + s:close() + return name .. ": reused " .. reused + end) + end + + local w1 = waiter("w1") + local w2 = waiter("w2") + + -- runs once this thread sleeps below: its 1ms keepalive timer + -- is then ahead of the sleep timer, and both expire in this + -- cycle before w1's posted resumption runs + local spinner = ngx.thread.spawn(function() + coroutine.yield() + + -- hand the only connection to w1, to be closed before w1 runs + assert(sock1:setkeepalive(1)) + + local t = ngx.now() + repeat + ngx.update_time() + until ngx.now() - t >= 0.005 + end) + + -- wakes up after the reserved connection closed, to cancel w1 + ngx.sleep(0.002) + local killed, err = ngx.thread.kill(w1) + ngx.thread.wait(spinner) + + local _, res = ngx.thread.wait(w2) + ngx.say("killed: ", killed or err) + ngx.say(res) + } +--- stream_response +killed: 1 +w2: reused 0 +--- error_log +stream lua tcp socket queue connect operation for connection pool "reserved_closed" +--- no_error_log +[error] + + + +=== TEST 65: conn queuing: a pre-pool connection cannot replace an assigned connection +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local pool = "replace_reserved" + local legacy = assert(ngx.socket.connect("127.0.0.1", port, + {pool = pool})) + local opts = {pool = pool, pool_size = 1, backlog = 2} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + assert(holder:setkeepalive()) + holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local waiter = ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + sock:settimeout(1000) + assert(sock:connect("127.0.0.1", port, opts)) + local reused = sock:getreusedtimes() + assert(sock:setkeepalive()) + return "waiter: reused " .. reused + end) + + -- The already reused connection belongs to the posted waiter; + -- a pre-pool connection must not replace it or take its slot. + assert(holder:setkeepalive()) + assert(legacy:setkeepalive()) + + local newcomer = ngx.socket.tcp() + newcomer:settimeout(1000) + assert(newcomer:connect("127.0.0.1", port, opts)) + local reused = newcomer:getreusedtimes() + local ok, res = ngx.thread.wait(waiter) + assert(ok, res) + + -- Hold the newcomer open: there must be no second live connection + -- available in this pool of size one. + local extra = ngx.socket.tcp() + extra:settimeout(50) + local connected, err = extra:connect("127.0.0.1", port, opts) + ngx.say(res) + ngx.say("newcomer: reused ", reused) + ngx.say("extra: ", connected or err) + if connected then + extra:close() + end + + newcomer:close() + } +--- stream_response +waiter: reused 2 +newcomer: reused 3 +extra: timeout +--- error_log +lua tcp socket queue connect operation for connection pool "replace_reserved" +--- no_error_log +[error] + + + +=== TEST 66: conn queuing: a pre-pool connection cannot wake another waiter after handoff +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local semaphore = require "ngx.semaphore" + local connected = semaphore.new() + local release = semaphore.new() + local port = $TEST_NGINX_SERVER_PORT + local pool = "replace_reserved_waiters" + local legacy = assert(ngx.socket.connect("127.0.0.1", port, + {pool = pool})) + local opts = {pool = pool, pool_size = 1, backlog = 2} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + assert(holder:setkeepalive()) + holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local w1 = ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + sock:settimeout(1000) + assert(sock:connect("127.0.0.1", port, opts)) + local reused = sock:getreusedtimes() + connected:post() + assert(release:wait(2)) + sock:close() + return "w1: reused " .. reused + end) + + local w2 = ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + sock:settimeout(100) + local ok, err = sock:connect("127.0.0.1", port, opts) + if not ok then + return "w2: " .. err + end + + local reused = sock:getreusedtimes() + sock:close() + return "w2: reused " .. reused + end) + + -- Returning a pre-pool connection must neither replace w1's + -- assigned connection nor resume w2 for the same slot. + assert(holder:setkeepalive()) + assert(legacy:setkeepalive()) + assert(connected:wait(2)) + + local ok, res2 = ngx.thread.wait(w2) + assert(ok, res2) + release:post() + local ok, res1 = ngx.thread.wait(w1) + assert(ok, res1) + ngx.say(res1) + ngx.say(res2) + } +--- stream_response +w1: reused 2 +w2: timeout +--- error_log +lua tcp socket queue connect operation for connection pool "replace_reserved_waiters" +--- no_error_log +[error] +--- timeout: 4 + + + +=== TEST 67: conn queuing: a pre-pool connection cannot take a posted slot +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local pool = "legacy_posted_slot" + local legacy = assert(ngx.socket.connect("127.0.0.1", port, + {pool = pool})) + local opts = {pool = pool, pool_size = 1, backlog = 2} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local waiter = ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + sock:settimeout(1000) + assert(sock:connect("127.0.0.1", port, opts)) + local reused = sock:getreusedtimes() + assert(sock:setkeepalive()) + return "waiter: reused " .. reused + end) + + -- The freed slot belongs to the posted waiter. The uncounted + -- legacy connection must be closed rather than added to cache. + assert(holder:close()) + assert(legacy:setkeepalive()) + + local newcomer = ngx.socket.tcp() + newcomer:settimeout(1000) + assert(newcomer:connect("127.0.0.1", port, opts)) + local reused = newcomer:getreusedtimes() + local ok, res = ngx.thread.wait(waiter) + assert(ok, res) + + -- Hold the newcomer open: there must be no second live connection + -- available in this pool of size one. + local extra = ngx.socket.tcp() + extra:settimeout(50) + local connected, err = extra:connect("127.0.0.1", port, opts) + ngx.say(res) + ngx.say("newcomer: reused ", reused) + ngx.say("extra: ", connected or err) + if connected then + extra:close() + end + + newcomer:close() + } +--- stream_response +waiter: reused 0 +newcomer: reused 1 +extra: timeout +--- error_log +lua tcp socket queue connect operation for connection pool "legacy_posted_slot" +--- no_error_log +[error] + + + +=== TEST 68: conn queuing: a pre-pool connection frees no second posted slot +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local semaphore = require "ngx.semaphore" + local connected = semaphore.new() + local release = semaphore.new() + local port = $TEST_NGINX_SERVER_PORT + local pool = "legacy_posted_waiters" + local legacy = assert(ngx.socket.connect("127.0.0.1", port, + {pool = pool})) + local opts = {pool = pool, pool_size = 1, backlog = 2} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local w1 = ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + sock:settimeout(1000) + assert(sock:connect("127.0.0.1", port, opts)) + local reused = sock:getreusedtimes() + connected:post() + assert(release:wait(2)) + sock:close() + return "w1: reused " .. reused + end) + + local w2 = ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + sock:settimeout(100) + local ok, err = sock:connect("127.0.0.1", port, opts) + if not ok then + return "w2: " .. err + end + + local reused = sock:getreusedtimes() + sock:close() + return "w2: reused " .. reused + end) + + -- The close hands the only slot to w1. Returning an uncounted + -- connection must not also resume w2 for that same slot. + assert(holder:close()) + assert(legacy:setkeepalive()) + assert(connected:wait(2)) + + local ok, res2 = ngx.thread.wait(w2) + assert(ok, res2) + release:post() + local ok, res1 = ngx.thread.wait(w1) + assert(ok, res1) + ngx.say(res1) + ngx.say(res2) + } +--- stream_response +w1: reused 0 +w2: timeout +--- error_log +lua tcp socket queue connect operation for connection pool "legacy_posted_waiters" +--- no_error_log +[error] +--- timeout: 4 + + + +=== TEST 69: conn queuing: a pre-pool connection cannot exceed a busy pool +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local pool = "legacy_busy_pool" + local legacy = assert(ngx.socket.connect("127.0.0.1", port, + {pool = pool})) + local opts = {pool = pool, pool_size = 1, backlog = 1} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + assert(legacy:setkeepalive()) + + local extra = ngx.socket.tcp() + extra:settimeout(50) + local ok, err = extra:connect("127.0.0.1", port, opts) + ngx.say("extra: ", ok or err) + if ok then + extra:close() + end + + assert(holder:close()) + local sock = assert(ngx.socket.connect("127.0.0.1", port, opts)) + ngx.say("after close: reused ", sock:getreusedtimes()) + sock:close() + } +--- stream_response +extra: timeout +after close: reused 0 +--- error_log +lua tcp socket queue connect operation for connection pool "legacy_busy_pool" +--- no_error_log +[error] + + + +=== TEST 70: conn queuing: a pre-pool connection can use spare capacity +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local pool = "legacy_spare_capacity" + local legacy = assert(ngx.socket.connect("127.0.0.1", port, + {pool = pool})) + local opts = {pool = pool, pool_size = 2, backlog = 1} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + assert(legacy:setkeepalive()) + + local newcomer = assert(ngx.socket.connect("127.0.0.1", port, opts)) + ngx.say("newcomer: reused ", newcomer:getreusedtimes()) + + local extra = ngx.socket.tcp() + extra:settimeout(50) + local ok, err = extra:connect("127.0.0.1", port, opts) + ngx.say("extra: ", ok or err) + if ok then + extra:close() + end + + newcomer:close() + holder:close() + } +--- stream_response +newcomer: reused 1 +extra: timeout +--- error_log +lua tcp socket queue connect operation for connection pool "legacy_spare_capacity" +--- no_error_log +[error] + + + +=== TEST 71: conn queuing: a posted reuse leaves spare capacity available +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local semaphore = require "ngx.semaphore" + local ready = semaphore.new() + local release = semaphore.new() + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "posted_reuse_capacity", pool_size = 2, + backlog = 3} + local holder1 = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local holder2 = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local waiter = ngx.thread.spawn(function() + local sock = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local reused = sock:getreusedtimes() + ready:post() + assert(release:wait(2)) + sock:close() + return "waiter: reused " .. reused + end) + + assert(holder1:setkeepalive()) + assert(holder2:close()) + + -- The waiter owns holder1's connection. The slot closed by + -- holder2 is available even before that posted reuse runs. + local newcomer = ngx.socket.tcp() + newcomer:settimeout(100) + local ok, err = newcomer:connect("127.0.0.1", port, opts) + local res = ok and "newcomer: reused " .. newcomer:getreusedtimes() + or "newcomer: " .. err + assert(ready:wait(2)) + + -- Both slots are now occupied: subtracting the reservation must + -- not give another caller a third slot. + local extra = ngx.socket.tcp() + extra:settimeout(50) + local connected, extra_err = extra:connect("127.0.0.1", port, opts) + if connected then + extra:close() + end + + if ok then + newcomer:close() + end + + release:post() + local waited, wres = ngx.thread.wait(waiter) + assert(waited, wres) + ngx.say(wres) + ngx.say(res) + ngx.say("extra: ", connected or extra_err) + } +--- stream_response +waiter: reused 1 +newcomer: reused 0 +extra: timeout +--- error_log +lua tcp socket queue connect operation for connection pool "posted_reuse_capacity" +--- no_error_log +[error] +--- timeout: 4 + + + +=== TEST 72: conn queuing: a reserved connection does not consume backlog twice +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "reserved_backlog", pool_size = 1, backlog = 1} + local holder = 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:settimeout(1000) + local ok, err = sock:connect("127.0.0.1", port, opts) + if not ok then + return name .. ": " .. err + end + + local reused = sock:getreusedtimes() + assert(sock:setkeepalive()) + return name .. ": reused " .. reused + end) + end + + local w1 = waiter("w1") + assert(holder:setkeepalive()) + + -- w1 already owns the connection. Only w2 uses the backlog slot; + -- the next connect must still be rejected while it waits. + local w2 = waiter("w2") + local extra = ngx.socket.tcp() + extra:settimeout(50) + local ok, err = extra:connect("127.0.0.1", port, opts) + ngx.say("extra: ", ok or err) + if ok then + extra:close() + end + + local ok1, res1 = ngx.thread.wait(w1) + local ok2, res2 = ngx.thread.wait(w2) + assert(ok1, res1) + assert(ok2, res2) + ngx.say(res1) + ngx.say(res2) + } +--- stream_response +extra: too many waiting connect operations +w1: reused 1 +w2: reused 2 +--- error_log +lua tcp socket queue connect operation for connection pool "reserved_backlog" +--- no_error_log +[error] + + + +=== TEST 73: conn queuing: cancelling the last posted waiter releases its reservation +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "cancel_last_reserved", pool_size = 1, + backlog = 2} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local waiter = ngx.thread.spawn(function() + assert(ngx.socket.connect("127.0.0.1", port, opts)) + error("cancelled waiter resumed") + end) + + assert(holder:setkeepalive()) + assert(ngx.thread.kill(waiter)) + local newcomer = assert(ngx.socket.connect("127.0.0.1", port, opts)) + ngx.say("newcomer: reused ", newcomer:getreusedtimes()) + + local extra = ngx.socket.tcp() + extra:settimeout(50) + local ok, err = extra:connect("127.0.0.1", port, opts) + ngx.say("extra: ", ok or err) + if ok then + extra:close() + end + + newcomer:close() + } +--- stream_response +newcomer: reused 1 +extra: timeout +--- error_log +lua tcp socket queue connect operation for connection pool "cancel_last_reserved" +--- no_error_log +[error] + + + +=== TEST 74: conn queuing: a timer exiting before its handoff runs releases it +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "aborted_handoff", pool_size = 1, backlog = 1} + + assert(ngx.timer.at(0, function() + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + ngx.thread.spawn(function() + assert(ngx.socket.connect("127.0.0.1", port, opts)) + error("aborted waiter resumed") + end) + + -- the handoff is posted, then the timer ends before it runs + assert(holder:setkeepalive()) + return ngx.exit(ngx.OK) + end)) + + ngx.sleep(0.05) + + local sock = assert(ngx.socket.connect("127.0.0.1", port, opts)) + ngx.say("reused: ", sock:getreusedtimes()) + + local extra = ngx.socket.tcp() + extra:settimeout(50) + local ok, err = extra:connect("127.0.0.1", port, opts) + ngx.say("extra: ", ok or err) + if ok then + extra:close() + end + + sock:close() + } +--- stream_response +reused: 1 +extra: timeout +--- error_log +lua tcp socket abort queueing +--- no_error_log +[error] + + + +=== TEST 75: conn queuing: cancelling an owner cannot release a recycled item +--- stream_server_config + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "recycled_handoff", pool_size = 2, backlog = 2} + local holder1 = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local holder2 = 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:settimeout(500) + assert(sock:connect("127.0.0.1", port, opts)) + local reused = sock:getreusedtimes() + assert(sock:setkeepalive()) + return name .. ": reused " .. reused + end) + end + + local w1 = waiter("w1") + local w2 = waiter("w2") + local spinner = ngx.thread.spawn(function() + coroutine.yield() + assert(holder1:setkeepalive(1)) + + local t = ngx.now() + repeat + ngx.update_time() + until ngx.now() - t >= 0.005 + end) + + ngx.sleep(0.002) + + -- w1's connection closed, returning its item to the free list. + -- Reuse that item for w2 before cancelling w1: the cancellation + -- must release only w1's empty slot, leaving w2's connection alone. + assert(holder2:setkeepalive()) + local killed, err = ngx.thread.kill(w1) + assert(killed, err) + assert(ngx.thread.wait(spinner)) + + local sock = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local reused = sock:getreusedtimes() + local ok, res = ngx.thread.wait(w2) + assert(ok, res) + ngx.say("killed: ", killed) + ngx.say(res) + ngx.say("newcomer: reused ", reused) + sock:close() + } +--- stream_response +killed: 1 +w2: reused 1 +newcomer: reused 0 +--- error_log +lua tcp socket keepalive max idle timeout +--- no_error_log +[error] + + + +=== TEST 76: conn queuing: reconnect cannot overwrite a waiting socket +--- stream_server_config + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "busy_queued", pool_size = 1, backlog = 1} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local sock = ngx.socket.tcp() + sock:settimeout(1000) + local waiter = ngx.thread.spawn(function() + assert(sock:connect("127.0.0.1", port, opts)) + error("cancelled waiter resumed") + end) + + local other = {pool = "busy_other", pool_size = 1, backlog = 1} + local ok, err = sock:connect("127.0.0.1", port, other) + ngx.say("second: ", ok or err) + ngx.say("killed: ", assert(ngx.thread.kill(waiter))) + holder:close() + + -- Cancellation clears the busy state before this socket is reused. + assert(sock:connect("127.0.0.1", port, other)) + ngx.say("retry: reused ", sock:getreusedtimes()) + sock:close() + } +--- stream_response +second: socket busy connecting +killed: 1 +retry: reused 0 +--- error_log +lua tcp socket abort queueing +--- no_error_log +[error] + + + +=== TEST 77: conn queuing: rejected reconnect preserves a posted socket's pool +--- stream_server_config + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "busy_posted", pool_size = 1, backlog = 1} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local sock = ngx.socket.tcp() + local waiter = ngx.thread.spawn(function() + assert(sock:connect("127.0.0.1", port, opts)) + assert(sock:setkeepalive()) + end) + + assert(holder:setkeepalive()) + local other = {pool = "posted_other", pool_size = 1, backlog = 1} + local ok, err = sock:connect("127.0.0.1", port, other) + ngx.say("second: ", ok or err) + assert(ngx.thread.wait(waiter)) + + local next_sock = assert(ngx.socket.connect("127.0.0.1", port, opts)) + ngx.say("original pool: reused ", next_sock:getreusedtimes()) + next_sock:close() + } +--- stream_response +second: socket busy connecting +original pool: reused 2 +--- error_log +lua tcp socket queue connect operation for connection pool "busy_posted" +--- no_error_log +[error] + + + +=== TEST 78: conn queuing: a timed out socket can connect again +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "busy_timeout", pool_size = 1, backlog = 1} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local sock = ngx.socket.tcp() + sock:settimeout(50) + local waiter = ngx.thread.spawn(function() + local ok, err = sock:connect("127.0.0.1", port, opts) + return ok or err + end) + local ok, res = ngx.thread.wait(waiter) + assert(ok, res) + ngx.say("first: ", res) + + sock:settimeout(1000) + local other = {pool = "timeout_retry", pool_size = 1, backlog = 1} + assert(sock:connect("127.0.0.1", port, other)) + ngx.say("retry: reused ", sock:getreusedtimes()) + sock:close() + holder:close() + } +--- stream_response +first: timeout +retry: reused 0 +--- error_log +lua tcp socket queue connect operation for connection pool "busy_timeout" +--- no_error_log +[error] + + + +=== TEST 79: conn queuing: a socket cancelled after handoff can connect again +--- stream_server_config + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "busy_cancelled", pool_size = 1, backlog = 1} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local sock = ngx.socket.tcp() + local waiter = ngx.thread.spawn(function() + assert(sock:connect("127.0.0.1", port, opts)) + error("cancelled waiter resumed") + end) + + assert(holder:setkeepalive()) + ngx.say("killed: ", assert(ngx.thread.kill(waiter))) + assert(sock:connect("127.0.0.1", port, opts)) + ngx.say("retry: reused ", sock:getreusedtimes()) + sock:close() + } +--- stream_response +killed: 1 +retry: reused 1 +--- error_log +lua tcp socket abort queueing +--- no_error_log +[error] + + + +=== TEST 80: conn queuing: killing a parent cancels its posted child coroutine +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "nested_cancel", pool_size = 1, backlog = 1} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local parent = ngx.thread.spawn(function() + coroutine.wrap(function() + local sock = ngx.socket.tcp() + assert(sock:connect("127.0.0.1", port, opts)) + error("cancelled descendant resumed") + end)() + end) + + assert(holder:setkeepalive()) + assert(ngx.thread.kill(parent)) + parent = nil + collectgarbage("collect") + collectgarbage("collect") + + local sock = assert(ngx.socket.connect("127.0.0.1", port, opts)) + ngx.say("reused: ", sock:getreusedtimes()) + + local extra = ngx.socket.tcp() + extra:settimeout(50) + local ok, err = extra:connect("127.0.0.1", port, opts) + ngx.say("extra: ", ok or err) + if ok then + assert(extra:close()) + end + + assert(sock:close()) + } +--- stream_response +reused: 1 +extra: timeout +--- error_log +lua tcp socket abort queueing +--- no_error_log +[error] + + + +=== TEST 81: conn queuing: a waiter resumed in the preread phase takes the handed connection +--- stream_server_config + lua_socket_log_errors off; + + preread_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "preread_handoff", pool_size = 1, backlog = 2} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local waiter = ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + sock:settimeout(1000) + assert(sock:connect("127.0.0.1", port, opts)) + local reused = sock:getreusedtimes() + assert(sock:setkeepalive()) + return reused + end) + + -- the handoff resumes the waiter outside the content phase + assert(holder:setkeepalive()) + local ok, reused = ngx.thread.wait(waiter) + assert(ok, reused) + ngx.ctx.reused = reused + } + + content_by_lua_block { + ngx.say("preread waiter: reused ", ngx.ctx.reused) + } +--- stream_response +preread waiter: reused 1 +--- error_log +lua tcp socket queue connect operation for connection pool "preread_handoff" +--- no_error_log +[error] + + + +=== TEST 82: conn queuing: a stale connection returned to the pool passes its slot to a waiter +--- config + location = /ka { + return 200 "ok"; + } +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "stale_handoff", pool_size = 1, backlog = 1} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local waiter = ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + sock:settimeout(1000) + local ok, err = sock:connect("127.0.0.1", port, opts) + if not ok then + return "waiter: " .. err + end + + local reused = sock:getreusedtimes() + assert(sock:setkeepalive()) + return "waiter: reused " .. reused + end) + + local req = "GET /ka HTTP/1.1\r\nHost: localhost\r\n\r\n" + assert(holder:send(req)) + local read_headers = holder:receiveuntil("\r\n\r\n") + assert(read_headers()) + assert(holder:receive(2)) + + -- An unread response makes the connection readable when it goes + -- back to the pool: it is closed there, and its slot must go to + -- the waiter exactly once. + assert(holder:send(req)) + ngx.sleep(0.05) + + local ok, err = holder:setkeepalive() + ngx.say("setkeepalive: ", ok or err) + + local ok, res = ngx.thread.wait(waiter) + assert(ok, res) + ngx.say(res) + + local sock = ngx.socket.tcp() + sock:settimeout(1000) + assert(sock:connect("127.0.0.1", port, opts)) + ngx.say("newcomer: reused ", sock:getreusedtimes()) + + local extra = ngx.socket.tcp() + extra:settimeout(50) + local connected, err = extra:connect("127.0.0.1", port, opts) + ngx.say("extra: ", connected or err) + if connected then + extra:close() + end + + sock:close() + } +--- stream_response +setkeepalive: connection in dubious state +waiter: reused 0 +newcomer: reused 1 +extra: timeout +--- error_log +lua tcp socket keepalive close handler: fd: +--- no_error_log +[error] + + + +=== TEST 83: conn queuing: waiters refused by the backend pass their slots on +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "refused_backend", pool_size = 1, backlog = 2} + local holder = 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:settimeout(1000) + ngx.update_time() + local start = ngx.now() + + -- nothing listens on port 1 + local ok, err = sock:connect("127.0.0.1", 1, opts) + ngx.update_time() + local slow = ngx.now() - start >= 0.5 and " (slow)" or "" + if ok then + sock:close() + return name .. ": connected" .. slow + end + + return name .. ": " .. err .. slow + end) + end + + local w1 = waiter("w1") + local w2 = waiter("w2") + + -- Each refused waiter must hand the slot to the next one + -- instead of leaving it to the connect timeout. + assert(holder:close()) + + for _, th in ipairs({w1, w2}) do + local ok, res = ngx.thread.wait(th) + assert(ok, res) + ngx.say(res) + end + + local sock = assert(ngx.socket.connect("127.0.0.1", port, opts)) + ngx.say("after: reused ", sock:getreusedtimes()) + + local extra = ngx.socket.tcp() + extra:settimeout(50) + local connected, err = extra:connect("127.0.0.1", port, opts) + ngx.say("extra: ", connected or err) + if connected then + extra:close() + end + + sock:close() + } +--- stream_response +w1: connection refused +w2: connection refused +after: reused 0 +extra: timeout +--- error_log +lua tcp socket queue connect operation for connection pool "refused_backend" +--- no_error_log +[error] + + + +=== TEST 84: conn queuing: a request ending without closing its socket resumes a waiter +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "unclosed_holder", pool_size = 1, backlog = 1} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + assert(ngx.timer.at(0, function() + local sock = ngx.socket.tcp() + sock:settimeout(1000) + local ok, err = sock:connect("127.0.0.1", port, opts) + if not ok then + ngx.log(ngx.WARN, "waiter: ", err) + return + end + + local reused = sock:getreusedtimes() + local extra = ngx.socket.tcp() + extra:settimeout(50) + local connected, cerr = extra:connect("127.0.0.1", port, opts) + if connected then + extra:close() + end + + sock:close() + ngx.log(ngx.WARN, "waiter: reused ", reused, ", extra: ", + connected or cerr) + end)) + + -- let the timer queue behind the holder + ngx.sleep(0.01) + ngx.say(holder and "done") + + -- the holder is left open: the request cleanup closes it and + -- must hand its slot to the waiting timer + } +--- stream_response +done +--- error_log +waiter: reused 0, extra: timeout +--- no_error_log +[error] +--- wait: 0.5 + + + +=== TEST 85: conn queuing: a waiter cancelled by the end of its timer frees its backlog place +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "ended_waiter", pool_size = 1, backlog = 1} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + assert(ngx.timer.at(0, function() + ngx.thread.spawn(function() + assert(ngx.socket.connect("127.0.0.1", port, opts)) + error("cancelled waiter resumed") + end) + + -- the timer ends while its light thread is still queued + return ngx.exit(ngx.OK) + end)) + + ngx.sleep(0.05) + + -- the timer's waiter was cancelled while still queued, so the only + -- backlog place must be free again + local newcomer = ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + sock:settimeout(1000) + local ok, err = sock:connect("127.0.0.1", port, opts) + if not ok then + return "newcomer: " .. err + end + + local reused = sock:getreusedtimes() + sock:close() + return "newcomer: reused " .. reused + end) + + assert(holder:setkeepalive()) + local ok, r = ngx.thread.wait(newcomer) + assert(ok, r) + ngx.say(r) + } +--- stream_response +newcomer: reused 1 +--- error_log +lua tcp socket abort queueing +--- no_error_log +[error] + + + +=== TEST 86: conn queuing: a reconnect rejected as busy reading keeps the socket's pool +--- config + location = /ka { + return 200 "ok"; + } +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local sock = ngx.socket.tcp() + assert(sock:connect("127.0.0.1", port, {pool = "reading_pool"})) + assert(sock:send("GET /ka HTTP/1.1\r\nHost: localhost\r\n\r\n")) + + local reader = ngx.thread.spawn(function() + local read_headers = sock:receiveuntil("\r\n\r\n") + assert(read_headers()) + return assert(sock:receive(2)) + end) + + -- rejected while the reader waits: the pool key must not change + local ok, err = sock:connect("127.0.0.1", port, {pool = "other_pool"}) + ngx.say("reconnect: ", ok or err) + + local ok, body = ngx.thread.wait(reader) + assert(ok, body) + assert(sock:setkeepalive()) + + local again = assert(ngx.socket.connect("127.0.0.1", port, + {pool = "reading_pool"})) + ngx.say("reading_pool: reused ", again:getreusedtimes()) + again:close() + } +--- stream_response +reconnect: socket busy reading +reading_pool: reused 1 +--- error_log +lua tcp socket keepalive create connection pool for key "reading_pool" +--- no_error_log +[error] + + + +=== TEST 87: conn queuing: a pre-pool connection replaces an idle connection in a full pool +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local pool = "legacy_replaces_idle" + local legacy = assert(ngx.socket.connect("127.0.0.1", port, + {pool = pool})) + local opts = {pool = pool, pool_size = 2, backlog = 1} + + -- fill the pool: one active connection and one idle connection, + -- reused once, while a free item is still left + local idle = assert(ngx.socket.connect("127.0.0.1", port, opts)) + assert(idle:setkeepalive()) + idle = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local active = assert(ngx.socket.connect("127.0.0.1", port, opts)) + assert(idle:setkeepalive()) + + -- the uncounted connection may only take the idle one's place + assert(legacy:setkeepalive()) + + local sock = assert(ngx.socket.connect("127.0.0.1", port, opts)) + ngx.say("reused: ", sock:getreusedtimes()) + + local extra = ngx.socket.tcp() + extra:settimeout(50) + local connected, err = extra:connect("127.0.0.1", port, opts) + ngx.say("extra: ", connected or err) + if connected then + extra:close() + end + + sock:close() + active:close() + } +--- stream_response +reused: 1 +extra: timeout +--- error_log +lua tcp socket queue connect operation for connection pool "legacy_replaces_idle" +--- no_error_log +[error] + + + +=== TEST 88: conn queuing: a fast shutdown frees pools with queued connects and idle connections +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local ffi = require "ffi" + pcall(ffi.cdef, [[ + int getpid(void); + int kill(int pid, int sig); + ]]) + + -- collected by lua_close(), which runs after the pools are freed + local marker = newproxy(true) + getmetatable(marker).__gc = function() + ngx.log(ngx.WARN, "Lua VM closed after the pool cleanup") + end + package.loaded.fast_shutdown_marker = marker + + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "fast_shutdown", pool_size = 1, backlog = 2} + + -- another pool keeps an idle connection until the shutdown + local idle = assert(ngx.socket.connect("127.0.0.1", port, + {pool = "fast_shutdown_idle", + pool_size = 1, backlog = 1})) + assert(idle:setkeepalive()) + + assert(ngx.timer.at(0, function() + local holder = assert(ngx.socket.connect("127.0.0.1", port, + opts)) + for _ = 1, 2 do + ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + sock:settimeout(10000) + sock:connect("127.0.0.1", port, opts) + end) + end + + ngx.sleep(10) + holder:close() + end)) + + -- let the timer queue its connects + ngx.sleep(0.05) + ngx.say("ok") + + -- the process exits once the response is out, with both connects + -- still queued (the http server driving the test runs in it too) + assert(ngx.timer.at(0.05, function() + ffi.C.kill(ffi.C.getpid(), 15) + end)) + } +--- stream_response +ok +--- error_log +lua tcp socket keepalive: free connection pool +Lua VM closed after the pool cleanup +--- no_error_log +[alert] +--- wait: 0.3 + + + +=== TEST 89: conn queuing: a waiter resumed with an empty slot reuses a connection cached before it runs +--- stream_server_config + lua_socket_log_errors off; + + content_by_lua_block { + local port = $TEST_NGINX_SERVER_PORT + local opts = {pool = "empty_slot_reuse", pool_size = 2, backlog = 1} + local a = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local b = assert(ngx.socket.connect("127.0.0.1", port, opts)) + + local w = ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + sock:settimeout(1000) + assert(sock:connect("127.0.0.1", port, opts)) + return sock + end) + + -- a frees an empty slot for w; before w runs, b goes back to the + -- pool with nobody waiting any more + assert(a:close()) + assert(b:setkeepalive()) + + local ok, sock = ngx.thread.wait(w) + assert(ok, sock) + ngx.say("w: reused ", sock:getreusedtimes()) + + -- w gave its spare slot back: one more connection fits + local c = ngx.socket.tcp() + c:settimeout(100) + local ok, err = c:connect("127.0.0.1", port, opts) + ngx.say("c: ", ok and "reused " .. c:getreusedtimes() or err) + + local extra = ngx.socket.tcp() + extra:settimeout(50) + local connected, err = extra:connect("127.0.0.1", port, opts) + ngx.say("extra: ", connected or err) + if connected then + extra:close() + end + + if ok then + c:close() + end + + sock:close() + } +--- stream_response +w: reused 1 +c: reused 0 +extra: timeout +--- error_log +lua tcp socket resumed connect operation reuses an idle connection, releasing its spare slot +--- no_error_log +[error] diff --git a/t/196-socket-keepalive-queue.t b/t/196-socket-keepalive-queue.t new file mode 100644 index 00000000..53edcbb6 --- /dev/null +++ b/t/196-socket-keepalive-queue.t @@ -0,0 +1,381 @@ +# vim:set ft= ts=4 sw=4 et fdm=marker: + +use Test::Nginx::Socket::Lua::Stream; + +plan tests => repeat_each() * blocks() * 5; + +log_level 'debug'; +worker_connections(256); +no_long_string(); + +$ENV{TEST_NGINX_HTML_DIR} = html_dir(); + +# Keep the regular CI run short. Increase this for repeated cancellation and +# pool teardown in one worker, for example TEST_NGINX_SOCKET_POOL_ROUNDS=300. +our $Rounds = $ENV{TEST_NGINX_SOCKET_POOL_ROUNDS} // 5; +die "TEST_NGINX_SOCKET_POOL_ROUNDS must be a positive integer\n" + unless $Rounds =~ /\A[1-9][0-9]*\z/; + +our $StressTimeout = 10 + $Rounds; + +our $Helpers = <<'_EOC_'; +>>> queue.lua +local port = tonumber(ngx.var.backend_port) +local phase = ngx.get_phase() +assert(phase == ngx.var.expected_phase, phase) + +-- preread cannot send output: keep the lines for the content phase +local function say(...) + if phase == "content" then + return ngx.say(...) + end + + local out = ngx.ctx.out or {} + ngx.ctx.out = out + out[#out + 1] = table.concat({...}) +end + + +local function check_full(opts) + local extra = ngx.socket.tcp() + extra:settimeout(20) + local ok, err = extra:connect("127.0.0.1", port, opts) + assert(not ok and err == "timeout", err) +end + + +local function run_handoff(name, cancel, close) + local opts = {pool = phase .. name, pool_size = 1, backlog = 3} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local pending = ngx.socket.tcp() + pending:settimeout(1000) + local first = ngx.thread.spawn(function() + assert(pending:connect("127.0.0.1", port, opts)) + assert(not cancel, "cancelled waiter resumed") + local reused = pending:getreusedtimes() + assert(pending:setkeepalive()) + return reused + end) + local second = ngx.thread.spawn(function() + local sock = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local reused = sock:getreusedtimes() + assert(sock:setkeepalive()) + return reused + end) + + if close then + assert(holder:close()) + + else + assert(holder:setkeepalive()) + end + + local other = {pool = phase .. "other", pool_size = 1, backlog = 1} + local ok, err = pending:connect("127.0.0.1", port, other) + assert(not ok and err == "socket busy connecting", err) + + if cancel then + assert(ngx.thread.kill(first)) + end + + collectgarbage("collect") + + -- This connect must queue behind the remaining operation, including + -- when the cancelled operation owned an empty slot instead of a peer. + local sock = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local reused = sock:getreusedtimes() + local ok, second_reused = ngx.thread.wait(second) + assert(ok, second_reused) + + if not cancel then + local ok, first_reused = ngx.thread.wait(first) + assert(ok, first_reused) + say(name, ": ", first_reused, ", ", second_reused, ", ", reused) + + else + say(name, ": ", second_reused, ", ", reused) + end + + check_full(opts) + assert(sock:close()) + + -- Both a consumed connect and a cancelled connect leave a reusable + -- socket object. The old pool has no remaining connection at this point. + assert(pending:connect("127.0.0.1", port, opts)) + assert(pending:getreusedtimes() == 0) + assert(pending:close()) +end + + +say("phase: ", phase) +run_handoff("handoff", false, false) +run_handoff("cancel handoff", true, false) +run_handoff("cancel empty", true, true) + +local opts = {pool = phase .. "timeout", pool_size = 1, backlog = 1} +local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) +local sock = ngx.socket.tcp() +sock:settimeout(20) +local ok, err = sock:connect("127.0.0.1", port, opts) +assert(not ok and err == "timeout", err) +assert(holder:close()) +sock:settimeout(1000) +assert(sock:connect("127.0.0.1", port, opts)) +say("timeout retry: ", sock:getreusedtimes()) +assert(sock:close()) + +>>> stress.lua +local rounds = tonumber(ngx.var.rounds) +local port = tonumber(ngx.var.backend_port) +local seed = 2544990 +local completed, queued, posted, saturated, gc = 0, 0, 0, 0, 0 +local all_posted_cancelled = 0 + + +local function random(n) + seed = (seed * 1664525 + 1013904223) % 4294967296 + return (seed % n) + 1 +end + + +local function full_gc() + collectgarbage("collect") + gc = gc + 1 +end + + +local function waiter(state, i, opts, round) + local sock = ngx.socket.tcp() + sock:settimeout(2000) + assert(sock:connect("127.0.0.1", port, opts)) + state.live = state.live + 1 + assert(state.live <= opts.pool_size, "too many active peers") + + if i % 3 == 0 then + ngx.sleep(0) + end + + state.live = state.live - 1 + + if (i + round) % 2 == 0 then + assert(sock:setkeepalive()) + + else + assert(sock:close()) + end + + state.completed[i] = true + completed = completed + 1 + return true +end + + +local function cancel(state, i, assigned) + if state.cancelled[i] then + return + end + + assert(ngx.thread.kill(state.threads[i])) + state.cancelled[i] = true + + if assigned then + posted = posted + 1 + + else + queued = queued + 1 + end +end + + +for size = 1, 3 do + local opts = {pool = "stress" .. size, pool_size = size, + backlog = 4 * size + 4} + + for round = 1, rounds do + local state = {live = size, threads = {}, cancelled = {}, + completed = {}} + local holders = {} + local mode = (round - 1) % 5 + local count = 2 * size + 3 + + for i = 1, size do + holders[i] = assert(ngx.socket.connect("127.0.0.1", port, opts)) + end + + for i = 1, count do + state.threads[i] = ngx.thread.spawn(waiter, state, i, opts, round) + end + + if mode == 0 then + cancel(state, size + 1, false) + + elseif mode == 3 then + for i = size + 1, count do + if random(3) == 1 then + cancel(state, i, false) + end + end + end + + full_gc() + + -- No yielding between handing out slots and cancelling their owners. + for i = 1, size do + state.live = state.live - 1 + + if mode == 1 or (mode ~= 2 and random(2) == 1) then + assert(holders[i]:setkeepalive()) + + else + assert(holders[i]:close()) + end + end + + if mode == 1 or mode == 2 then + for i = 1, size do + cancel(state, i, true) + end + + all_posted_cancelled = all_posted_cancelled + 1 + + elseif mode == 3 then + for i = 1, size do + if random(2) == 1 then + cancel(state, i, true) + end + end + + elseif mode == 4 then + cancel(state, random(size), true) + cancel(state, size + 1 + random(count - size - 1), false) + end + + full_gc() + + for i = 1, count do + if state.cancelled[i] then + assert(not state.completed[i], "cancelled waiter ran") + + else + local ok, res = ngx.thread.wait(state.threads[i]) + assert(ok and res == true, res) + assert(state.completed[i], "waiter did not finish") + end + end + + assert(state.live == 0, "active peer left behind") + + local final = {} + + for i = 1, size do + local sock = ngx.socket.tcp() + sock:settimeout(1000) + assert(sock:connect("127.0.0.1", port, opts)) + final[i] = sock + end + + local extra = ngx.socket.tcp() + extra:settimeout(5) + local ok, err = extra:connect("127.0.0.1", port, opts) + assert(not ok and err == "timeout", "excess peer admitted") + saturated = saturated + 1 + + for i = 1, size do + if round % 2 == 0 then + assert(final[i]:close()) + + else + assert(final[i]:setkeepalive()) + end + end + + -- Drop dead threads and their sockets before forcing collection. + state = nil + holders = nil + final = nil + full_gc() + end +end + +assert(saturated == 3 * rounds) +assert(gc == 9 * rounds) +assert(completed + queued + posted == 21 * rounds) + +if rounds >= 5 then + assert(queued > 0 and posted > 0 and all_posted_cancelled > 0) +end + +ngx.say("rounds: ", saturated) +ngx.say("gc: ", gc) +ngx.say("all waiters accounted for") +_EOC_ + +add_block_preprocessor(sub { + my $block = shift; + + $block->set_value("user_files", $Helpers); +}); + +run_tests(); + +__DATA__ + +=== TEST 1: queued connects resume, cancel and time out in preread +--- stream_server_config + lua_socket_log_errors off; + set $backend_port $TEST_NGINX_SERVER_PORT; + set $expected_phase preread; + preread_by_lua_file html/queue.lua; + content_by_lua_block { + for _, line in ipairs(ngx.ctx.out) do + ngx.say(line) + end + } +--- stream_response +phase: preread +handoff: 1, 2, 3 +cancel handoff: 1, 2 +cancel empty: 0, 1 +timeout retry: 0 +--- no_error_log +[error] +[alert] +[crit] + + + +=== TEST 2: queued connects resume, cancel and time out in content +--- stream_server_config + lua_socket_log_errors off; + set $backend_port $TEST_NGINX_SERVER_PORT; + set $expected_phase content; + content_by_lua_file html/queue.lua; +--- stream_response +phase: content +handoff: 1, 2, 3 +cancel handoff: 1, 2 +cancel empty: 0, 1 +timeout retry: 0 +--- no_error_log +[error] +[alert] +[crit] + + + +=== TEST 3: repeated cancellation preserves capacity across pool lifetimes +--- stream_server_config eval + qq{ + lua_socket_log_errors off; + set \$backend_port \$TEST_NGINX_SERVER_PORT; + set \$rounds $::Rounds; + content_by_lua_file html/stress.lua; + } +--- stream_response eval + "rounds: " . (3 * $::Rounds) . "\ngc: " . (9 * $::Rounds) + . "\nall waiters accounted for\n" +--- no_error_log +[error] +[alert] +[crit] +--- timeout eval: $::StressTimeout From b3aa80ca696d583b663e5a5e5e38682ba33f76f0 Mon Sep 17 00:00:00 2001 From: lijunlong Date: Thu, 5 Nov 2026 18:37:39 +0800 Subject: [PATCH 2/2] tests: update shared dictionary metatable entry count. Account for the compare_expire method registered by lua-resty-core. Update the shared dictionary expectation from 22 to 23 entries. Validation: t/062-count.t passes all 45 assertions with nginx 1.31.8 and lua-resty-core 81c8704. --- t/062-count.t | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/t/062-count.t b/t/062-count.t index aafa6035..d4367d7d 100644 --- a/t/062-count.t +++ b/t/062-count.t @@ -123,7 +123,7 @@ n = 13 ngx.say("n = ", n) } --- stream_response -n = 22 +n = 23 --- no_error_log [error]