bugfix: clean child coroutine ops when killing uthreads.

Cancel descendant coroutine operations before deleting killed user threads so delayed cosocket events cannot resume collected Lua states.

Author: zhuizhuhaomeng

src/ngx_stream_lua_uthread.c

@@ -34,6 +34,49 @@
 static int ngx_stream_lua_uthread_spawn(lua_State *L);
 static int ngx_stream_lua_uthread_wait(lua_State *L);
 static int ngx_stream_lua_uthread_kill(lua_State *L);
+static void ngx_stream_lua_uthread_cleanup_descendants(
+    ngx_stream_lua_ctx_t *ctx, ngx_stream_lua_co_ctx_t *parent);
+
+
+static void
+ngx_stream_lua_uthread_cleanup_descendants(ngx_stream_lua_ctx_t *ctx,
+    ngx_stream_lua_co_ctx_t *parent)
+{
+    ngx_uint_t                         i;
+    ngx_list_part_t                   *part;
+    ngx_stream_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_stream_lua_cleanup_pending_operation(&coctx[i]);
+                break;
+            }
+        }
+    }
+}
 
 
 void
@@ -282,6 +325,7 @@ ngx_stream_lua_uthread_kill(lua_State *L)
 
     default:
         ngx_stream_lua_cleanup_pending_operation(sub_coctx);
+        ngx_stream_lua_uthread_cleanup_descendants(ctx, sub_coctx);
         ngx_stream_lua_del_thread(r, L, ctx, sub_coctx);
         ctx->uthreads--;
 

t/127-uthread-kill.t

@@ -7,7 +7,7 @@ our $StapScript = $t::StapThread::StapScript;
 
 repeat_each(2);
 
-plan tests => repeat_each() * (blocks() * 5 + 1) - 2;
+plan tests => repeat_each() * (blocks() * 5 - 1) - 2;
 
 $ENV{TEST_NGINX_RESOLVER} ||= '8.8.8.8';
 $ENV{TEST_NGINX_MEMCACHED_PORT} ||= '11211';
@@ -394,3 +394,79 @@ ok_count=10
 --- no_error_log
 [error]
 [alert]
+
+
+
+=== TEST 8: kill uthread with child coroutine pending cosocket read after GC
+--- stream_config
+    server {
+        listen 127.0.0.1:$TEST_NGINX_RAND_PORT_1;
+
+        content_by_lua_block {
+            local body = string.rep("x", 8192)
+
+            ngx.print(body:sub(1, 64))
+            ngx.flush(true)
+            ngx.sleep(0.2)
+            ngx.print(body:sub(65))
+        }
+    }
+--- stream_server_config
+    lua_socket_log_errors off;
+
+    content_by_lua_block {
+        local port = $TEST_NGINX_RAND_PORT_1
+        local pinned_sockets = {}
+        local slow_read_pending = false
+
+        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
+
+                slow_read_pending = true
+                return sock:receive(8192)
+            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)
+    }
+--- stream_response
+survived pinned=1
+--- no_error_log
+[alert]