bugfix: queued connects were not resumed when several pool slots were freed in the same cycle.

src/ngx_stream_lua_socket_tcp.c

@@ -607,6 +607,7 @@ ngx_stream_lua_socket_tcp_create_socket_pool(lua_State *L,
 
     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->cache);
     ngx_queue_init(&sp->free);
 
@@ -749,6 +750,12 @@ ngx_stream_lua_socket_tcp_connect_helper(lua_State *L,
                 ngx_queue_insert_tail(&spool->wait_connect_op,
                                       &conn_op_ctx->queue);
 
+                if (spool->next_connect_op
+                    == ngx_queue_sentinel(&spool->wait_connect_op))
+                {
+                    spool->next_connect_op = &conn_op_ctx->queue;
+                }
+
                 ngx_log_debug3(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0,
                                "stream lua tcp socket queued connect "
                                "operation for %d(ms), u: %p, ctx: %p",
@@ -4854,9 +4861,14 @@ ngx_stream_lua_socket_tcp_conn_op_timeout_handler(ngx_event_t *ev)
     ngx_stream_lua_socket_tcp_conn_op_ctx_t         *conn_op_ctx;
 
     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 = conn_op_ctx->u;
     r = u->request;
 
     coctx = u->write_co_ctx;
@@ -4942,11 +4954,14 @@ ngx_stream_lua_socket_tcp_resume_conn_op(
      * operation resumption, so that there is no resumption happens when Nginx
      * is exiting.
      */
-    if (ngx_queue_empty(&spool->wait_connect_op)) {
+    q = spool->next_connect_op;
+    if (q == ngx_queue_sentinel(&spool->wait_connect_op)) {
         return;
     }
 
-    q = ngx_queue_head(&spool->wait_connect_op);
+    /* advance before posting so each slot resumes a distinct waiter */
+    spool->next_connect_op = ngx_queue_next(q);
+
     conn_op_ctx = ngx_queue_data(q,
                                  ngx_stream_lua_socket_tcp_conn_op_ctx_t,
                                  queue);
@@ -4988,7 +5003,6 @@ ngx_stream_lua_socket_tcp_conn_op_ctx_cleanup(void *data)
 static void
 ngx_stream_lua_socket_tcp_conn_op_resume_handler(ngx_event_t *ev)
 {
-    ngx_queue_t                                     *q;
     ngx_stream_lua_request_t                        *r;
     ngx_stream_lua_cleanup_t                        *cln;
     ngx_stream_lua_ctx_t                            *ctx;
@@ -5002,27 +5016,7 @@ ngx_stream_lua_socket_tcp_conn_op_resume_handler(ngx_event_t *ev)
     r = u->request;
     spool = u->socket_pool;
 
-    if (ngx_queue_empty(&spool->wait_connect_op)) {
-#if (NGX_DEBUG)
-        ngx_stream_lua_assert(!(spool->backlog >= 0
-                              && spool->connections > spool->size));
-
-#else
-        if (spool->backlog >= 0 && spool->connections > spool->size) {
-            ngx_log_error(NGX_LOG_ERR, ngx_cycle->log, 0,
-                          "stream lua tcp socket connections count "
-                          "mismatched for connection pool \"%s\", connections: "
-                          "%i, size: %i",
-                          spool->key, spool->connections, spool->size);
-            spool->connections = spool->size;
-        }
-#endif
-
-        return;
-    }
-
-    q = ngx_queue_head(&spool->wait_connect_op);
-    ngx_queue_remove(q);
+    ngx_queue_remove(&conn_op_ctx->queue);
 
     coctx = u->write_co_ctx;
     coctx->cleanup = NULL;
@@ -6522,6 +6516,8 @@ ngx_stream_lua_socket_shutdown_pool_helper(
     ngx_stream_lua_socket_pool_item_t               *item;
     ngx_stream_lua_socket_tcp_conn_op_ctx_t         *conn_op_ctx;
 
+    spool->next_connect_op = ngx_queue_sentinel(&spool->wait_connect_op);
+
     while (!ngx_queue_empty(&spool->cache)) {
         q = ngx_queue_head(&spool->cache);
 
@@ -6552,6 +6548,10 @@ ngx_stream_lua_socket_shutdown_pool_helper(
             ngx_del_timer(&conn_op_ctx->event);
         }
 
+        if (conn_op_ctx->event.posted) {
+            ngx_delete_posted_event(&conn_op_ctx->event);
+        }
+
         ngx_stream_lua_socket_tcp_free_conn_op_ctx(conn_op_ctx);
     }
 
@@ -6963,6 +6963,10 @@ ngx_stream_lua_tcp_queue_conn_op_cleanup(void *data)
         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);

src/ngx_stream_lua_socket_tcp.h

@@ -87,6 +87,8 @@ typedef struct {
 
     ngx_int_t                          backlog;
 
+    ngx_queue_t                       *next_connect_op;
+
     u_char                             key[1];
 
 } ngx_stream_lua_socket_pool_t;

t/068-socket-keepalive.t

@@ -2492,3 +2492,257 @@ qr/\Qbad argument #2 to 'setkeepalive' (number expected, got string)\E/
 --- no_error_log 
 [crit]
 --- timeout: 4
+
+
+
+=== TEST 55: conn queuing: slots freed in the same cycle each resume a queued connect
+--- stream_server_config
+    content_by_lua_block {
+        local port = $TEST_NGINX_MEMCACHED_PORT
+        local opts = {pool = "same_cycle", 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))
+
+        local function waiter(name)
+            return ngx.thread.spawn(function()
+                local sock = ngx.socket.tcp()
+                sock:settimeouts(500, 500, 500)
+
+                local ok, err = sock:connect("127.0.0.1", port, opts)
+                if not ok then
+                    return name .. ": " .. err
+                end
+
+                -- keep the slot past the other waiter's connect timeout,
+                -- so only a resumption can let it connect
+                ngx.sleep(1)
+                sock:close()
+                return name .. ": connected"
+            end)
+        end
+
+        local w1 = waiter("w1")
+        local w2 = waiter("w2")
+
+        -- release both connections in the same cycle
+        assert(sock1:setkeepalive())
+        assert(sock2:setkeepalive())
+
+        local _, res1 = ngx.thread.wait(w1)
+        local _, res2 = ngx.thread.wait(w2)
+        ngx.say(res1)
+        ngx.say(res2)
+    }
+--- stream_response
+w1: connected
+w2: connected
+--- error_log
+stream lua tcp socket queue connect operation for connection pool "same_cycle"
+--- no_error_log
+[error]
+--- timeout: 4
+
+
+
+=== TEST 56: conn queuing: close multiple connections before resumptions run
+--- stream_server_config
+    content_by_lua_block {
+        local semaphore = require "ngx.semaphore"
+        local connected = semaphore.new()
+        local release = semaphore.new()
+        local port = $TEST_NGINX_SERVER_PORT
+        local opts = {pool = "same_cycle_close", 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))
+
+        local function waiter(name)
+            local sock = ngx.socket.tcp()
+            sock:settimeout(500)
+            assert(sock:connect("127.0.0.1", port, opts))
+            local reused = sock:getreusedtimes()
+            connected:post()
+            assert(release:wait(2))
+            sock:close()
+            return name .. ": reused " .. reused
+        end
+
+        local w1 = ngx.thread.spawn(waiter, "w1")
+        local w2 = ngx.thread.spawn(waiter, "w2")
+
+        assert(sock1:close())
+        assert(sock2:close())
+
+        -- neither waiter may free its slot before both have connected
+        assert(connected:wait(2))
+        assert(connected:wait(2))
+        release:post(2)
+
+        local _, res1 = ngx.thread.wait(w1)
+        local _, res2 = ngx.thread.wait(w2)
+        ngx.say(res1)
+        ngx.say(res2)
+    }
+--- stream_response
+w1: reused 0
+w2: reused 0
+--- error_log
+stream lua tcp socket queue connect operation for connection pool "same_cycle_close"
+--- no_error_log
+[error]
+--- timeout: 4
+
+
+
+=== TEST 57: conn queuing: cancel a posted resumption and resume the next waiter
+--- stream_server_config
+    content_by_lua_block {
+        local semaphore = require "ngx.semaphore"
+        local connected = semaphore.new()
+        local release = semaphore.new()
+        local port = $TEST_NGINX_SERVER_PORT
+        local opts = {pool = "cancel_posted", pool_size = 2, backlog = 3}
+        local order = {}
+
+        local sock1 = assert(ngx.socket.connect("127.0.0.1", port, opts))
+        local sock2 = assert(ngx.socket.connect("127.0.0.1", port, opts))
+
+        local function waiter(name)
+            local sock = ngx.socket.tcp()
+            sock:settimeout(500)
+            assert(sock:connect("127.0.0.1", port, opts))
+            order[#order + 1] = name
+            local reused = sock:getreusedtimes()
+            connected:post()
+            assert(release:wait(2))
+            sock:close()
+            return name .. ": reused " .. reused
+        end
+
+        local w1 = ngx.thread.spawn(waiter, "w1")
+        local w2 = ngx.thread.spawn(waiter, "w2")
+        local w3 = ngx.thread.spawn(waiter, "w3")
+
+        assert(sock1:setkeepalive())
+        assert(sock2:setkeepalive())
+        assert(ngx.thread.kill(w1))
+
+        assert(connected:wait(2))
+        assert(connected:wait(2))
+        release:post(2)
+
+        local _, res2 = ngx.thread.wait(w2)
+        local _, res3 = ngx.thread.wait(w3)
+        ngx.say(res2)
+        ngx.say(res3)
+        ngx.say("order: ", table.concat(order, ", "))
+    }
+--- stream_response
+w2: reused 1
+w3: reused 1
+order: w2, w3
+--- error_log
+stream lua tcp socket abort queueing
+--- no_error_log
+[error]
+--- timeout: 4
+
+
+
+=== TEST 58: conn queuing: advance past a timed out waiter
+--- stream_server_config
+    lua_socket_log_errors off;
+
+    content_by_lua_block {
+        local semaphore = require "ngx.semaphore"
+        local expired = semaphore.new()
+        local port = $TEST_NGINX_SERVER_PORT
+        local opts = {pool = "expired_head", pool_size = 1, backlog = 2}
+        local sock = assert(ngx.socket.connect("127.0.0.1", port, opts))
+
+        local function waiter1()
+            local s = ngx.socket.tcp()
+            s:settimeout(10)
+            local ok, err = s:connect("127.0.0.1", port, opts)
+            expired:post()
+            return "w1: " .. (err or "connected")
+        end
+
+        local function waiter2()
+            local s = ngx.socket.tcp()
+            s:settimeout(500)
+            assert(s:connect("127.0.0.1", port, opts))
+            local reused = s:getreusedtimes()
+            s:close()
+            return "w2: reused " .. reused
+        end
+
+        local w1 = ngx.thread.spawn(waiter1)
+        local w2 = ngx.thread.spawn(waiter2)
+        assert(expired:wait(2))
+        assert(sock:setkeepalive())
+
+        local _, res1 = ngx.thread.wait(w1)
+        local _, res2 = ngx.thread.wait(w2)
+        ngx.say(res1)
+        ngx.say(res2)
+    }
+--- stream_response
+w1: timeout
+w2: reused 1
+--- error_log
+stream lua tcp socket queue connect operation for connection pool "expired_head"
+--- no_error_log
+[error]
+--- timeout: 4
+
+
+
+=== TEST 59: conn queuing: append after all existing waiters have been posted
+--- stream_server_config
+    content_by_lua_block {
+        local semaphore = require "ngx.semaphore"
+        local connected = semaphore.new()
+        local release = semaphore.new()
+        local port = $TEST_NGINX_SERVER_PORT
+        local opts = {pool = "append_posted", pool_size = 1, backlog = 2}
+        local sock = assert(ngx.socket.connect("127.0.0.1", port, opts))
+
+        local function waiter1()
+            local s = ngx.socket.tcp()
+            s:settimeout(500)
+            assert(s:connect("127.0.0.1", port, opts))
+            connected:post()
+            assert(release:wait(2))
+            s:close()
+            return "w1: connected"
+        end
+
+        local function waiter2()
+            local s = ngx.socket.tcp()
+            s:settimeout(500)
+            assert(s:connect("127.0.0.1", port, opts))
+            s:close()
+            return "w2: connected"
+        end
+
+        local w1 = ngx.thread.spawn(waiter1)
+        assert(sock:close())
+        local w2 = ngx.thread.spawn(waiter2)
+
+        assert(connected:wait(2))
+        release:post()
+        local _, res1 = ngx.thread.wait(w1)
+        local _, res2 = ngx.thread.wait(w2)
+        ngx.say(res1)
+        ngx.say(res2)
+    }
+--- stream_response
+w1: connected
+w2: connected
+--- error_log
+stream lua tcp socket queue connect operation for connection pool "append_posted"
+--- no_error_log
+[error]
+--- timeout: 4