From d337206ed52c8b2cdefabb04ef8ec54bc94cf6f9 Mon Sep 17 00:00:00 2001 From: Jun Ouyang Date: Thu, 22 Oct 2026 18:40:40 +0800 Subject: [PATCH] feature: added the pool_size argument to the balancer keepalive FFI API. The balancer keepalive cache has a single capacity per upstream {} block, set by balancer_keepalive (32 by default), and evicts the least recently used idle connection regardless of which pool it belongs to. When many pools share one upstream, as is typical when peers are picked dynamically in balancer_by_lua*, a burst to one peer can push out the idle connections of every other peer. ngx_http_lua_ffi_balancer_enable_keepalive() now takes a pool_size argument. A pool is the same key the cache already matches on for reuse: the peer address, the host given to set_current_peer() and the local address. When a connection is about to be cached and its pool already holds pool_size idle connections, the least recently used idle connection of that same pool is closed and its slot reused, instead of taking a slot from the global LRU. pool_size 0 keeps the previous behaviour, and pool_size never raises the balancer_keepalive total. To keep this O(1), the hash buckets now hold one descriptor per pool instead of the cached connections themselves. A descriptor keeps the pool's idle connections (most recently used first) and their count, so reusing a connection, checking whether a pool is full and evicting its oldest connection no longer scan the other cached connections. Scanning the bucket costs about 10ns per idle connection of the pool, i.e. ~5us per released connection with a full pool of 512. Descriptors are preallocated next to the cache items, one per item, as a pool only exists while it holds idle connections. The key comparison used by the reuse path is factored out into ngx_http_lua_balancer_item_match(). This changes the FFI signature, lua-resty-core needs to be updated accordingly. --- src/ngx_http_lua_balancer.c | 296 +++++++++--- src/ngx_http_lua_common.h | 1 + t/195-balancer-keepalive-pool-size.t | 688 +++++++++++++++++++++++++++ 3 files changed, 923 insertions(+), 62 deletions(-) create mode 100644 t/195-balancer-keepalive-pool-size.t 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]