diff --git a/.github/workflows/build_and_test.yml b/.github/workflows/build_and_test.yml index 63ff4d423f..5a9ebdf6f3 100644 --- a/.github/workflows/build_and_test.yml +++ b/.github/workflows/build_and_test.yml @@ -147,7 +147,9 @@ jobs: git clone --depth=1 https://github.com/openresty/stream-lua-nginx-module.git ../stream-lua-nginx-module # Local repos — used directly from the working directory - git clone --depth=1 https://github.com/openresty/test-nginx.git + # TEMPORARY, drop once openresty/test-nginx#190 is merged: HUP fix + # (kill(1) deadlocks under the ASan job's LD_PRELOAD on Ubuntu 22.04) + git clone --depth=1 -b bugfix/hup-signal-without-kill https://github.com/oowl/test-nginx.git git clone --depth=1 https://github.com/openresty/openresty-devel-utils.git git clone --depth=1 https://github.com/openresty/mockeagain.git git clone --depth=1 https://github.com/openresty/lua-cjson.git lua-cjson diff --git a/src/ngx_http_lua_balancer.c b/src/ngx_http_lua_balancer.c index bd56d9da19..d9c17cf9b5 100644 --- a/src/ngx_http_lua_balancer.c +++ b/src/ngx_http_lua_balancer.c @@ -16,24 +16,40 @@ #include "ngx_http_lua_directive.h" #define NGX_BALANCER_DEF_HOST_LEN 32 + +/* + * the idle connections of one pool key, so that reusing, counting and + * evicting within a pool never has to scan the other cached connections + */ typedef struct { - ngx_queue_t queue; - ngx_queue_t hnode; + ngx_queue_t hnode; /* in buckets, or free_pools */ + ngx_queue_t items; /* most recently used first */ + ngx_uint_t count; ngx_uint_t hash; - ngx_http_lua_srv_conf_t *lscf; - ngx_connection_t *connection; - socklen_t socklen; - ngx_sockaddr_t sockaddr; - ngx_sockaddr_t local_sockaddr; - ngx_str_t host; +} ngx_http_lua_balancer_ka_pool_t; + + +typedef struct { + ngx_queue_t queue; + ngx_queue_t pnode; /* in pool->items */ + ngx_http_lua_balancer_ka_pool_t *pool; + ngx_uint_t hash; + ngx_http_lua_srv_conf_t *lscf; + ngx_connection_t *connection; + socklen_t socklen; + ngx_sockaddr_t sockaddr; + ngx_sockaddr_t local_sockaddr; + ngx_str_t host; /* try to avoid allocating memory from the connection pool */ - u_char host_data[NGX_BALANCER_DEF_HOST_LEN]; + u_char host_data[NGX_BALANCER_DEF_HOST_LEN]; } ngx_http_lua_balancer_ka_item_t; /*balancer keepalive item*/ struct ngx_http_lua_balancer_peer_data_s { ngx_uint_t keepalive_requests; ngx_msec_t keepalive_timeout; + /* max idle connections cached per pool, 0 means no per-pool limit */ + ngx_uint_t keepalive_pool_size; void *data; @@ -96,6 +112,17 @@ static ngx_connection_t *ngx_http_lua_balancer_get_cached_item( ngx_http_lua_srv_conf_t *lscf, ngx_peer_connection_t *pc, ngx_str_t *name); static ngx_uint_t ngx_http_lua_balancer_calc_hash(ngx_str_t *name, struct sockaddr *sockaddr, socklen_t socklen, ngx_addr_t *local); +static ngx_int_t ngx_http_lua_balancer_item_match( + ngx_http_lua_balancer_ka_item_t *item, ngx_uint_t hash, ngx_str_t *name, + struct sockaddr *sockaddr, socklen_t socklen, ngx_addr_t *local); +static ngx_http_lua_balancer_ka_pool_t *ngx_http_lua_balancer_find_pool( + ngx_http_lua_srv_conf_t *lscf, ngx_uint_t hash, ngx_str_t *name, + struct sockaddr *sockaddr, socklen_t socklen, ngx_addr_t *local); +static void ngx_http_lua_balancer_attach_item(ngx_http_lua_srv_conf_t *lscf, + ngx_http_lua_balancer_ka_item_t *item, ngx_str_t *name, + struct sockaddr *sockaddr, socklen_t socklen, ngx_addr_t *local); +static void ngx_http_lua_balancer_detach_item( + ngx_http_lua_balancer_ka_item_t *item); static struct sockaddr *ngx_http_lua_balancer_default_server_sockaddr; @@ -291,6 +318,7 @@ ngx_http_lua_balancer_init(ngx_conf_t *cf, ngx_http_upstream_srv_conf_t *us) ngx_queue_t *buckets; ngx_http_lua_srv_conf_t *lscf; ngx_http_lua_balancer_ka_item_t *cached; + ngx_http_lua_balancer_ka_pool_t *pools; lscf = ngx_http_conf_upstream_srv_conf(us, ngx_http_lua_module); @@ -321,6 +349,24 @@ ngx_http_lua_balancer_init(ngx_conf_t *cf, ngx_http_upstream_srv_conf_t *us) cached[i].lscf = lscf; } + /* + * a pool exists only while it holds idle connections, so there can + * never be more pools than cache items + */ + + pools = ngx_pcalloc(cf->pool, + sizeof(ngx_http_lua_balancer_ka_pool_t) + * lscf->balancer.max_cached); + if (pools == NULL) { + return NGX_ERROR; + } + + ngx_queue_init(&lscf->balancer.free_pools); + + for (i = 0; i < lscf->balancer.max_cached; i++) { + ngx_queue_insert_head(&lscf->balancer.free_pools, &pools[i].hnode); + } + bucket_cnt = lscf->balancer.max_cached / 2; bucket_cnt = bucket_cnt > 0 ? bucket_cnt : 1; buckets = ngx_pcalloc(cf->pool, sizeof(ngx_queue_t) * bucket_cnt); @@ -430,6 +476,7 @@ ngx_http_lua_balancer_get_peer(ngx_peer_connection_t *pc, void *data) bp->more_tries = 0; bp->keepalive_requests = 0; bp->keepalive_timeout = 0; + bp->keepalive_pool_size = 0; bp->keepalive = 0; bp->total_tries++; @@ -583,6 +630,7 @@ ngx_http_lua_balancer_free_peer(ngx_peer_connection_t *pc, void *data, ngx_connection_t *c; ngx_http_upstream_t *u; ngx_http_lua_balancer_ka_item_t *item; + ngx_http_lua_balancer_ka_pool_t *pool; ngx_http_lua_balancer_peer_data_t *bp = data; ngx_http_lua_srv_conf_t *lscf = bp->conf; @@ -633,13 +681,50 @@ ngx_http_lua_balancer_free_peer(ngx_peer_connection_t *pc, void *data, goto invalid; } - if (ngx_queue_empty(&lscf->balancer.free)) { + host = &bp->host; + hash = ngx_http_lua_balancer_calc_hash(host, + bp->sockaddr, bp->socklen, + pc->local); + + pool = NULL; + + if (bp->keepalive_pool_size) { + pool = ngx_http_lua_balancer_find_pool(lscf, hash, host, + bp->sockaddr, + bp->socklen, + pc->local); + } + + if (pool != NULL && pool->count >= bp->keepalive_pool_size) { + /* + * the pool already holds pool_size idle connections: evict + * its least recently used one instead of pushing out the + * idle connections of other pools + */ + + item = ngx_queue_data(ngx_queue_last(&pool->items), + ngx_http_lua_balancer_ka_item_t, pnode); + + ngx_log_debug4(NGX_LOG_DEBUG_HTTP, pc->log, 0, + "lua balancer: keepalive pool full, " + "evicting connection %p, pool size: %ui, " + "host: %V, name: %V", + item->connection, bp->keepalive_pool_size, + bp->addr_text, host); + + q = &item->queue; + ngx_queue_remove(q); + ngx_http_lua_balancer_detach_item(item); + + ngx_http_lua_balancer_close(item->connection); + + } else if (ngx_queue_empty(&lscf->balancer.free)) { q = ngx_queue_last(&lscf->balancer.cache); item = ngx_queue_data(q, ngx_http_lua_balancer_ka_item_t, queue); ngx_queue_remove(q); - ngx_queue_remove(&item->hnode); + ngx_http_lua_balancer_detach_item(item); ngx_http_lua_balancer_close(item->connection); @@ -651,19 +736,13 @@ ngx_http_lua_balancer_free_peer(ngx_peer_connection_t *pc, void *data, queue); } - host = &bp->host; ngx_log_debug3(NGX_LOG_DEBUG_HTTP, pc->log, 0, "lua balancer: keepalive saving connection %p, " "host: %V, name: %V", c, bp->addr_text, host); ngx_queue_insert_head(&lscf->balancer.cache, q); - hash = ngx_http_lua_balancer_calc_hash(host, - bp->sockaddr, bp->socklen, - pc->local); item->hash = hash; - hash %= lscf->balancer.bucket_cnt; - ngx_queue_insert_head(&lscf->balancer.buckets[hash], &item->hnode); item->connection = c; pc->connection = NULL; @@ -707,7 +786,6 @@ ngx_http_lua_balancer_free_peer(ngx_peer_connection_t *pc, void *data, ngx_http_lua_balancer_close(c); ngx_queue_remove(&item->queue); - ngx_queue_remove(&item->hnode); ngx_queue_insert_head(&item->lscf->balancer.free, &item->queue); return; @@ -720,6 +798,9 @@ ngx_http_lua_balancer_free_peer(ngx_peer_connection_t *pc, void *data, ngx_str_null(&item->host); } + ngx_http_lua_balancer_attach_item(lscf, item, host, bp->sockaddr, + bp->socklen, pc->local); + if (c->read->ready) { ngx_http_lua_balancer_close_handler(c->read); } @@ -820,7 +901,7 @@ ngx_http_lua_balancer_close_handler(ngx_event_t *ev) ngx_http_lua_balancer_close(c); ngx_queue_remove(&item->queue); - ngx_queue_remove(&item->hnode); + ngx_http_lua_balancer_detach_item(item); ngx_queue_insert_head(&item->lscf->balancer.free, &item->queue); } @@ -1024,7 +1105,8 @@ ngx_http_lua_ffi_balancer_bind_to_local_addr(ngx_http_request_t *r, int ngx_http_lua_ffi_balancer_enable_keepalive(ngx_http_request_t *r, - unsigned long timeout, unsigned int max_requests, char **err) + unsigned long timeout, unsigned int max_requests, unsigned int pool_size, + char **err) { ngx_http_upstream_t *u; ngx_http_lua_ctx_t *ctx; @@ -1062,6 +1144,7 @@ ngx_http_lua_ffi_balancer_enable_keepalive(ngx_http_request_t *r, bp->keepalive_timeout = (ngx_msec_t) timeout; bp->keepalive_requests = (ngx_uint_t) max_requests; + bp->keepalive_pool_size = (ngx_uint_t) pool_size; bp->keepalive = 1; return NGX_OK; @@ -1456,64 +1539,153 @@ ngx_http_lua_balancer_get_cached_item(ngx_http_lua_srv_conf_t *lscf, ngx_peer_connection_t *pc, ngx_str_t *name) { ngx_uint_t hash; - ngx_queue_t *q; - ngx_queue_t *head; ngx_connection_t *c; - struct sockaddr *sockaddr; - socklen_t socklen; - ngx_addr_t *local; ngx_http_lua_balancer_ka_item_t *item; + ngx_http_lua_balancer_ka_pool_t *pool; + + hash = ngx_http_lua_balancer_calc_hash(name, pc->sockaddr, pc->socklen, + pc->local); + + pool = ngx_http_lua_balancer_find_pool(lscf, hash, name, pc->sockaddr, + pc->socklen, pc->local); + if (pool == NULL) { + return NULL; + } - sockaddr = pc->sockaddr; - socklen = pc->socklen; - local = pc->local; + item = ngx_queue_data(ngx_queue_head(&pool->items), + ngx_http_lua_balancer_ka_item_t, pnode); + + c = item->connection; + ngx_http_lua_balancer_detach_item(item); + ngx_queue_remove(&item->queue); + ngx_queue_insert_head(&lscf->balancer.free, &item->queue); + c->idle = 0; + c->sent = 0; + c->log = pc->log; + c->read->log = pc->log; + c->write->log = pc->log; + c->pool->log = pc->log; + + if (c->read->timer_set) { + ngx_del_timer(c->read); + } + + pc->cached = 1; + pc->connection = c; + return c; +} + + +static ngx_int_t +ngx_http_lua_balancer_item_match(ngx_http_lua_balancer_ka_item_t *item, + ngx_uint_t hash, ngx_str_t *name, struct sockaddr *sockaddr, + socklen_t socklen, ngx_addr_t *local) +{ + if (item->hash != hash) { + return 0; + } + + return name->len == item->host.len + && ngx_memn2cmp((u_char *) &item->sockaddr, + (u_char *) sockaddr, + item->socklen, socklen) == 0 + && ngx_strncasecmp(name->data, + item->host.data, name->len) == 0 + && (local == NULL + || ngx_memn2cmp((u_char *) &item->local_sockaddr, + (u_char *) local->sockaddr, + socklen, local->socklen) == 0); +} + + +static ngx_http_lua_balancer_ka_pool_t * +ngx_http_lua_balancer_find_pool(ngx_http_lua_srv_conf_t *lscf, + ngx_uint_t hash, ngx_str_t *name, struct sockaddr *sockaddr, + socklen_t socklen, ngx_addr_t *local) +{ + ngx_queue_t *q; + ngx_queue_t *head; + ngx_http_lua_balancer_ka_item_t *item; + ngx_http_lua_balancer_ka_pool_t *pool; - hash = ngx_http_lua_balancer_calc_hash(name, sockaddr, socklen, pc->local); head = &lscf->balancer.buckets[hash % lscf->balancer.bucket_cnt]; - c = NULL; for (q = ngx_queue_head(head); - q != ngx_queue_sentinel(head); - q = ngx_queue_next(q)) + q != ngx_queue_sentinel(head); + q = ngx_queue_next(q)) { - item = ngx_queue_data(q, ngx_http_lua_balancer_ka_item_t, hnode); - if (item->hash != hash) { + pool = ngx_queue_data(q, ngx_http_lua_balancer_ka_pool_t, hnode); + + if (pool->hash != hash) { continue; } - if (name->len == item->host.len - && ngx_memn2cmp((u_char *) &item->sockaddr, - (u_char *) sockaddr, - item->socklen, socklen) == 0 - && ngx_strncasecmp(name->data, - item->host.data, name->len) == 0 - && (local == NULL - || ngx_memn2cmp((u_char *) &item->local_sockaddr, - (u_char *) local->sockaddr, - socklen, local->socklen) == 0)) - { - c = item->connection; - ngx_queue_remove(q); - ngx_queue_remove(&item->queue); - ngx_queue_insert_head(&lscf->balancer.free, &item->queue); - c->idle = 0; - c->sent = 0; - c->log = pc->log; - c->read->log = pc->log; - c->write->log = pc->log; - c->pool->log = pc->log; - - if (c->read->timer_set) { - ngx_del_timer(c->read); - } + /* a pool is never empty, all its items share the same key */ + + item = ngx_queue_data(ngx_queue_head(&pool->items), + ngx_http_lua_balancer_ka_item_t, pnode); - pc->cached = 1; - pc->connection = c; - return c; + if (ngx_http_lua_balancer_item_match(item, hash, name, sockaddr, + socklen, local)) + { + return pool; } } return NULL; } + +static void +ngx_http_lua_balancer_attach_item(ngx_http_lua_srv_conf_t *lscf, + ngx_http_lua_balancer_ka_item_t *item, ngx_str_t *name, + struct sockaddr *sockaddr, socklen_t socklen, ngx_addr_t *local) +{ + ngx_queue_t *q; + ngx_queue_t *head; + ngx_http_lua_balancer_ka_pool_t *pool; + + pool = ngx_http_lua_balancer_find_pool(lscf, item->hash, name, sockaddr, + socklen, local); + + if (pool == NULL) { + /* item is not attached yet, so a free pool is always left */ + + ngx_http_lua_assert(!ngx_queue_empty(&lscf->balancer.free_pools)); + + q = ngx_queue_head(&lscf->balancer.free_pools); + ngx_queue_remove(q); + + pool = ngx_queue_data(q, ngx_http_lua_balancer_ka_pool_t, hnode); + pool->hash = item->hash; + pool->count = 0; + ngx_queue_init(&pool->items); + + head = &lscf->balancer.buckets[item->hash % lscf->balancer.bucket_cnt]; + ngx_queue_insert_head(head, &pool->hnode); + } + + ngx_queue_insert_head(&pool->items, &item->pnode); + pool->count++; + item->pool = pool; +} + + +static void +ngx_http_lua_balancer_detach_item(ngx_http_lua_balancer_ka_item_t *item) +{ + ngx_http_lua_balancer_ka_pool_t *pool; + + pool = item->pool; + + ngx_queue_remove(&item->pnode); + item->pool = NULL; + + if (--pool->count == 0) { + ngx_queue_remove(&pool->hnode); + ngx_queue_insert_head(&item->lscf->balancer.free_pools, + &pool->hnode); + } +} + /* vi:set ft=c ts=4 sw=4 et fdm=marker: */ diff --git a/src/ngx_http_lua_common.h b/src/ngx_http_lua_common.h index 28538cc224..03940ec794 100644 --- a/src/ngx_http_lua_common.h +++ b/src/ngx_http_lua_common.h @@ -386,6 +386,7 @@ struct ngx_http_lua_srv_conf_s { ngx_uint_t max_cached; ngx_queue_t cache; ngx_queue_t free; + ngx_queue_t free_pools; ngx_queue_t *buckets; ngx_uint_t bucket_cnt; ngx_http_upstream_init_pt original_init_upstream; diff --git a/t/195-balancer-keepalive-pool-size.t b/t/195-balancer-keepalive-pool-size.t new file mode 100644 index 0000000000..535f74b67b --- /dev/null +++ b/t/195-balancer-keepalive-pool-size.t @@ -0,0 +1,688 @@ +# vim:set ft= ts=4 sw=4 et: + +use Test::Nginx::Socket::Lua; + +log_level('debug'); +repeat_each(1); + +plan tests => repeat_each() * (blocks() * 3 + 13); + +no_long_string(); + +our $BalancerHelper = <<'_EOC_'; + init_by_lua_block { + local ffi = require "ffi" + + -- declares the symbol, with whatever prototype the lua-resty-core + -- version in use has + require "ngx.balancer" + + -- call the C function through an explicit prototype, so that the + -- tests do not depend on the FFI declaration of the lua-resty-core + -- version in use + local enable_keepalive = ffi.cast( + "int (*)(void *r, unsigned long timeout, unsigned int max_requests," + .. " unsigned int pool_size, char **err)", + ffi.C.ngx_http_lua_ffi_balancer_enable_keepalive) + + local errmsg = ffi.new("char *[1]") + + -- timeout is in milliseconds, as the C function expects it + function enable_keepalive_pool(pool_size, timeout, max_requests) + local r = require("resty.core.base").get_request() + local rc = enable_keepalive(r, timeout or 60000, + max_requests or 100, pool_size, + errmsg) + if rc ~= 0 then + return nil, ffi.string(errmsg[0]) + end + + return true + end + } +_EOC_ + +our $Backend = <<'_EOC_'; + server { + listen 127.0.0.1:$TEST_NGINX_RAND_PORT_1; + listen 127.0.0.1:$TEST_NGINX_RAND_PORT_2; + + keepalive_requests 1000; + + location = /conn { + content_by_lua_block { + -- the peer address shows which local address was bound + ngx.print(ngx.var.connection, "|", ngx.var.remote_addr) + } + } + + location = /conn_short { + keepalive_timeout 100ms; + + content_by_lua_block { + ngx.print(ngx.var.connection) + } + } + } +_EOC_ + +our $ProxyLoc = <<'_EOC_'; + location = /proxy { + internal; + proxy_http_version 1.1; + proxy_set_header Connection ""; + proxy_pass http://backend/conn; + } + + location = /proxy_short { + internal; + proxy_http_version 1.1; + proxy_set_header Connection ""; + proxy_pass http://backend/conn_short; + } +_EOC_ + +add_block_preprocessor(sub { + my $block = shift; + + if (defined $block->http_upstream) { + $block->set_value("http_config", $BalancerHelper + . $block->http_upstream . $Backend); + } + + if (defined $block->config) { + $block->set_value("config", $block->config . $ProxyLoc); + } + + if (!defined $block->request) { + $block->set_value("request", "GET /t"); + } + + if (!defined $block->no_error_log) { + $block->set_value("no_error_log", "[error]"); + } +}); + +run_tests(); + +__DATA__ + +=== TEST 1: pool_size caps the idle connections of a pool +--- http_upstream + upstream backend { + server 0.0.0.1; + keepalive 0; + + balancer_by_lua_block { + local b = require "ngx.balancer" + assert(b.set_current_peer("127.0.0.1", $TEST_NGINX_RAND_PORT_1, + "one")) + assert(enable_keepalive_pool(1)) + } + } +--- config + location = /t { + content_by_lua_block { + local function conns(n) + local reqs = {} + for i = 1, n do + reqs[i] = { "/proxy" } + end + + local ids = {} + for i, res in ipairs({ ngx.location.capture_multi(reqs) }) do + assert(res.status == 200, "bad status " .. res.status) + ids[i] = res.body + end + + return ids + end + + -- 3 concurrent requests open 3 connections, but only one of + -- them may stay idle in the pool once they are all released + local first = {} + for _, id in ipairs(conns(3)) do + first[id] = true + end + + local reused = 0 + for _, id in ipairs(conns(2)) do + if first[id] then + reused = reused + 1 + end + end + + ngx.say("reused: ", reused) + } + } +--- response_body +reused: 1 +--- error_log +lua balancer: keepalive pool full, evicting connection + + + +=== TEST 2: a full pool does not evict the idle connections of other pools +--- http_upstream + upstream backend { + server 0.0.0.1; + keepalive 0; + balancer_keepalive 2; + + balancer_by_lua_block { + local b = require "ngx.balancer" + assert(b.set_current_peer("127.0.0.1", $TEST_NGINX_RAND_PORT_1, + ngx.var.arg_pool)) + assert(enable_keepalive_pool(1)) + } + } +--- config + location = /t { + content_by_lua_block { + local b1 = ngx.location.capture("/proxy", { args = "pool=b" }).body + + -- two concurrent connections for pool "a": the second one to be + -- released must evict the first one, not the idle one of "b" + ngx.location.capture_multi({ + { "/proxy", { args = "pool=a" } }, + { "/proxy", { args = "pool=a" } }, + }) + + local b2 = ngx.location.capture("/proxy", { args = "pool=b" }).body + + ngx.say("pool b reused: ", b1 == b2) + } + } +--- response_body +pool b reused: true + + + +=== TEST 3: no pool_size: a busy pool evicts idle connections of other pools +--- http_upstream + upstream backend { + server 0.0.0.1; + keepalive 0; + balancer_keepalive 2; + + balancer_by_lua_block { + local b = require "ngx.balancer" + assert(b.set_current_peer("127.0.0.1", $TEST_NGINX_RAND_PORT_1, + ngx.var.arg_pool)) + assert(enable_keepalive_pool(0)) + } + } +--- config + location = /t { + content_by_lua_block { + local b1 = ngx.location.capture("/proxy", { args = "pool=b" }).body + + ngx.location.capture_multi({ + { "/proxy", { args = "pool=a" } }, + { "/proxy", { args = "pool=a" } }, + }) + + local b2 = ngx.location.capture("/proxy", { args = "pool=b" }).body + + ngx.say("pool b reused: ", b1 == b2) + } + } +--- response_body +pool b reused: false + + + +=== TEST 4: error paths still report through the err argument +--- http_upstream + upstream backend { + server 0.0.0.1; + keepalive 0; + + balancer_by_lua_block { + -- no current peer set yet + local ok, err = enable_keepalive_pool(1) + ngx.log(ngx.WARN, "enable keepalive: ", ok, " ", err) + + local b = require "ngx.balancer" + assert(b.set_current_peer("127.0.0.1", $TEST_NGINX_RAND_PORT_1)) + } + } +--- config + location = /t { + content_by_lua_block { + local res = ngx.location.capture("/proxy") + ngx.say("status: ", res.status) + } + } +--- response_body +status: 200 +--- error_log +enable keepalive: nil no current peer set + + + +=== TEST 5: many pools churning through a small cache +--- http_upstream + upstream backend { + server 0.0.0.1; + keepalive 0; + balancer_keepalive 4; + + balancer_by_lua_block { + local b = require "ngx.balancer" + assert(b.set_current_peer("127.0.0.1", $TEST_NGINX_RAND_PORT_1, + ngx.var.arg_pool)) + assert(enable_keepalive_pool(2)) + } + } +--- config + location = /t { + content_by_lua_block { + local bad = 0 + + for round = 1, 3 do + for p = 1, 8 do + local args = "pool=p" .. p + local res = { ngx.location.capture_multi({ + { "/proxy", { args = args } }, + { "/proxy", { args = args } }, + { "/proxy", { args = args } }, + }) } + + res[4] = ngx.location.capture("/proxy", { args = args }) + + for _, r in ipairs(res) do + if r.status ~= 200 then + bad = bad + 1 + end + end + end + end + + ngx.say("failed requests: ", bad) + } + } +--- response_body +failed requests: 0 +--- no_error_log +[error] +[alert] +[crit] + + + +=== TEST 6: an idle connection closed by the peer leaves its pool +--- http_upstream + upstream backend { + server 0.0.0.1; + keepalive 0; + + balancer_by_lua_block { + local b = require "ngx.balancer" + assert(b.set_current_peer("127.0.0.1", $TEST_NGINX_RAND_PORT_1, + "one")) + assert(enable_keepalive_pool(2)) + } + } +--- config + location = /t { + content_by_lua_block { + local r1 = ngx.location.capture("/proxy_short") + + -- the backend closes the idle connection after 100ms + ngx.sleep(0.3) + + local r2 = ngx.location.capture("/proxy_short") + local r3 = ngx.location.capture("/proxy_short") + + ngx.say("status: ", r1.status, " ", r2.status, " ", r3.status) + ngx.say("closed one reused: ", r1.body == r2.body) + ngx.say("new one reused: ", r2.body == r3.body) + } + } +--- response_body +status: 200 200 200 +closed one reused: false +new one reused: true +--- error_log +lua balancer: keepalive closing connection +--- no_error_log +[error] +[alert] + + + +=== TEST 7: a pool holding exactly pool_size connections evicts nothing +--- http_upstream + upstream backend { + server 0.0.0.1; + keepalive 0; + + balancer_by_lua_block { + local b = require "ngx.balancer" + assert(b.set_current_peer("127.0.0.1", $TEST_NGINX_RAND_PORT_1, + "one")) + assert(enable_keepalive_pool(3)) + } + } +--- config + location = /t { + content_by_lua_block { + local function conns(n) + local reqs = {} + for i = 1, n do + reqs[i] = { "/proxy" } + end + + local ids = {} + for i, res in ipairs({ ngx.location.capture_multi(reqs) }) do + assert(res.status == 200, "bad status " .. res.status) + ids[i] = res.body + end + + return ids + end + + local first = {} + for _, id in ipairs(conns(3)) do + first[id] = true + end + + local reused = 0 + for _, id in ipairs(conns(3)) do + if first[id] then + reused = reused + 1 + end + end + + ngx.say("reused: ", reused) + } + } +--- response_body +reused: 3 +--- no_error_log +[error] +lua balancer: keepalive pool full + + + +=== TEST 8: pool_size larger than balancer_keepalive never raises the total +--- http_upstream + upstream backend { + server 0.0.0.1; + keepalive 0; + balancer_keepalive 2; + + balancer_by_lua_block { + local b = require "ngx.balancer" + assert(b.set_current_peer("127.0.0.1", $TEST_NGINX_RAND_PORT_1, + "one")) + assert(enable_keepalive_pool(5)) + } + } +--- config + location = /t { + content_by_lua_block { + local function conns(n) + local reqs = {} + for i = 1, n do + reqs[i] = { "/proxy" } + end + + local ids = {} + for i, res in ipairs({ ngx.location.capture_multi(reqs) }) do + assert(res.status == 200, "bad status " .. res.status) + ids[i] = res.body + end + + return ids + end + + local first = {} + for _, id in ipairs(conns(3)) do + first[id] = true + end + + local reused = 0 + for _, id in ipairs(conns(3)) do + if first[id] then + reused = reused + 1 + end + end + + ngx.say("reused: ", reused) + } + } +--- response_body +reused: 2 +--- no_error_log +[error] +lua balancer: keepalive pool full + + + +=== TEST 9: the host and the peer address both split pools +--- http_upstream + upstream backend { + server 0.0.0.1; + keepalive 0; + + balancer_by_lua_block { + local b = require "ngx.balancer" + assert(b.set_current_peer("127.0.0.1", tonumber(ngx.var.arg_port), + ngx.var.arg_host)) + assert(enable_keepalive_pool(1)) + } + } +--- config + location = /t { + content_by_lua_block { + local keys = { + "port=$TEST_NGINX_RAND_PORT_1&host=a", + "port=$TEST_NGINX_RAND_PORT_1&host=b", + "port=$TEST_NGINX_RAND_PORT_2&host=a", + } + + -- concurrent first: every key opens its own connection + local reqs = {} + for i, args in ipairs(keys) do + reqs[i] = { "/proxy", { args = args } } + end + + local first = {} + for i, res in ipairs({ ngx.location.capture_multi(reqs) }) do + assert(res.status == 200, "bad status " .. res.status) + first[i] = res.body + end + + -- with pool_size 1, two keys sharing a pool would keep only one + -- of their two connections, so one key would get the other's + local second = {} + for i, args in ipairs(keys) do + local res = ngx.location.capture("/proxy", { args = args }) + assert(res.status == 200, "bad status " .. res.status) + second[i] = res.body + end + + for i = 1, #keys do + ngx.say("key ", i, " reused its own: ", first[i] == second[i]) + end + } + } +--- response_body +key 1 reused its own: true +key 2 reused its own: true +key 3 reused its own: true +--- no_error_log +[error] + + + +=== TEST 10: the bound local address splits pools +--- http_upstream + upstream backend { + server 0.0.0.1; + keepalive 0; + + balancer_by_lua_block { + local b = require "ngx.balancer" + assert(b.set_current_peer("127.0.0.1", $TEST_NGINX_RAND_PORT_1, + "one")) + + if ngx.var.arg_local then + assert(b.bind_to_local_addr(ngx.var.arg_local)) + end + + assert(enable_keepalive_pool(1)) + } + } +--- config + location = /t { + content_by_lua_block { + local function get(args) + local res = ngx.location.capture("/proxy", { args = args }) + assert(res.status == 200, "bad status " .. res.status) + return res.body + end + + local a1 = get("") + local l1 = get("local=127.0.0.2") + local a2 = get("") + local l2 = get("local=127.0.0.2") + + ngx.say("unbound reused its own: ", a1 == a2) + ngx.say("bound reused its own: ", l1 == l2) + ngx.say("bound from: ", l1:match("|(.+)$")) + } + } +--- response_body +unbound reused its own: true +bound reused its own: true +bound from: 127.0.0.2 +--- no_error_log +[error] + + + +=== TEST 11: an idle connection expiring on its idle timeout leaves its pool +--- http_upstream + upstream backend { + server 0.0.0.1; + keepalive 0; + + balancer_by_lua_block { + local b = require "ngx.balancer" + assert(b.set_current_peer("127.0.0.1", $TEST_NGINX_RAND_PORT_1, + "one")) + -- 100ms idle timeout, the backend keeps the connection open + assert(enable_keepalive_pool(2, 100)) + } + } +--- config + location = /t { + content_by_lua_block { + local r1 = ngx.location.capture("/proxy") + + ngx.sleep(0.3) + + local r2 = ngx.location.capture("/proxy") + local r3 = ngx.location.capture("/proxy") + + ngx.say("status: ", r1.status, " ", r2.status, " ", r3.status) + ngx.say("expired one reused: ", r1.body == r2.body) + ngx.say("new one reused: ", r2.body == r3.body) + } + } +--- response_body +status: 200 200 200 +expired one reused: false +new one reused: true +--- error_log +lua balancer: keepalive closing connection +--- no_error_log +[error] +[alert] + + + +=== TEST 12: a connection reaching max_requests is not cached +--- http_upstream + upstream backend { + server 0.0.0.1; + keepalive 0; + + balancer_by_lua_block { + local b = require "ngx.balancer" + assert(b.set_current_peer("127.0.0.1", $TEST_NGINX_RAND_PORT_1, + "one")) + assert(enable_keepalive_pool(2, nil, 2)) + } + } +--- config + location = /t { + content_by_lua_block { + local r1 = ngx.location.capture("/proxy") + local r2 = ngx.location.capture("/proxy") + local r3 = ngx.location.capture("/proxy") + local r4 = ngx.location.capture("/proxy") + + ngx.say("status: ", r1.status, " ", r2.status, " ", r3.status, + " ", r4.status) + ngx.say("2nd reused the 1st: ", r1.body == r2.body) + ngx.say("3rd reused the 2nd: ", r2.body == r3.body) + ngx.say("4th reused the 3rd: ", r3.body == r4.body) + } + } +--- response_body +status: 200 200 200 200 +2nd reused the 1st: true +3rd reused the 2nd: false +4th reused the 3rd: true +--- no_error_log +[error] +[alert] + + + +=== TEST 13: a failed try is not cached, the retry's connection is +--- http_upstream + upstream backend { + server 0.0.0.1; + keepalive 0; + + balancer_by_lua_block { + local b = require "ngx.balancer" + + if not ngx.ctx.tried then + ngx.ctx.tried = true + assert(b.set_more_tries(1)) + -- nothing listens on this port + assert(b.set_current_peer("127.0.0.1", + $TEST_NGINX_RAND_PORT_3, "one")) + + else + assert(b.set_current_peer("127.0.0.1", + $TEST_NGINX_RAND_PORT_1, "one")) + end + + assert(enable_keepalive_pool(1)) + } + } +--- config + location = /t { + content_by_lua_block { + local r1 = ngx.location.capture("/proxy") + local r2 = ngx.location.capture("/proxy") + + ngx.say("status: ", r1.status, " ", r2.status) + ngx.say("retry connection reused: ", r1.body == r2.body) + } + } +--- response_body +status: 200 200 +retry connection reused: true +--- error_log +Connection refused +--- no_error_log +[alert] +[crit]