Skip to content

Commit 9bc2e1f

Browse files
zhuizhuhaomengzhuizhuhaomeng
committed
optimize: resume queued connects in constant time.
Replace the scan over posted connect events with a cursor pointing to the next unposted operation. Advance it when scheduling a waiter and when its waiter times out or is cancelled, and reset it on pool shutdown. Keep operations on the existing wait queue until resumed. Place the cursor after the pool's hot fields so their offsets stay unchanged, adding eight bytes per pool on 64-bit builds. Add regressions for closing several connections in one cycle, cancelling a posted operation, timing out the next waiter, and appending waiters after all existing operations have been posted. Keepalive, preread, timeout, and HUP checks pass 454 assertions. An isolated ASan probe verifies shutdown cleanup. Short fixed-CPU load tests show no throughput regression against the previous implementation. Template source: openresty/meta-lua-nginx-module@6b1be53.
1 parent fb91acd commit 9bc2e1f

3 files changed

Lines changed: 232 additions & 48 deletions

File tree

‎src/ngx_stream_lua_socket_tcp.c‎

Lines changed: 26 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -607,6 +607,7 @@ ngx_stream_lua_socket_tcp_create_socket_pool(lua_State *L,
607607

608608
ngx_queue_init(&sp->cache_connect_op);
609609
ngx_queue_init(&sp->wait_connect_op);
610+
sp->next_connect_op = ngx_queue_sentinel(&sp->wait_connect_op);
610611
ngx_queue_init(&sp->cache);
611612
ngx_queue_init(&sp->free);
612613

@@ -749,6 +750,12 @@ ngx_stream_lua_socket_tcp_connect_helper(lua_State *L,
749750
ngx_queue_insert_tail(&spool->wait_connect_op,
750751
&conn_op_ctx->queue);
751752

753+
if (spool->next_connect_op
754+
== ngx_queue_sentinel(&spool->wait_connect_op))
755+
{
756+
spool->next_connect_op = &conn_op_ctx->queue;
757+
}
758+
752759
ngx_log_debug3(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0,
753760
"stream lua tcp socket queued connect "
754761
"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)
48544861
ngx_stream_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx;
48554862

48564863
conn_op_ctx = ev->data;
4864+
u = conn_op_ctx->u;
4865+
4866+
if (u->socket_pool->next_connect_op == &conn_op_ctx->queue) {
4867+
u->socket_pool->next_connect_op = ngx_queue_next(&conn_op_ctx->queue);
4868+
}
4869+
48574870
ngx_queue_remove(&conn_op_ctx->queue);
48584871

4859-
u = conn_op_ctx->u;
48604872
r = u->request;
48614873

48624874
coctx = u->write_co_ctx;
@@ -4942,34 +4954,17 @@ ngx_stream_lua_socket_tcp_resume_conn_op(
49424954
* operation resumption, so that there is no resumption happens when Nginx
49434955
* is exiting.
49444956
*/
4945-
if (ngx_queue_empty(&spool->wait_connect_op)) {
4946-
return;
4947-
}
4948-
4949-
/* each freed slot must resume its own queued connect operation.
4950-
* A resumed operation stays in wait_connect_op until its posted event
4951-
* runs, so when several slots are freed in the same cycle, picking the
4952-
* head every time re-posts the same event, which ngx_post_event()
4953-
* ignores, and the extra slots are lost while the queue still counts
4954-
* them as busy. Resume the first operation not posted yet instead.
4955-
*/
4956-
for (q = ngx_queue_head(&spool->wait_connect_op);
4957-
q != ngx_queue_sentinel(&spool->wait_connect_op);
4958-
q = ngx_queue_next(q))
4959-
{
4960-
conn_op_ctx = ngx_queue_data(q,
4961-
ngx_stream_lua_socket_tcp_conn_op_ctx_t,
4962-
queue);
4963-
4964-
if (!conn_op_ctx->event.posted) {
4965-
break;
4966-
}
4967-
}
4968-
4957+
q = spool->next_connect_op;
49694958
if (q == ngx_queue_sentinel(&spool->wait_connect_op)) {
49704959
return;
49714960
}
49724961

4962+
/* advance before posting so each slot resumes a distinct waiter */
4963+
spool->next_connect_op = ngx_queue_next(q);
4964+
4965+
conn_op_ctx = ngx_queue_data(q,
4966+
ngx_stream_lua_socket_tcp_conn_op_ctx_t,
4967+
queue);
49734968
ngx_log_debug4(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0,
49744969
"stream lua tcp socket post connect operation "
49754970
"resumption u: %p, ctx: %p for connection pool \"%s\", "
@@ -5021,27 +5016,6 @@ ngx_stream_lua_socket_tcp_conn_op_resume_handler(ngx_event_t *ev)
50215016
r = u->request;
50225017
spool = u->socket_pool;
50235018

5024-
if (ngx_queue_empty(&spool->wait_connect_op)) {
5025-
#if (NGX_DEBUG)
5026-
ngx_stream_lua_assert(!(spool->backlog >= 0
5027-
&& spool->connections > spool->size));
5028-
5029-
#else
5030-
if (spool->backlog >= 0 && spool->connections > spool->size) {
5031-
ngx_log_error(NGX_LOG_ERR, ngx_cycle->log, 0,
5032-
"stream lua tcp socket connections count "
5033-
"mismatched for connection pool \"%s\", connections: "
5034-
"%i, size: %i",
5035-
spool->key, spool->connections, spool->size);
5036-
spool->connections = spool->size;
5037-
}
5038-
#endif
5039-
5040-
return;
5041-
}
5042-
5043-
/* several operations can be posted at once, so remove this one
5044-
* rather than the queue head */
50455019
ngx_queue_remove(&conn_op_ctx->queue);
50465020

50475021
coctx = u->write_co_ctx;
@@ -6542,6 +6516,8 @@ ngx_stream_lua_socket_shutdown_pool_helper(
65426516
ngx_stream_lua_socket_pool_item_t *item;
65436517
ngx_stream_lua_socket_tcp_conn_op_ctx_t *conn_op_ctx;
65446518

6519+
spool->next_connect_op = ngx_queue_sentinel(&spool->wait_connect_op);
6520+
65456521
while (!ngx_queue_empty(&spool->cache)) {
65466522
q = ngx_queue_head(&spool->cache);
65476523

@@ -6572,8 +6548,6 @@ ngx_stream_lua_socket_shutdown_pool_helper(
65726548
ngx_del_timer(&conn_op_ctx->event);
65736549
}
65746550

6575-
/* more than one queued operation can be posted for
6576-
* resumption now, do not free one with its event still posted */
65776551
if (conn_op_ctx->event.posted) {
65786552
ngx_delete_posted_event(&conn_op_ctx->event);
65796553
}
@@ -6989,6 +6963,10 @@ ngx_stream_lua_tcp_queue_conn_op_cleanup(void *data)
69896963
ngx_del_timer(&conn_op_ctx->event);
69906964
}
69916965

6966+
if (u->socket_pool->next_connect_op == &conn_op_ctx->queue) {
6967+
u->socket_pool->next_connect_op = ngx_queue_next(&conn_op_ctx->queue);
6968+
}
6969+
69926970
ngx_queue_remove(&conn_op_ctx->queue);
69936971
ngx_queue_insert_head(&u->socket_pool->cache_connect_op,
69946972
&conn_op_ctx->queue);

‎src/ngx_stream_lua_socket_tcp.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,8 @@ typedef struct {
8787

8888
ngx_int_t backlog;
8989

90+
ngx_queue_t *next_connect_op;
91+
9092
u_char key[1];
9193

9294
} ngx_stream_lua_socket_pool_t;

‎t/068-socket-keepalive.t‎

Lines changed: 204 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2542,3 +2542,207 @@ stream lua tcp socket queue connect operation for connection pool "same_cycle"
25422542
--- no_error_log
25432543
[error]
25442544
--- timeout: 4
2545+
2546+
2547+
2548+
=== TEST 56: conn queuing: close multiple connections before resumptions run
2549+
--- stream_server_config
2550+
content_by_lua_block {
2551+
local semaphore = require "ngx.semaphore"
2552+
local connected = semaphore.new()
2553+
local release = semaphore.new()
2554+
local port = $TEST_NGINX_SERVER_PORT
2555+
local opts = {pool = "same_cycle_close", pool_size = 2, backlog = 2}
2556+
2557+
local sock1 = assert(ngx.socket.connect("127.0.0.1", port, opts))
2558+
local sock2 = assert(ngx.socket.connect("127.0.0.1", port, opts))
2559+
2560+
local function waiter(name)
2561+
local sock = ngx.socket.tcp()
2562+
sock:settimeout(500)
2563+
assert(sock:connect("127.0.0.1", port, opts))
2564+
local reused = sock:getreusedtimes()
2565+
connected:post()
2566+
assert(release:wait(2))
2567+
sock:close()
2568+
return name .. ": reused " .. reused
2569+
end
2570+
2571+
local w1 = ngx.thread.spawn(waiter, "w1")
2572+
local w2 = ngx.thread.spawn(waiter, "w2")
2573+
2574+
assert(sock1:close())
2575+
assert(sock2:close())
2576+
2577+
-- neither waiter may free its slot before both have connected
2578+
assert(connected:wait(2))
2579+
assert(connected:wait(2))
2580+
release:post(2)
2581+
2582+
local _, res1 = ngx.thread.wait(w1)
2583+
local _, res2 = ngx.thread.wait(w2)
2584+
ngx.say(res1)
2585+
ngx.say(res2)
2586+
}
2587+
--- stream_response
2588+
w1: reused 0
2589+
w2: reused 0
2590+
--- error_log
2591+
stream lua tcp socket queue connect operation for connection pool "same_cycle_close"
2592+
--- no_error_log
2593+
[error]
2594+
--- timeout: 4
2595+
2596+
2597+
2598+
=== TEST 57: conn queuing: cancel a posted resumption and resume the next waiter
2599+
--- stream_server_config
2600+
content_by_lua_block {
2601+
local semaphore = require "ngx.semaphore"
2602+
local connected = semaphore.new()
2603+
local release = semaphore.new()
2604+
local port = $TEST_NGINX_SERVER_PORT
2605+
local opts = {pool = "cancel_posted", pool_size = 2, backlog = 3}
2606+
local order = {}
2607+
2608+
local sock1 = assert(ngx.socket.connect("127.0.0.1", port, opts))
2609+
local sock2 = assert(ngx.socket.connect("127.0.0.1", port, opts))
2610+
2611+
local function waiter(name)
2612+
local sock = ngx.socket.tcp()
2613+
sock:settimeout(500)
2614+
assert(sock:connect("127.0.0.1", port, opts))
2615+
order[#order + 1] = name
2616+
local reused = sock:getreusedtimes()
2617+
connected:post()
2618+
assert(release:wait(2))
2619+
sock:close()
2620+
return name .. ": reused " .. reused
2621+
end
2622+
2623+
local w1 = ngx.thread.spawn(waiter, "w1")
2624+
local w2 = ngx.thread.spawn(waiter, "w2")
2625+
local w3 = ngx.thread.spawn(waiter, "w3")
2626+
2627+
assert(sock1:setkeepalive())
2628+
assert(sock2:setkeepalive())
2629+
assert(ngx.thread.kill(w1))
2630+
2631+
assert(connected:wait(2))
2632+
assert(connected:wait(2))
2633+
release:post(2)
2634+
2635+
local _, res2 = ngx.thread.wait(w2)
2636+
local _, res3 = ngx.thread.wait(w3)
2637+
ngx.say(res2)
2638+
ngx.say(res3)
2639+
ngx.say("order: ", table.concat(order, ", "))
2640+
}
2641+
--- stream_response
2642+
w2: reused 1
2643+
w3: reused 1
2644+
order: w2, w3
2645+
--- error_log
2646+
stream lua tcp socket abort queueing
2647+
--- no_error_log
2648+
[error]
2649+
--- timeout: 4
2650+
2651+
2652+
2653+
=== TEST 58: conn queuing: advance past a timed out waiter
2654+
--- stream_server_config
2655+
lua_socket_log_errors off;
2656+
2657+
content_by_lua_block {
2658+
local semaphore = require "ngx.semaphore"
2659+
local expired = semaphore.new()
2660+
local port = $TEST_NGINX_SERVER_PORT
2661+
local opts = {pool = "expired_head", pool_size = 1, backlog = 2}
2662+
local sock = assert(ngx.socket.connect("127.0.0.1", port, opts))
2663+
2664+
local function waiter1()
2665+
local s = ngx.socket.tcp()
2666+
s:settimeout(10)
2667+
local ok, err = s:connect("127.0.0.1", port, opts)
2668+
expired:post()
2669+
return "w1: " .. (err or "connected")
2670+
end
2671+
2672+
local function waiter2()
2673+
local s = ngx.socket.tcp()
2674+
s:settimeout(500)
2675+
assert(s:connect("127.0.0.1", port, opts))
2676+
local reused = s:getreusedtimes()
2677+
s:close()
2678+
return "w2: reused " .. reused
2679+
end
2680+
2681+
local w1 = ngx.thread.spawn(waiter1)
2682+
local w2 = ngx.thread.spawn(waiter2)
2683+
assert(expired:wait(2))
2684+
assert(sock:setkeepalive())
2685+
2686+
local _, res1 = ngx.thread.wait(w1)
2687+
local _, res2 = ngx.thread.wait(w2)
2688+
ngx.say(res1)
2689+
ngx.say(res2)
2690+
}
2691+
--- stream_response
2692+
w1: timeout
2693+
w2: reused 1
2694+
--- error_log
2695+
stream lua tcp socket queue connect operation for connection pool "expired_head"
2696+
--- no_error_log
2697+
[error]
2698+
--- timeout: 4
2699+
2700+
2701+
2702+
=== TEST 59: conn queuing: append after all existing waiters have been posted
2703+
--- stream_server_config
2704+
content_by_lua_block {
2705+
local semaphore = require "ngx.semaphore"
2706+
local connected = semaphore.new()
2707+
local release = semaphore.new()
2708+
local port = $TEST_NGINX_SERVER_PORT
2709+
local opts = {pool = "append_posted", pool_size = 1, backlog = 2}
2710+
local sock = assert(ngx.socket.connect("127.0.0.1", port, opts))
2711+
2712+
local function waiter1()
2713+
local s = ngx.socket.tcp()
2714+
s:settimeout(500)
2715+
assert(s:connect("127.0.0.1", port, opts))
2716+
connected:post()
2717+
assert(release:wait(2))
2718+
s:close()
2719+
return "w1: connected"
2720+
end
2721+
2722+
local function waiter2()
2723+
local s = ngx.socket.tcp()
2724+
s:settimeout(500)
2725+
assert(s:connect("127.0.0.1", port, opts))
2726+
s:close()
2727+
return "w2: connected"
2728+
end
2729+
2730+
local w1 = ngx.thread.spawn(waiter1)
2731+
assert(sock:close())
2732+
local w2 = ngx.thread.spawn(waiter2)
2733+
2734+
assert(connected:wait(2))
2735+
release:post()
2736+
local _, res1 = ngx.thread.wait(w1)
2737+
local _, res2 = ngx.thread.wait(w2)
2738+
ngx.say(res1)
2739+
ngx.say(res2)
2740+
}
2741+
--- stream_response
2742+
w1: connected
2743+
w2: connected
2744+
--- error_log
2745+
stream lua tcp socket queue connect operation for connection pool "append_posted"
2746+
--- no_error_log
2747+
[error]
2748+
--- timeout: 4

0 commit comments

Comments
 (0)