fix: remove request level pipe proc in openresty 1.21.4.2 (#78)

openresty optimized the pipe api and destroyed the pipeline process at the end of the request.

The APISIX plugin runner process lifecycle does not apply to request level,
Instead, it is pulled up in advance and processed continuously through RPC calls until the process exits.

Patches:
* lua-resty-core-pipe_runner.patch
* ngx_lua-pipe_runner.patch

Author: AlinsRan

patch/1.21.4/lua-resty-core-pipe_runner.patch

@@ -0,0 +1,25 @@
+diff --git lib/ngx/pipe.lua lib/ngx/pipe.lua
+index 3757ab7..413a10e 100644
+--- lib/ngx/pipe.lua
++++ lib/ngx/pipe.lua
+@@ -47,8 +47,7 @@ typedef struct {
+     ngx_http_lua_pipe_t    *pipe;
+ } ngx_http_lua_ffi_pipe_proc_t;
+ 
+-int ngx_http_lua_ffi_pipe_spawn(ngx_http_request_t *r,
+-    ngx_http_lua_ffi_pipe_proc_t *proc,
++int ngx_http_lua_ffi_pipe_spawn(ngx_http_lua_ffi_pipe_proc_t *proc,
+     const char *file, const char **argv, int merge_stderr, size_t buffer_size,
+     const char **environ, u_char *errbuf, size_t *errbuf_size);
+ 
+@@ -625,9 +624,8 @@ do
+ 
+         local errbuf = get_string_buf(ERR_BUF_SIZE)
+         local errbuf_size = get_size_ptr()
+-        local r = get_request()
+         errbuf_size[0] = ERR_BUF_SIZE
+-        local rc = C.ngx_http_lua_ffi_pipe_spawn(r, proc, exe, proc_args,
++        local rc = C.ngx_http_lua_ffi_pipe_spawn(proc, exe, proc_args,
+                                                  merge_stderr, buffer_size,
+                                                  proc_envs, errbuf, errbuf_size)
+         if rc == FFI_ERROR then

patch/1.21.4/ngx_lua-pipe-runner.patch

@@ -0,0 +1,70 @@
+diff --git src/ngx_http_lua_pipe.c src/ngx_http_lua_pipe.c
+index 8c0884bc..8369160f 100644
+--- src/ngx_http_lua_pipe.c
++++ src/ngx_http_lua_pipe.c
+@@ -599,8 +599,7 @@ ngx_http_lua_execvpe(const char *program, char * const argv[],
+ 
+ 
+ int
+-ngx_http_lua_ffi_pipe_spawn(ngx_http_request_t *r,
+-    ngx_http_lua_ffi_pipe_proc_t *proc,
++ngx_http_lua_ffi_pipe_spawn(ngx_http_lua_ffi_pipe_proc_t *proc,
+     const char *file, const char **argv, int merge_stderr, size_t buffer_size,
+     const char **environ, u_char *errbuf, size_t *errbuf_size)
+ {
+@@ -620,7 +619,6 @@ ngx_http_lua_ffi_pipe_spawn(ngx_http_request_t *r,
+     ngx_http_lua_pipe_node_t       *pipe_node;
+     struct sigaction                sa;
+     ngx_http_lua_pipe_signal_t     *sig;
+-    ngx_pool_cleanup_t             *cln;
+     sigset_t                        set;
+ 
+     pool_size = ngx_align(NGX_MIN_POOL_SIZE + buffer_size * 2,
+@@ -908,21 +906,6 @@ ngx_http_lua_ffi_pipe_spawn(ngx_http_request_t *r,
+         pp->stderr_fd = stderr_fd;
+     }
+ 
+-    if (pp->cleanup == NULL) {
+-        cln = ngx_pool_cleanup_add(r->pool, 0);
+-
+-        if (cln == NULL) {
+-            *errbuf_size = ngx_snprintf(errbuf, *errbuf_size, "no memory")
+-                           - errbuf;
+-            goto close_in_out_err_fd;
+-        }
+-
+-        cln->handler = (ngx_pool_cleanup_pt) ngx_http_lua_ffi_pipe_proc_destroy;
+-        cln->data = proc;
+-        pp->cleanup = &cln->handler;
+-        pp->r = r;
+-    }
+-
+     node = (ngx_rbtree_node_t *) (pp + 1);
+     node->key = pid;
+     pipe_node = (ngx_http_lua_pipe_node_t *) &node->color;
+@@ -1193,12 +1176,6 @@ ngx_http_lua_ffi_pipe_proc_destroy(ngx_http_lua_ffi_pipe_proc_t *proc)
+         }
+     }
+ 
+-    if (pipe->cleanup != NULL) {
+-        *pipe->cleanup = NULL;
+-        ngx_http_lua_cleanup_free(pipe->r, pipe->cleanup);
+-        pipe->cleanup = NULL;
+-    }
+-
+     ngx_http_lua_pipe_proc_finalize(proc);
+     ngx_destroy_pool(pipe->pool);
+     proc->pipe = NULL;
+diff --git src/ngx_http_lua_pipe.h src/ngx_http_lua_pipe.h
+index f1c92835..ecb86c88 100644
+--- src/ngx_http_lua_pipe.h
++++ src/ngx_http_lua_pipe.h
+@@ -57,8 +57,6 @@ struct ngx_http_lua_pipe_s {
+     ngx_http_lua_pipe_ctx_t            *stdout_ctx;
+     ngx_http_lua_pipe_ctx_t            *stderr_ctx;
+     ngx_http_lua_pipe_retval_handler    retval_handler;
+-    ngx_http_cleanup_pt                *cleanup;
+-    ngx_http_request_t                 *r;
+     size_t                              buffer_size;
+     unsigned                            closed:1;
+     unsigned                            dead:1;

t/pipe.t

@@ -0,0 +1,215 @@
+use t::APISIX_NGINX 'no_plan';
+
+
+add_block_preprocessor(sub {
+    my ($block) = @_;
+
+    if (!$block->request) {
+        $block->set_value("request", "GET /t");
+    }
+});
+
+run_tests;
+
+__DATA__
+
+=== TEST 1: check pipe spawn arguments
+--- config
+    location = /t {
+        content_by_lua_block {
+            local ngx_pipe = require "ngx.pipe"
+
+            local function check_error(...)
+                local data, err = pcall(...)
+                if not data then
+                    ngx.say(err)
+                else
+                    ngx.say('ok')
+                end
+            end
+
+            check_error(ngx_pipe.spawn, nil)
+            check_error(ngx_pipe.spawn, {})
+            check_error(ngx_pipe.spawn, {"ls"}, {buffer_size = 0})
+            check_error(ngx_pipe.spawn, {"ls"}, {buffer_size = 0.5})
+            check_error(ngx_pipe.spawn, {"ls"}, {buffer_size = "1"})
+            check_error(ngx_pipe.spawn, {"ls"}, {buffer_size = true})
+        }
+    }
+--- request
+--- response_body
+bad args arg: table expected, got nil
+bad args arg: non-empty table expected
+bad buffer_size option
+bad buffer_size option
+ok
+bad buffer_size option
+
+
+
+=== TEST 2: spawn process, with buffer_size option
+--- config
+    location = /t {
+        content_by_lua_block {
+            local ngx_pipe = require "ngx.pipe"
+            local proc, err = ngx_pipe.spawn({"ls"}, {buffer_size = 256})
+            if not proc then
+                ngx.say(err)
+            else
+                ngx.say('ok')
+            end
+        }
+    }
+--- response_body
+ok
+--- error_log eval
+qr/lua pipe spawn process:[0-9A-F]+ pid:\d+ merge_stderr:0 buffer_size:256/
+--- no_error_log
+[error]
+
+
+
+=== TEST 3: ensure process is destroyed in GC
+--- config
+    location = /t {
+        content_by_lua_block {
+            local ngx_pipe = require "ngx.pipe"
+            do
+                local proc, err = ngx_pipe.spawn({"ls", "-l"})
+                if not proc then
+                    ngx.say(err)
+                    return
+                end
+            end
+
+            collectgarbage()
+            ngx.say("ok")
+        }
+    }
+--- response_body
+ok
+--- no_error_log
+[error]
+--- error_log
+lua pipe destroy process:
+
+
+
+=== TEST 4: check phase for process wait
+--- config
+    location = /t {
+        content_by_lua_block {
+            local ngx_pipe = require "ngx.pipe"
+            local proc, err = ngx_pipe.spawn({"sleep", 0.1})
+            if not proc then
+                ngx.say(err)
+                return
+            end
+
+            package.loaded.proc = proc
+        }
+
+        log_by_lua_block {
+            package.loaded.proc:wait()
+        }
+    }
+--- error_log
+API disabled in the context of log_by_lua
+
+
+
+=== TEST 5: check process wait arguments
+--- config
+    location = /t {
+        content_by_lua_block {
+            local ngx_pipe = require "ngx.pipe"
+            local proc, err = ngx_pipe.spawn({"sleep", 0.1})
+            proc.wait()
+        }
+    }
+--- error_code: 500
+--- ignore_response_body
+--- error_log eval
+qr/\[error\] .*? runtime error: content_by_lua\(nginx\.conf\:\d+\):\d+: not a process instance/
+--- no_error_log
+[crit]
+
+
+
+=== TEST 6: wait an already waited process
+--- config
+    location = /t {
+        content_by_lua_block {
+            local ngx_pipe = require "ngx.pipe"
+            local proc, err = ngx_pipe.spawn({"ls"})
+            if not proc then
+                ngx.say(err)
+                return
+            end
+
+            local ok, err = proc:wait()
+            if not ok then
+                ngx.say(err)
+                return
+            end
+
+            local ok, err = proc:wait()
+            if not ok then
+                ngx.say(err)
+            end
+        }
+    }
+--- response_body
+exited
+
+
+
+=== TEST 7: more than one coroutines wait a process
+--- config
+    location = /t {
+        content_by_lua_block {
+            local ngx_pipe = require "ngx.pipe"
+            local proc, err = ngx_pipe.spawn({"sleep", 0.1})
+            if not proc then
+                ngx.say(err)
+                return
+            end
+
+            local function wait()
+                local ok, err = proc:wait()
+                if not ok then
+                    ngx.say(err)
+                end
+            end
+
+            local th1 = ngx.thread.spawn(wait)
+            local th2 = ngx.thread.spawn(wait)
+            ngx.thread.wait(th1)
+            ngx.thread.wait(th2)
+            ngx.thread.spawn(wait)
+        }
+    }
+--- response_body
+pipe busy waiting
+exited
+
+
+
+=== TEST 8: kill living sub-process during Lua VM destruction.
+--- config
+    location = /t {
+        content_by_lua_block {
+            local ngx_pipe = require "ngx.pipe"
+            local proc, err = ngx_pipe.spawn({"sleep", 3600})
+            if not proc then
+                ngx.say(err)
+                return
+            end
+            ngx.say("ok")
+        }
+    }
+--- response_body
+ok
+--- shutdown_error_log
+lua pipe destroy process:
+lua pipe kill process: