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