diff --git a/src/ngx_http_lua_uthread.c b/src/ngx_http_lua_uthread.c index 2208b2a819..e660abac6b 100644 --- a/src/ngx_http_lua_uthread.c +++ b/src/ngx_http_lua_uthread.c @@ -25,6 +25,49 @@ static int ngx_http_lua_uthread_spawn(lua_State *L); static int ngx_http_lua_uthread_wait(lua_State *L); static int ngx_http_lua_uthread_kill(lua_State *L); +static void ngx_http_lua_uthread_cleanup_descendants(ngx_http_lua_ctx_t *ctx, + ngx_http_lua_co_ctx_t *parent); + + +static void +ngx_http_lua_uthread_cleanup_descendants(ngx_http_lua_ctx_t *ctx, + ngx_http_lua_co_ctx_t *parent) +{ + ngx_uint_t i; + ngx_list_part_t *part; + ngx_http_lua_co_ctx_t *coctx, *cur; + + if (ctx->user_co_ctx == NULL) { + return; + } + + part = &ctx->user_co_ctx->part; + coctx = part->elts; + + for (i = 0; /* void */; i++) { + + if (i >= part->nelts) { + if (part->next == NULL) { + break; + } + + part = part->next; + coctx = part->elts; + i = 0; + } + + if (&coctx[i] == parent || coctx[i].is_uthread) { + continue; + } + + for (cur = coctx[i].parent_co_ctx; cur; cur = cur->parent_co_ctx) { + if (cur == parent) { + ngx_http_lua_cleanup_pending_operation(&coctx[i]); + break; + } + } + } +} void @@ -264,6 +307,7 @@ ngx_http_lua_uthread_kill(lua_State *L) default: ngx_http_lua_cleanup_pending_operation(sub_coctx); + ngx_http_lua_uthread_cleanup_descendants(ctx, sub_coctx); ngx_http_lua_del_thread(r, L, ctx, sub_coctx); ctx->uthreads--; diff --git a/t/127-uthread-kill.t b/t/127-uthread-kill.t index c397d433b4..b6f7e664fe 100644 --- a/t/127-uthread-kill.t +++ b/t/127-uthread-kill.t @@ -617,3 +617,122 @@ GET /t ok_count=10 --- no_error_log [error] + + + +=== TEST 11: kill uthread with child coroutine pending cosocket read after GC +--- config + lua_socket_log_errors off; + + location = /slow-body { + content_by_lua_block { + local body = string.rep("x", 8192) + + ngx.header["Content-Type"] = "text/plain" + ngx.header["Content-Length"] = #body + + ngx.print(body:sub(1, 64)) + ngx.flush(true) + ngx.sleep(0.2) + ngx.print(body:sub(65)) + } + } + + location = /t { + access_by_lua_block { + local port = tonumber(ngx.var.server_port) + local pinned_sockets = {} + local slow_read_pending = false + + local function read_content_length(sock) + local line, err = sock:receive("*l") + if not line then + return nil, err + end + + local content_length + while true do + line, err = sock:receive("*l") + if not line then + return nil, err + end + + if line == "" then + return content_length + end + + local name, value = line:match("^([^:]+):%s*(.*)$") + if name and name:lower() == "content-length" then + content_length = tonumber(value) + end + end + end + + local fast = ngx.thread.spawn(function() + for _ = 1, 1000 do + if slow_read_pending then + break + end + ngx.sleep(0.001) + end + + return "ok" + end) + + local slow = ngx.thread.spawn(function() + local child = coroutine.create(function() + local sock = ngx.socket.tcp() + pinned_sockets[#pinned_sockets + 1] = sock + + sock:settimeout(10000) + local ok, err = sock:connect("127.0.0.1", port) + if not ok then + return nil, err + end + + ok, err = sock:send("GET /slow-body HTTP/1.1\r\nHost: localhost\r\n\r\n") + if not ok then + return nil, err + end + + local content_length + content_length, err = read_content_length(sock) + if not content_length then + return nil, err + end + + slow_read_pending = true + return sock:receive(content_length) + end) + + local ok, body, err = coroutine.resume(child) + if not ok then + return nil, body + end + + return body, err + end) + + ngx.thread.wait(fast, slow) + ngx.thread.kill(fast) + ngx.thread.kill(slow) + + fast = nil + slow = nil + collectgarbage("collect") + collectgarbage("collect") + + ngx.sleep(0.3) + + ngx.say("survived pinned=", #pinned_sockets) + } + + content_by_lua_block { + } + } +--- request +GET /t +--- response_body +survived pinned=1 +--- no_error_log +[alert]