From 9661d95b5cbfaf5c34bf4a627896d16d6c2a34d3 Mon Sep 17 00:00:00 2001 From: AxisNow-Dev Date: Wed, 4 Nov 2026 13:19:43 +0800 Subject: [PATCH] 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, request 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 across request phases and protocols. Document the longer stress run and update the shared dictionary entry count for the new lua-resty-core compare_expire method. Co-authored-by: lijunlong --- README.markdown | 13 + src/ngx_http_lua_socket_tcp.c | 655 +++++++----- src/ngx_http_lua_socket_tcp.h | 23 +- t/062-count.t | 2 +- t/068-socket-keepalive.t | 1773 +++++++++++++++++++++++++++++++- t/196-socket-keepalive-queue.t | 413 ++++++++ 6 files changed, 2602 insertions(+), 277 deletions(-) create mode 100644 t/196-socket-keepalive-queue.t diff --git a/README.markdown b/README.markdown index 96c01ebba3..c2bcfc3b41 100644 --- a/README.markdown +++ b/README.markdown @@ -1058,6 +1058,19 @@ To run specific test files: To run a specific test block in a particular test file, add the line `--- ONLY` to the test block you want to run, and then use the `prove` utility to run that `.t` file. +The cosocket queue regression in `t/196-socket-keepalive-queue.t` checks +handoff, cancellation, and timeout recovery in the rewrite, access, and content +phases. Its stress case uses a fixed random seed and runs five rounds for each +pool size from one to three. Increase `TEST_NGINX_SOCKET_POOL_ROUNDS` for a +longer run in one worker: + + TEST_NGINX_SOCKET_POOL_ROUNDS=300 prove -I. -I/path/to/test-nginx/lib t/196-socket-keepalive-queue.t + +Each round waits for all surviving operations, forces garbage collection, and +checks that the pool admits exactly its configured number of connections. +The test also supports the HTTP/2, HTTP/3, and Valgrind modes of Test::Nginx; +its request and QUIC idle timeouts increase with the round count. + There are also various testing modes based on mockeagain, valgrind, and etc. Refer to the [Test::Nginx documentation](https://search.cpan.org/perldoc?Test::Nginx) for more details for various advanced testing modes. See also the test reports for the Nginx test cluster running on Amazon EC2: . [Back to TOC](#table-of-contents) diff --git a/src/ngx_http_lua_socket_tcp.c b/src/ngx_http_lua_socket_tcp.c index 53317295cb..018d79a765 100644 --- a/src/ngx_http_lua_socket_tcp.c +++ b/src/ngx_http_lua_socket_tcp.c @@ -109,6 +109,9 @@ static void ngx_http_lua_socket_tcp_create_socket_pool(lua_State *L, ngx_int_t backlog, ngx_http_lua_socket_pool_t **spool); static ngx_int_t ngx_http_lua_get_keepalive_peer(ngx_http_request_t *r, ngx_http_lua_socket_tcp_upstream_t *u); +static ngx_int_t ngx_http_lua_socket_tcp_reuse_peer(ngx_http_request_t *r, + ngx_http_lua_socket_tcp_upstream_t *u, + ngx_http_lua_socket_pool_item_t *item); static void ngx_http_lua_socket_keepalive_dummy_handler(ngx_event_t *ev); static int ngx_http_lua_socket_tcp_connect_helper(lua_State *L, ngx_http_lua_socket_tcp_upstream_t *u, ngx_http_request_t *r, @@ -118,8 +121,14 @@ static void ngx_http_lua_socket_tcp_conn_op_timeout_handler( ngx_event_t *ev); static int ngx_http_lua_socket_tcp_conn_op_timeout_retval_handler( ngx_http_request_t *r, ngx_http_lua_socket_tcp_upstream_t *u, lua_State *L); -static void ngx_http_lua_socket_tcp_resume_conn_op( - ngx_http_lua_socket_pool_t *spool); +static void ngx_http_lua_socket_tcp_release_slot( + ngx_http_lua_socket_pool_t *spool, ngx_http_lua_socket_pool_item_t *item); +static void ngx_http_lua_socket_tcp_cleanup_failed_connect( + ngx_http_request_t *r, ngx_http_lua_socket_tcp_upstream_t *u); +static ngx_int_t ngx_http_lua_socket_tcp_resume_conn_op( + ngx_http_lua_socket_pool_t *spool, ngx_http_lua_socket_pool_item_t *item); +static void ngx_http_lua_socket_tcp_cancel_conn_op( + ngx_http_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx); static void ngx_http_lua_socket_tcp_conn_op_ctx_cleanup(void *data); static void ngx_http_lua_socket_tcp_conn_op_resume_handler(ngx_event_t *ev); static ngx_int_t ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev); @@ -557,11 +566,12 @@ ngx_http_lua_socket_tcp_create_socket_pool(lua_State *L, ngx_http_request_t *r, sp->backlog = backlog; sp->size = pool_size; sp->connections = 0; + sp->waiting = 0; sp->lua_vm = ngx_http_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); @@ -604,7 +614,7 @@ ngx_http_lua_socket_tcp_connect_helper(lua_State *L, ngx_http_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx; spool = u->socket_pool; - if (spool != NULL) { + if (spool != NULL && !resuming) { rc = ngx_http_lua_get_keepalive_peer(r, u); if (rc == NGX_OK) { @@ -612,114 +622,120 @@ ngx_http_lua_socket_tcp_connect_helper(lua_State *L, return 1; } - /* rc == NGX_DECLINED */ + if (rc == NGX_ERROR) { + ngx_http_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_HTTP, u->peer.log, 0, - "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_http_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_HTTP, u->peer.log, 0, + "lua tcp socket queue connect operation for " + "connection pool \"%s\", connections: %i", + spool->key, spool->connections); - } else { - conn_op_ctx = ngx_alloc( - sizeof(ngx_http_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_http_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_http_lua_socket_tcp_conn_op_ctx_t), + ngx_cycle->log); + if (conn_op_ctx == NULL) { + return luaL_error(L, "no memory"); + } + + 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"); + } + } - 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->cleanup = NULL; - conn_op_ctx->port = port; + ngx_memcpy(conn_op_host->data, host_ref, host_len); + conn_op_host->data[host_len] = '\0'; + conn_op_host->len = host_len; - u->write_co_ctx = ctx->cur_co_ctx; + conn_op_ctx->port = port; + conn_op_ctx->item = NULL; + conn_op_ctx->queued = 1; - conn_op_ctx->u = u; - ctx->cur_co_ctx->cleanup = - ngx_http_lua_tcp_queue_conn_op_cleanup; - ctx->cur_co_ctx->data = conn_op_ctx; + u->conn_waiting = 1; + u->write_co_ctx = ctx->cur_co_ctx; - ngx_memzero(&conn_op_ctx->event, sizeof(ngx_event_t)); - conn_op_ctx->event.handler = - ngx_http_lua_socket_tcp_conn_op_timeout_handler; - conn_op_ctx->event.data = conn_op_ctx; - conn_op_ctx->event.log = ngx_cycle->log; + conn_op_ctx->u = u; + ctx->cur_co_ctx->cleanup = + ngx_http_lua_tcp_queue_conn_op_cleanup; + ctx->cur_co_ctx->data = conn_op_ctx; - ngx_add_timer(&conn_op_ctx->event, u->connect_timeout); + ngx_memzero(&conn_op_ctx->event, sizeof(ngx_event_t)); + conn_op_ctx->event.handler = + ngx_http_lua_socket_tcp_conn_op_timeout_handler; + conn_op_ctx->event.data = conn_op_ctx; + conn_op_ctx->event.log = ngx_cycle->log; - ngx_queue_insert_tail(&spool->wait_connect_op, - &conn_op_ctx->queue); + ngx_add_timer(&conn_op_ctx->event, u->connect_timeout); - if (spool->next_connect_op - == ngx_queue_sentinel(&spool->wait_connect_op)) - { - spool->next_connect_op = &conn_op_ctx->queue; - } + ngx_queue_insert_tail(&spool->wait_connect_op, + &conn_op_ctx->queue); - ngx_log_debug3(NGX_LOG_DEBUG_HTTP, ngx_cycle->log, 0, - "lua tcp socket queued connect operation for " - "%d(ms), u: %p, ctx: %p", - u->connect_timeout, conn_op_ctx->u, conn_op_ctx); + spool->waiting++; - return lua_yield(L, 0); - } + ngx_log_debug3(NGX_LOG_DEBUG_HTTP, ngx_cycle->log, 0, + "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; @@ -900,19 +916,13 @@ ngx_http_lua_socket_tcp_connect_helper(lua_State *L, failed: - if (spool != NULL) { - spool->connections--; - ngx_http_lua_socket_tcp_resume_conn_op(spool); - } + ngx_http_lua_socket_tcp_cleanup_failed_connect(r, u); return 2; no_memory_and_not_resuming: - if (spool != NULL) { - spool->connections--; - ngx_http_lua_socket_tcp_resume_conn_op(spool); - } + ngx_http_lua_socket_tcp_cleanup_failed_connect(r, u); return luaL_error(L, "no memory"); } @@ -1096,10 +1106,6 @@ ngx_http_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; @@ -1172,13 +1178,6 @@ ngx_http_lua_socket_tcp_connect(lua_State *L) return 2; } - 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); @@ -1222,6 +1221,10 @@ ngx_http_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_http_lua_socket_tcp_upstream_t)); u->request = r; /* set the controlling request */ @@ -4329,11 +4332,8 @@ ngx_http_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; @@ -4342,6 +4342,7 @@ ngx_http_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_http_get_module_loc_conf(r, ngx_http_lua_module); @@ -4354,7 +4355,6 @@ ngx_http_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_http_get_module_ctx(r, ngx_http_lua_module); if (ctx == NULL) { @@ -4396,39 +4396,35 @@ ngx_http_lua_socket_tcp_conn_op_timeout_retval_handler(ngx_http_request_t *r, } -static void -ngx_http_lua_socket_tcp_resume_conn_op(ngx_http_lua_socket_pool_t *spool) +static ngx_int_t +ngx_http_lua_socket_tcp_resume_conn_op(ngx_http_lua_socket_pool_t *spool, + ngx_http_lua_socket_pool_item_t *item) { ngx_queue_t *q; ngx_http_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx; -#if (NGX_DEBUG) - ngx_http_lua_assert(spool->connections >= 0); - -#else - if (spool->connections < 0) { - ngx_log_error(NGX_LOG_ERR, ngx_cycle->log, 0, - "lua tcp socket connections count mismatched for " - "connection pool \"%s\", connections: %i, size: %i", - spool->key, spool->connections, spool->size); - spool->connections = 0; - } -#endif + ngx_http_lua_assert(spool->connections > 0); + ngx_http_lua_assert(spool->waiting >= 0); + ngx_http_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_http_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_HTTP, ngx_cycle->log, 0, "lua tcp socket post connect operation resumption " "u: %p, ctx: %p for connection pool \"%s\", " @@ -4443,24 +4439,103 @@ ngx_http_lua_socket_tcp_resume_conn_op(ngx_http_lua_socket_pool_t *spool) ngx_http_lua_socket_tcp_conn_op_resume_handler; ngx_post_event((&conn_op_ctx->event), &ngx_posted_events); + + return NGX_OK; } static void -ngx_http_lua_socket_tcp_conn_op_ctx_cleanup(void *data) +ngx_http_lua_socket_tcp_release_slot(ngx_http_lua_socket_pool_t *spool, + ngx_http_lua_socket_pool_item_t *item) { - ngx_http_lua_socket_tcp_upstream_t *u; - ngx_http_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx = data; + if (ngx_http_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_debug3(NGX_LOG_DEBUG_HTTP, u->request->connection->log, 0, - "cleanup lua tcp socket conn_op_ctx: %p, u: %p, " - "request: \"%V\"", - conn_op_ctx, u, &u->request->uri); + spool->connections--; + if (spool->connections == 0) { + ngx_http_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_http_lua_socket_tcp_cleanup_failed_connect(ngx_http_request_t *r, + ngx_http_lua_socket_tcp_upstream_t *u) +{ + if (u->conn_closed) { + return; + } + + ngx_http_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_http_lua_socket_tcp_release_slot(u->socket_pool, NULL); + } + } +} + + +static void +ngx_http_lua_socket_tcp_cancel_conn_op( + ngx_http_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx) +{ + ngx_uint_t queued; + ngx_http_lua_socket_pool_t *spool; + ngx_http_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_http_lua_assert(item->owner == conn_op_ctx); + item->owner = NULL; + } + +#if (nginx_version >= 1007005) + if (conn_op_ctx->event.posted) { +#else + if (conn_op_ctx->event.prev) { +#endif + ngx_delete_posted_event((&conn_op_ctx->event)); + } + + 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_http_lua_socket_tcp_release_slot(spool, item); + } +} + + +static void +ngx_http_lua_socket_tcp_conn_op_ctx_cleanup(void *data) +{ + ngx_http_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx = data; + + conn_op_ctx->cleanup = NULL; + ngx_http_lua_socket_tcp_cancel_conn_op(conn_op_ctx); } @@ -4472,16 +4547,12 @@ ngx_http_lua_socket_tcp_conn_op_resume_handler(ngx_event_t *ev) ngx_http_request_t *r; ngx_http_cleanup_t *cln; ngx_http_lua_co_ctx_t *coctx; - ngx_http_lua_socket_pool_t *spool; ngx_http_lua_socket_tcp_upstream_t *u; ngx_http_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; @@ -4492,8 +4563,7 @@ ngx_http_lua_socket_tcp_conn_op_resume_handler(ngx_event_t *ev) ctx = ngx_http_get_module_ctx(r, ngx_http_lua_module); if (ctx == NULL) { - ngx_queue_insert_head(&spool->cache_connect_op, - &conn_op_ctx->queue); + ngx_http_lua_socket_tcp_cancel_conn_op(conn_op_ctx); return; } @@ -4515,12 +4585,16 @@ ngx_http_lua_socket_tcp_conn_op_resume_handler(ngx_event_t *ev) } else { cln = ngx_http_lua_cleanup_add(r, 0); - if (cln != NULL) { - cln->handler = ngx_http_lua_socket_tcp_conn_op_ctx_cleanup; - cln->data = conn_op_ctx; - conn_op_ctx->cleanup = &cln->handler; + if (cln == NULL) { + ngx_http_lua_socket_tcp_cancel_conn_op(conn_op_ctx); + ngx_http_lua_finalize_request(r, NGX_HTTP_INTERNAL_SERVER_ERROR); + return; } + cln->handler = ngx_http_lua_socket_tcp_conn_op_ctx_cleanup; + cln->data = conn_op_ctx; + conn_op_ctx->cleanup = &cln->handler; + ctx->resume_handler = ngx_http_lua_socket_tcp_conn_op_resume; ngx_http_core_run_phases(r); } @@ -4536,6 +4610,9 @@ ngx_http_lua_socket_tcp_conn_op_resume_retval_handler(ngx_http_request_t *r, int nret; ngx_http_lua_ctx_t *ctx; ngx_http_lua_co_ctx_t *coctx; + ngx_int_t rc; + ngx_http_lua_socket_pool_t *spool; + ngx_http_lua_socket_pool_item_t *item; ngx_http_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx; ctx = ngx_http_get_module_ctx(r, ngx_http_lua_module); @@ -4552,9 +4629,60 @@ ngx_http_lua_socket_tcp_conn_op_resume_retval_handler(ngx_http_request_t *r, 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); + item = conn_op_ctx->item; + conn_op_ctx->item = NULL; + + if (item != NULL) { + ngx_http_lua_assert(item->owner == conn_op_ctx); + item->owner = NULL; + rc = ngx_http_lua_socket_tcp_reuse_peer(r, u, item); + ngx_queue_insert_head(&u->socket_pool->cache_connect_op, + &conn_op_ctx->queue); + if (rc != NGX_OK) { + ngx_http_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. */ + spool = u->socket_pool; + rc = ngx_http_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_HTTP, r->connection->log, 0, + "lua tcp socket resumed connect operation reuses " + "an idle connection, releasing its spare slot"); + + ngx_http_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_http_lua_socket_tcp_finalize(r, u); + ngx_http_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_http_lua_socket_tcp_connect_helper(L, u, r, ctx, conn_op_ctx->host.data, conn_op_ctx->host.len, @@ -4628,14 +4756,7 @@ ngx_http_lua_socket_tcp_finalize(ngx_http_request_t *r, return; } - spool->connections--; - - if (spool->connections == 0) { - ngx_http_lua_socket_free_pool(r->connection->log, spool); - return; - } - - ngx_http_lua_socket_tcp_resume_conn_op(spool); + ngx_http_lua_socket_tcp_release_slot(spool, NULL); } } @@ -5781,7 +5902,18 @@ ngx_http_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_http_lua_socket_tcp_finalize(r, u); + lua_pushinteger(L, 1); + return 1; + } q = ngx_queue_last(&spool->cache); ngx_queue_remove(q); @@ -5812,6 +5944,7 @@ ngx_http_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_HTTP, pc->log, 0, @@ -5879,11 +6012,9 @@ ngx_http_lua_socket_tcp_setkeepalive(lua_State *L) ngx_http_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_http_lua_socket_tcp_finalize. - * Therefore we need to resume it here. - */ - ngx_http_lua_socket_tcp_resume_conn_op(spool); + /* Transfer this connection and its slot together. */ + ngx_queue_remove(q); + ngx_http_lua_socket_tcp_release_slot(spool, item); lua_pushinteger(L, 1); return 1; @@ -5894,84 +6025,89 @@ static ngx_int_t ngx_http_lua_get_keepalive_peer(ngx_http_request_t *r, ngx_http_lua_socket_tcp_upstream_t *u) { - ngx_http_lua_socket_pool_item_t *item; - ngx_http_lua_socket_pool_t *spool; - ngx_http_cleanup_t *cln; ngx_queue_t *q; - ngx_peer_connection_t *pc; - ngx_connection_t *c; + ngx_http_lua_socket_pool_t *spool = u->socket_pool; + ngx_http_lua_socket_pool_item_t *item; ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0, "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_HTTP, u->peer.log, 0, + "lua tcp socket keepalive: connection pool empty"); + return NGX_DECLINED; + } - item = ngx_queue_data(q, ngx_http_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_http_lua_socket_pool_item_t, queue); - ngx_queue_remove(q); - ngx_queue_insert_head(&spool->free, q); + return ngx_http_lua_socket_tcp_reuse_peer(r, u, item); +} - ngx_log_debug2(NGX_LOG_DEBUG_HTTP, pc->log, 0, - "lua tcp socket get keepalive peer: using connection %p," - " fd:%d", c, c->fd); - 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; +static ngx_int_t +ngx_http_lua_socket_tcp_reuse_peer(ngx_http_request_t *r, + ngx_http_lua_socket_tcp_upstream_t *u, + ngx_http_lua_socket_pool_item_t *item) +{ + ngx_http_cleanup_t *cln; + ngx_peer_connection_t *pc = &u->peer; + ngx_connection_t *c = item->connection; + + ngx_http_lua_assert(item->owner == NULL); + ngx_queue_insert_head(&u->socket_pool->free, &item->queue); + + ngx_log_debug2(NGX_LOG_DEBUG_HTTP, pc->log, 0, + "lua tcp socket get keepalive peer: using connection %p," + " fd:%d", c, c->fd); + + 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_http_lua_socket_tcp_handler; - c->read->handler = ngx_http_lua_socket_tcp_handler; + c->write->handler = ngx_http_lua_socket_tcp_handler; + c->read->handler = ngx_http_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_http_lua_socket_tcp_log_error; - u->log.data = u; - u->reused = item->reused + 1; - u->udata_queue = item->udata_queue; - item->udata_queue = NULL; + u->log.handler = ngx_http_lua_socket_tcp_log_error; + u->log.data = u; + u->reused = item->reused + 1; + u->udata_queue = item->udata_queue; + item->udata_queue = NULL; #if 1 - u->write_event_handler = ngx_http_lua_socket_dummy_handler; - u->read_event_handler = ngx_http_lua_socket_dummy_handler; + u->write_event_handler = ngx_http_lua_socket_dummy_handler; + u->read_event_handler = ngx_http_lua_socket_dummy_handler; #endif - if (u->cleanup == NULL) { - cln = ngx_http_lua_cleanup_add(r, 0); - if (cln == NULL) { - u->ft_type |= NGX_HTTP_LUA_SOCKET_FT_ERROR; - return NGX_ERROR; - } - - cln->handler = ngx_http_lua_socket_tcp_cleanup; - cln->data = u; - u->cleanup = &cln->handler; + if (u->cleanup == NULL) { + cln = ngx_http_lua_cleanup_add(r, 0); + if (cln == NULL) { + u->ft_type |= NGX_HTTP_LUA_SOCKET_FT_ERROR; + return NGX_ERROR; } - return NGX_OK; + cln->handler = ngx_http_lua_socket_tcp_cleanup; + cln->data = u; + u->cleanup = &cln->handler; } - ngx_log_debug0(NGX_LOG_DEBUG_HTTP, pc->log, 0, - "lua tcp socket keepalive: connection pool empty"); - - return NGX_DECLINED; + return NGX_OK; } @@ -5993,8 +6129,9 @@ ngx_http_lua_socket_keepalive_rev_handler(ngx_event_t *ev) static ngx_int_t ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) { - ngx_http_lua_socket_pool_item_t *item; - ngx_http_lua_socket_pool_t *spool; + ngx_http_lua_socket_pool_item_t *item; + ngx_http_lua_socket_pool_t *spool; + ngx_http_lua_socket_tcp_conn_op_ctx_t *owner; int n; unsigned char buf[1]; @@ -6038,20 +6175,23 @@ ngx_http_lua_socket_keepalive_close_handler(ngx_event_t *ev) item = c->data; spool = item->socket_pool; + owner = item->owner; + item->owner = NULL; ngx_http_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_http_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_http_lua_socket_free_pool(ev->log, spool); + ngx_queue_insert_head(&spool->free, &item->queue); - } else { - ngx_http_lua_socket_tcp_resume_conn_op(spool); + if (owner == NULL) { + ngx_http_lua_socket_tcp_release_slot(spool, NULL); } return NGX_DECLINED; @@ -6086,13 +6226,19 @@ ngx_http_lua_socket_shutdown_pool_helper(ngx_http_lua_socket_pool_t *spool) ngx_http_lua_socket_pool_item_t *item; ngx_http_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_http_lua_socket_pool_item_t, queue); c = item->connection; + item->owner = NULL; ngx_http_lua_socket_tcp_close_connection(c); @@ -6114,6 +6260,14 @@ ngx_http_lua_socket_shutdown_pool_helper(ngx_http_lua_socket_pool_t *spool) conn_op_ctx = ngx_queue_data(q, ngx_http_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_http_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); } @@ -6485,13 +6639,8 @@ ngx_http_lua_socket_tcp_resume_helper(ngx_http_request_t *r, int socket_op) "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_http_lua_socket_tcp_resume_conn_op(u->socket_pool); + if (socket_op == SOCKET_OP_CONNECT && nret > 1) { + ngx_http_lua_socket_tcp_cleanup_failed_connect(r, u); } if (nret == NGX_AGAIN) { @@ -6529,42 +6678,13 @@ static void ngx_http_lua_tcp_queue_conn_op_cleanup(void *data) { ngx_http_lua_co_ctx_t *coctx = data; - ngx_http_lua_socket_tcp_upstream_t *u; - ngx_http_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx; - - conn_op_ctx = coctx->data; - u = conn_op_ctx->u; + ngx_http_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx = coctx->data; ngx_log_debug2(NGX_LOG_DEBUG_HTTP, ngx_cycle->log, 0, "lua tcp socket abort queueing, conn_op_ctx: %p, u: %p", - conn_op_ctx, u); - -#if (nginx_version >= 1007005) - if (conn_op_ctx->event.posted) { -#else - if (conn_op_ctx->event.prev) { -#endif - /* - * We need the extra parentheses around the argument - * of ngx_delete_posted_event() just to work around macro issues in - * nginx cores older than 1.7.5 (exclusive). - */ - 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_http_lua_socket_tcp_resume_conn_op(u->socket_pool); + ngx_http_lua_socket_tcp_cancel_conn_op(conn_op_ctx); } @@ -6584,8 +6704,7 @@ ngx_http_lua_tcp_resolve_cleanup(void *data) } if (u->socket_pool != NULL) { - u->socket_pool->connections--; - ngx_http_lua_socket_tcp_resume_conn_op(u->socket_pool); + ngx_http_lua_socket_tcp_release_slot(u->socket_pool, NULL); } rctx = u->resolved->ctx; diff --git a/src/ngx_http_lua_socket_tcp.h b/src/ngx_http_lua_socket_tcp.h index 73b5c1153b..cca846b79c 100644 --- a/src/ngx_http_lua_socket_tcp.h +++ b/src/ngx_http_lua_socket_tcp.h @@ -36,6 +36,9 @@ typedef struct ngx_http_lua_socket_tcp_upstream_s typedef struct ngx_http_lua_socket_udata_queue_s ngx_http_lua_socket_udata_queue_t; +typedef struct ngx_http_lua_socket_pool_item_s + ngx_http_lua_socket_pool_item_t; + typedef int (*ngx_http_lua_socket_tcp_retval_handler)(ngx_http_request_t *r, @@ -52,7 +55,11 @@ typedef struct { ngx_str_t host; ngx_http_cleanup_pt *cleanup; ngx_http_lua_socket_tcp_upstream_t *u; + ngx_http_lua_socket_pool_item_t *item; in_port_t port; + + /* waiting for a slot; otherwise this operation owns one */ + unsigned queued:1; } ngx_http_lua_socket_tcp_conn_op_ctx_t; @@ -67,20 +74,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_http_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_http_lua_socket_pool_t; @@ -181,7 +188,7 @@ typedef struct { } ngx_http_lua_socket_compiled_pattern_t; -typedef struct { +struct ngx_http_lua_socket_pool_item_s { ngx_http_lua_socket_pool_t *socket_pool; ngx_queue_t queue; @@ -194,7 +201,9 @@ typedef struct { ngx_uint_t reused; ngx_http_lua_socket_udata_queue_t *udata_queue; -} ngx_http_lua_socket_pool_item_t; + + ngx_http_lua_socket_tcp_conn_op_ctx_t *owner; +}; struct ngx_http_lua_socket_udata_queue_s { diff --git a/t/062-count.t b/t/062-count.t index c92476617f..eac15aebb1 100644 --- a/t/062-count.t +++ b/t/062-count.t @@ -284,7 +284,7 @@ n = 9 --- request GET /test --- response_body -n = 22 +n = 23 --- no_error_log [error] diff --git a/t/068-socket-keepalive.t b/t/068-socket-keepalive.t index 1a6decce54..bc308afc9f 100644 --- a/t/068-socket-keepalive.t +++ b/t/068-socket-keepalive.t @@ -4,7 +4,7 @@ use Test::Nginx::Socket::Lua; #repeat_each(2); -plan tests => repeat_each() * (blocks() * 4 + 34); +plan tests => repeat_each() * (blocks() * 4 + 35); our $HtmlDir = html_dir; @@ -3480,3 +3480,1774 @@ lua tcp socket queue connect operation for connection pool "append_posted" --- no_error_log [error] --- timeout: 4 + + + +=== TEST 61: conn queuing: a new connect does not take the idle connection handed to a queued one +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + + content_by_lua_block { + local port = ngx.var.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 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) + + -- 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) + } + } +--- request +GET /t +--- response_body +waiter: reused 1 +newcomer: reused 2 +--- error_log +lua tcp socket queue connect operation for connection pool "queue_jump" +--- no_error_log +[error] + + + +=== TEST 62: conn queuing: a new connect reuses an idle connection no queued one was resumed for +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + + content_by_lua_block { + local port = ngx.var.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 sock = ngx.socket.tcp() + sock:settimeout(1000) + assert(sock:connect("127.0.0.1", port, opts)) + local reused = sock:getreusedtimes() + -- hold the connection past the newcomer's connect timeout + ngx.sleep(0.5) + assert(sock: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) + } + } +--- request +GET /t +--- response_body +waiter: reused 1 +newcomer: reused 1 +--- error_log +lua tcp socket queue connect operation for connection pool "spare_idle" +--- no_error_log +[error] +--- timeout: 4 + + + +=== TEST 63: conn queuing: a cancelled resumed connect passes its idle connection to the next one +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + + content_by_lua_block { + local port = ngx.var.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 sock = ngx.socket.tcp() + sock:settimeout(1000) + 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") + + -- 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) + } + } +--- request +GET /t +--- response_body +w2: reused 1 +newcomer: reused 2 +--- error_log +lua tcp socket queue connect operation for connection pool "pass_reserved" +--- no_error_log +[error] + + + +=== TEST 64: conn queuing: cancelling a queued connect frees no slot +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + + content_by_lua_block { + local port = ngx.var.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 sock = ngx.socket.tcp() + sock:settimeout(300) + local ok, err = sock:connect("127.0.0.1", port, opts) + if not ok then + return name .. ": " .. err + end + local reused = sock:getreusedtimes() + sock: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() + } + } +--- request +GET /t +--- response_body +w2: timeout +--- error_log +lua tcp socket queue connect operation for connection pool "cancel_queued" +--- no_error_log +[alert] + + + +=== TEST 65: conn queuing: a resumed connect cancelled after its idle connection closed hands the slot on +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + + content_by_lua_block { + local port = ngx.var.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 sock = ngx.socket.tcp() + sock:settimeout(500) + local ok, err = sock:connect("127.0.0.1", port, opts) + if not ok then + return name .. ": " .. err + end + local reused = sock:getreusedtimes() + sock: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) + } + } +--- request +GET /t +--- response_body +killed: 1 +w2: reused 0 +--- error_log +lua tcp socket queue connect operation for connection pool "reserved_closed" +--- no_error_log +[error] + + + +=== TEST 66: conn queuing: a pre-pool connection cannot replace an assigned connection +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + lua_socket_log_errors off; + + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +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 67: conn queuing: a pre-pool connection cannot wake another waiter after handoff +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + lua_socket_log_errors off; + + content_by_lua_block { + local semaphore = require "ngx.semaphore" + local connected = semaphore.new() + local release = semaphore.new() + local port = ngx.var.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) + } + } +--- request +GET /t +--- response_body +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 68: conn queuing: a pre-pool connection cannot take a posted slot +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + lua_socket_log_errors off; + + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +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 69: conn queuing: a pre-pool connection frees no second posted slot +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + lua_socket_log_errors off; + + content_by_lua_block { + local semaphore = require "ngx.semaphore" + local connected = semaphore.new() + local release = semaphore.new() + local port = ngx.var.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) + } + } +--- request +GET /t +--- response_body +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 70: conn queuing: a pre-pool connection cannot exceed a busy pool +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + lua_socket_log_errors off; + + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +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 71: conn queuing: a pre-pool connection can use spare capacity +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + lua_socket_log_errors off; + + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +newcomer: reused 1 +extra: timeout +--- error_log +lua tcp socket queue connect operation for connection pool "legacy_spare_capacity" +--- no_error_log +[error] + + + +=== TEST 72: conn queuing: a posted reuse leaves spare capacity available +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + lua_socket_log_errors off; + + content_by_lua_block { + local semaphore = require "ngx.semaphore" + local ready = semaphore.new() + local release = semaphore.new() + local port = ngx.var.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) + } + } +--- request +GET /t +--- response_body +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 73: conn queuing: a reserved connection does not consume backlog twice +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + lua_socket_log_errors off; + + content_by_lua_block { + local port = ngx.var.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) + } + } +--- request +GET /t +--- response_body +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 74: conn queuing: cancelling the last posted waiter releases its reservation +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + lua_socket_log_errors off; + + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +newcomer: reused 1 +extra: timeout +--- error_log +lua tcp socket queue connect operation for connection pool "cancel_last_reserved" +--- no_error_log +[error] + + + +=== TEST 75: conn queuing: request cleanup releases an unexecuted handoff +--- config + set $port $TEST_NGINX_SERVER_PORT; + + location /abort { + content_by_lua_block { + local port = ngx.var.port + local opts = {pool = "aborted_handoff", pool_size = 1, backlog = 1} + 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) + + assert(holder:setkeepalive()) + return ngx.exit(204) + } + } + + location /t { + lua_socket_log_errors off; + content_by_lua_block { + local res = ngx.location.capture("/abort") + local port = ngx.var.port + local opts = {pool = "aborted_handoff", pool_size = 1, backlog = 1} + local sock = assert(ngx.socket.connect("127.0.0.1", port, opts)) + ngx.say("status: ", res.status) + 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() + } + } +--- request +GET /t +--- response_body +status: 204 +reused: 1 +extra: timeout +--- error_log +lua tcp socket abort queueing +--- no_error_log +[error] + + + +=== TEST 76: conn queuing: cancelling an owner cannot release a recycled item +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +killed: 1 +w2: reused 1 +newcomer: reused 0 +--- error_log +lua tcp socket keepalive max idle timeout +--- no_error_log +[error] + + + +=== TEST 77: conn queuing: reconnect cannot overwrite a waiting socket +--- config + location /t { + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +second: socket busy connecting +killed: 1 +retry: reused 0 +--- error_log +lua tcp socket abort queueing +--- no_error_log +[error] + + + +=== TEST 78: conn queuing: rejected reconnect preserves a posted socket's pool +--- config + location /t { + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +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 79: conn queuing: a timed out socket can connect again +--- config + location /t { + lua_socket_log_errors off; + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +first: timeout +retry: reused 0 +--- error_log +lua tcp socket queue connect operation for connection pool "busy_timeout" +--- no_error_log +[error] + + + +=== TEST 80: conn queuing: a socket cancelled after handoff can connect again +--- config + location /t { + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +killed: 1 +retry: reused 1 +--- error_log +lua tcp socket abort queueing +--- no_error_log +[error] + + + +=== TEST 81: conn queuing: killing a parent cancels its posted child coroutine +--- config + location /t { + lua_socket_log_errors off; + content_by_lua_block { + local port = ngx.var.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()) + } + } +--- request +GET /t +--- response_body +reused: 1 +extra: timeout +--- error_log +lua tcp socket abort queueing +--- no_error_log +[error] + + + +=== TEST 82: conn queuing: internal redirect cancels an unexecuted content handoff +--- config + location /t { + content_by_lua_block { + local port = ngx.var.server_port + local opts = {pool = "exec_cancel", pool_size = 1, backlog = 1} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + assert(sock:connect("127.0.0.1", port, opts)) + error("redirected waiter resumed") + end) + + assert(holder:setkeepalive()) + return ngx.exec("/after") + } + } + + location /after { + lua_socket_log_errors off; + content_by_lua_block { + collectgarbage("collect") + collectgarbage("collect") + + local opts = {pool = "exec_cancel", pool_size = 1, backlog = 1} + local port = ngx.var.server_port + 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()) + } + } +--- request +GET /t +--- response_body +reused: 1 +extra: timeout +--- error_log +lua tcp socket abort queueing +--- no_error_log +[error] + + + +=== TEST 83: conn queuing: internal redirect cancels an unexecuted rewrite handoff +--- config + location /t { + rewrite_by_lua_block { + local port = ngx.var.server_port + local opts = {pool = "rewrite_exec_cancel", pool_size = 1, + backlog = 1} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + assert(sock:connect("127.0.0.1", port, opts)) + error("redirected waiter resumed") + end) + + assert(holder:setkeepalive()) + return ngx.exec("/after") + } + + content_by_lua_block { + error("old content executed") + } + } + + location /after { + lua_socket_log_errors off; + content_by_lua_block { + collectgarbage("collect") + collectgarbage("collect") + + local opts = {pool = "rewrite_exec_cancel", pool_size = 1, + backlog = 1} + local port = ngx.var.server_port + 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()) + } + } +--- request +GET /t +--- response_body +reused: 1 +extra: timeout +--- error_log +lua tcp socket abort queueing +--- no_error_log +[error] + + + +=== TEST 84: conn queuing: internal redirect cancels an unexecuted access handoff +--- config + location /t { + access_by_lua_block { + local port = ngx.var.server_port + local opts = {pool = "access_exec_cancel", pool_size = 1, + backlog = 1} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + assert(sock:connect("127.0.0.1", port, opts)) + error("redirected waiter resumed") + end) + + assert(holder:setkeepalive()) + return ngx.exec("/after") + } + + content_by_lua_block { + error("old content executed") + } + } + + location /after { + lua_socket_log_errors off; + content_by_lua_block { + collectgarbage("collect") + collectgarbage("collect") + + local opts = {pool = "access_exec_cancel", pool_size = 1, + backlog = 1} + local port = ngx.var.server_port + 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()) + } + } +--- request +GET /t +--- response_body +reused: 1 +extra: timeout +--- error_log +lua tcp socket abort queueing +--- no_error_log +[error] + + + +=== TEST 85: conn queuing: a stale connection returned to the pool passes its slot to a waiter +--- config + set $port $TEST_NGINX_SERVER_PORT; + + location = /ka { + return 200 "ok"; + } + + location /t { + lua_socket_log_errors off; + + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +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 86: conn queuing: waiters refused by the backend pass their slots on +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + lua_socket_log_errors off; + + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +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 87: conn queuing: a request ending without closing its socket resumes a waiter +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + lua_socket_log_errors off; + + content_by_lua_block { + local port = ngx.var.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 + } + } +--- request +GET /t +--- response_body +done +--- error_log +waiter: reused 0, extra: timeout +--- no_error_log +[error] +--- wait: 0.5 + + + +=== TEST 88: conn queuing: a waiter cancelled by the end of its request frees its backlog place +--- config + set $port $TEST_NGINX_SERVER_PORT; + + location /sub { + content_by_lua_block { + local opts = {pool = "ended_waiter", pool_size = 1, backlog = 1} + ngx.thread.spawn(function() + assert(ngx.socket.connect("127.0.0.1", ngx.var.port, opts)) + error("cancelled waiter resumed") + end) + + return ngx.exit(200) + } + } + + location /t { + lua_socket_log_errors off; + + content_by_lua_block { + local port = ngx.var.port + local opts = {pool = "ended_waiter", pool_size = 1, backlog = 1} + local holder = assert(ngx.socket.connect("127.0.0.1", port, opts)) + local res = ngx.location.capture("/sub") + + -- the subrequest'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("status: ", res.status) + ngx.say(r) + } + } +--- request +GET /t +--- response_body +status: 200 +newcomer: reused 1 +--- error_log +lua tcp socket abort queueing +--- no_error_log +[error] + + + +=== TEST 89: conn queuing: a reconnect rejected as busy reading keeps the socket's pool +--- config + set $port $TEST_NGINX_SERVER_PORT; + + location = /ka { + return 200 "ok"; + } + + location /t { + lua_socket_log_errors off; + + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +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 90: conn queuing: a pre-pool connection replaces an idle connection in a full pool +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + lua_socket_log_errors off; + + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +reused: 1 +extra: timeout +--- error_log +lua tcp socket queue connect operation for connection pool "legacy_replaces_idle" +--- no_error_log +[error] + + + +=== TEST 91: conn queuing: a fast shutdown frees pools with queued connects and idle connections +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + 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 = ngx.var.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") + ngx.eof() + + -- the worker exits at once, with both connects still queued + ffi.C.kill(ffi.C.getpid(), 15) + } + } +--- request +GET /t +--- response_body +ok +--- error_log +lua tcp socket keepalive: free connection pool +Lua VM closed after the pool cleanup +--- no_error_log +[alert] +--- wait: 0.2 + + + +=== TEST 92: conn queuing: a waiter resumed with an empty slot reuses a connection cached before it runs +--- config + location /t { + set $port $TEST_NGINX_SERVER_PORT; + lua_socket_log_errors off; + + content_by_lua_block { + local port = ngx.var.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() + } + } +--- request +GET /t +--- response_body +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 0000000000..52a0033d5b --- /dev/null +++ b/t/196-socket-keepalive-queue.t @@ -0,0 +1,413 @@ +# vim:set ft= ts=4 sw=4 et fdm=marker: + +use Test::Nginx::Socket::Lua; + +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 $HttpVersion = $ENV{TEST_NGINX_USE_HTTP3} ? 3 + : $ENV{TEST_NGINX_USE_HTTP2} ? 2 : 1.1; + +our $Helpers = <<'_EOC_'; +>>> queue.lua +local port = ngx.var.server_port +local phase = ngx.get_phase() +assert(phase == ngx.var.expected_phase, phase) +assert(ngx.req.http_version() == tonumber(ngx.var.expected_version)) + + +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) + ngx.say(name, ": ", first_reused, ", ", second_reused, ", ", reused) + + else + ngx.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 + + +ngx.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)) +ngx.say("timeout retry: ", sock:getreusedtimes()) +assert(sock:close()) + +>>> stress.lua +local rounds = tonumber(ngx.var.rounds) +local port = ngx.var.server_port +local seed = 2544990 +local completed, queued, posted, saturated, gc = 0, 0, 0, 0, 0 +local all_posted_cancelled = 0 +assert(ngx.req.http_version() == tonumber(ngx.var.expected_version)) + + +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 rewrite +--- config eval + qq{ + location /t { + lua_socket_log_errors off; + set \$expected_phase rewrite; + set \$expected_version $::HttpVersion; + rewrite_by_lua_file html/queue.lua; + content_by_lua_block { } + } + } +--- request +GET /t +--- response_body +phase: rewrite +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 access +--- config eval + qq{ + location /t { + lua_socket_log_errors off; + set \$expected_phase access; + set \$expected_version $::HttpVersion; + access_by_lua_file html/queue.lua; + content_by_lua_block { } + } + } +--- request +GET /t +--- response_body +phase: access +handoff: 1, 2, 3 +cancel handoff: 1, 2 +cancel empty: 0, 1 +timeout retry: 0 +--- no_error_log +[error] +[alert] +[crit] + + + +=== TEST 3: queued connects resume, cancel and time out in content +--- config eval + qq{ + location /t { + lua_socket_log_errors off; + set \$expected_phase content; + set \$expected_version $::HttpVersion; + content_by_lua_file html/queue.lua; + } + } +--- request +GET /t +--- response_body +phase: content +handoff: 1, 2, 3 +cancel handoff: 1, 2 +cancel empty: 0, 1 +timeout retry: 0 +--- no_error_log +[error] +[alert] +[crit] + + + +=== TEST 4: repeated cancellation preserves capacity across pool lifetimes +--- config eval + qq{ + location /t { + lua_socket_log_errors off; + set \$rounds $::Rounds; + set \$expected_version $::HttpVersion; + content_by_lua_file html/stress.lua; + } + } +--- request +GET /t +--- response_body eval + "rounds: " . (3 * $::Rounds) . "\ngc: " . (9 * $::Rounds) + . "\nall waiters accounted for\n" +--- no_error_log +[error] +[alert] +[crit] +--- timeout eval: $::StressTimeout +--- quic_max_idle_timeout eval: $::StressTimeout