From 47b64e2af430239ac830131c7850352a8f0fcff0 Mon Sep 17 00:00:00 2001 From: kurt tu Date: Sun, 6 Sep 2026 21:20:21 +0800 Subject: [PATCH] feat: support cosocket and other yieldable APIs in init_worker_by_lua* Enable TCP/UDP cosockets, ngx.sleep, ngx.semaphore, ngx.thread.spawn, and coroutine.* inside init_worker_by_lua* by running a bounded event pump when the init code yields. Ported from ngx_http_lua_module and adapted to the stream subsystem. Key changes: - Add ngx_stream_lua_init_worker_pump() to drive the event loop during yields, including the posted delayed-events queue drain - Add ngx_stream_lua_init_worker_pump_loop() with timeout budget management - Add ngx_stream_lua_init_worker_toggle_accept() to disarm/re-arm listeners around the pump (prevents level-triggered busy-spin from pending accepts; mirrors nginx's reuseport/EPOLLEXCLUSIVE arming) - Freeze ngx.timer.at/every registrations via ctx->context gate; flush at runner exit with absolute expiry preservation (timer ordering identical to before) - Run the init chunk on an entry coroutine created with ngx_stream_lua_new_thread() on the global VM; finalize via finalize_fake_request on completion - Add lua_init_worker_timeout directive (default 0 = no timeout) - Add lua_init_worker_abort_on_error directive (default off) - Add INIT_WORKER to NGX_STREAM_LUA_CONTEXT_YIELDABLE mask in util.h - Change runtime error prefix to "lua entry thread aborted:" (breaking) - Update README: new directives and cosocket availability note --- README.md | 15 + src/ngx_stream_lua_common.h | 4 + src/ngx_stream_lua_initworkerby.c | 361 ++++++++++++++++++++- src/ngx_stream_lua_module.c | 25 ++ src/ngx_stream_lua_timer.c | 13 + src/ngx_stream_lua_util.h | 2 +- t/172-init-worker-cosocket.t | 469 +++++++++++++++++++++++++++ t/173-init-worker-accept.t | 71 ++++ t/174-init-worker-accept-workers.t | 63 ++++ t/175-init-worker-abort.t | 58 ++++ t/176-init-worker-accept-reuseport.t | 69 ++++ 11 files changed, 1141 insertions(+), 9 deletions(-) create mode 100644 t/172-init-worker-cosocket.t create mode 100644 t/173-init-worker-accept.t create mode 100644 t/174-init-worker-accept-workers.t create mode 100644 t/175-init-worker-abort.t create mode 100644 t/176-init-worker-accept-reuseport.t diff --git a/README.md b/README.md index cf62587c..1ebcfd13 100644 --- a/README.md +++ b/README.md @@ -139,6 +139,8 @@ behavior. * [init_by_lua_file](https://github.com/openresty/lua-nginx-module#init_by_lua_file) * [init_worker_by_lua_block](https://github.com/openresty/lua-nginx-module#init_worker_by_lua_block) * [init_worker_by_lua_file](https://github.com/openresty/lua-nginx-module#init_worker_by_lua_file) +* [lua_init_worker_timeout](https://github.com/openresty/lua-nginx-module#lua_init_worker_timeout) +* [lua_init_worker_abort_on_error](https://github.com/openresty/lua-nginx-module#lua_init_worker_abort_on_error) * [preread_by_lua_block](#preread_by_lua_block) * [preread_by_lua_file](#preread_by_lua_file) * [content_by_lua_block](https://github.com/openresty/lua-nginx-module#content_by_lua_block) @@ -179,6 +181,19 @@ behavior. * [lua_capture_error_log](https://github.com/openresty/lua-nginx-module#lua_capture_error_log) * [preread_by_lua_no_postpone](#preread_by_lua_no_postpone) +Cosockets (and other yieldable APIs) are supported in `init_worker_by_lua*`. +When a cosocket operation yields (for example, during a `connect` or +`receive` call), the module runs a lightweight event pump that drives the +Nginx event loop until the operation completes. See +[lua_init_worker_timeout](https://github.com/openresty/lua-nginx-module#lua_init_worker_timeout) +and +[lua_init_worker_abort_on_error](https://github.com/openresty/lua-nginx-module#lua_init_worker_abort_on_error) +in ngx_http_lua for the directives that bound how long the init code may +block worker startup and how Lua runtime errors are handled. + +Note that `ngx.timer.at` callbacks registered during `init_worker_by_lua*` +are deferred: they will not run until the init code finishes. + The [send_timeout](https://nginx.org/r/send_timeout) directive in the Nginx "http" subsystem is missing in the "stream" subsystem. As such, ngx_stream_lua_module uses the `lua_socket_send_timeout` directive for this diff --git a/src/ngx_stream_lua_common.h b/src/ngx_stream_lua_common.h index 1126f7dd..67141fb2 100644 --- a/src/ngx_stream_lua_common.h +++ b/src/ngx_stream_lua_common.h @@ -239,6 +239,10 @@ struct ngx_stream_lua_main_conf_s { ngx_stream_lua_main_conf_handler_pt init_worker_handler; ngx_str_t init_worker_src; + ngx_msec_t init_worker_timeout; + ngx_flag_t init_worker_abort_on_error; + ngx_queue_t deferred_timers; + ngx_stream_lua_balancer_peer_data_t *balancer_peer_data; /* neither yielding nor recursion is possible in * balancer_by_lua*, so there cannot be any races among diff --git a/src/ngx_stream_lua_initworkerby.c b/src/ngx_stream_lua_initworkerby.c index e5ffb4d1..7e10718e 100644 --- a/src/ngx_stream_lua_initworkerby.c +++ b/src/ngx_stream_lua_initworkerby.c @@ -28,6 +28,244 @@ static u_char *ngx_stream_lua_log_init_worker_error(ngx_log_t *log, u_char *buf, size_t len); +typedef struct { + unsigned done:1; + unsigned failed:1; + unsigned timeout:1; +} ngx_stream_lua_init_worker_state_t; + + +static void ngx_stream_lua_init_worker_done(void *data); + +static void ngx_stream_lua_init_worker_toggle_accept( + ngx_cycle_t *cycle, ngx_uint_t arm); + +static void ngx_stream_lua_init_worker_pump(ngx_cycle_t *cycle, + ngx_msec_t budget); + +static void ngx_stream_lua_init_worker_flush_timers( + ngx_stream_lua_main_conf_t *lmcf); + +static void ngx_stream_lua_init_worker_pump_loop(ngx_cycle_t *cycle, + ngx_stream_lua_main_conf_t *lmcf, + ngx_stream_lua_init_worker_state_t *st); + + +static void +ngx_stream_lua_init_worker_done(void *data) +{ + ngx_stream_lua_init_worker_state_t *st = data; + + st->done = 1; +} + + +/* + * Disarm/re-arm listening sockets around the pump. Mirrors nginx's + * arming logic (reuseport skip, accept_mutex skip, EPOLLEXCLUSIVE). + */ +static void +ngx_stream_lua_init_worker_toggle_accept(ngx_cycle_t *cycle, ngx_uint_t arm) +{ + ngx_listening_t *ls; + ngx_connection_t *c; + ngx_event_t *rev; + ngx_uint_t i; + ngx_uint_t flags; +#if (NGX_HAVE_EPOLLEXCLUSIVE) + ngx_uint_t exclusive; +#endif + + if (ngx_use_accept_mutex) { + return; + } + +#if (NGX_HAVE_EPOLLEXCLUSIVE) + { + ngx_core_conf_t *ccf; + + ccf = (ngx_core_conf_t *) ngx_get_conf(cycle->conf_ctx, + ngx_core_module); + exclusive = ((ngx_event_flags & NGX_USE_EPOLL_EVENT) + && ccf->worker_processes > 1); + } +#endif + + ls = cycle->listening.elts; + + for (i = 0; i < cycle->listening.nelts; i++) { + +#if (NGX_HAVE_REUSEPORT) + if (ls[i].reuseport && ls[i].worker != ngx_worker) { + /* other workers' sockets: ls[i].connection is NULL here */ + continue; + } +#endif + + c = ls[i].connection; + + if (c == NULL) { + continue; + } + + rev = c->read; + + if (!arm) { + if (rev->active + && ngx_del_event(rev, NGX_READ_EVENT, 0) == NGX_ERROR) + { + ngx_log_error(NGX_LOG_ALERT, cycle->log, 0, + "init_worker: failed to disarm listen fd %d", + c->fd); + } + + continue; + } + + if (rev->active) { + continue; + } + + flags = 0; + +#if (NGX_HAVE_EPOLLEXCLUSIVE) + if (exclusive +#if (NGX_HAVE_REUSEPORT) + /* upstream registers reuseport sockets with plain flags, + * before the EPOLLEXCLUSIVE branch (ngx_event.c:907) */ + && !ls[i].reuseport +#endif + ) + { + flags = NGX_EXCLUSIVE_EVENT; + } +#endif + + if (ngx_add_event(rev, NGX_READ_EVENT, flags) == NGX_ERROR) { + ngx_log_error(NGX_LOG_ALERT, cycle->log, 0, + "init_worker: failed to re-arm listen fd %d, " + "worker will not accept on it", c->fd); + } + } +} + + +static void +ngx_stream_lua_init_worker_pump(ngx_cycle_t *cycle, ngx_msec_t budget) +{ + ngx_msec_t timer; + + timer = ngx_event_find_timer(); + + if (timer == NGX_TIMER_INFINITE || timer > budget) { + timer = budget; + } + + if (!ngx_queue_empty(&ngx_posted_next_events)) { + ngx_event_move_posted_next(cycle); + timer = 0; + } + +#ifdef HAVE_POSTED_DELAYED_EVENTS_PATCH + if (!ngx_queue_empty(&ngx_posted_delayed_events)) { + timer = 0; + } +#endif + + (void) ngx_process_events(cycle, timer, NGX_UPDATE_TIME | NGX_POST_EVENTS); + + ngx_event_expire_timers(); + + ngx_event_process_posted(cycle, &ngx_posted_events); + +#ifdef HAVE_POSTED_DELAYED_EVENTS_PATCH + ngx_event_process_posted(cycle, &ngx_posted_delayed_events); +#endif +} + + +static void +ngx_stream_lua_init_worker_flush_timers( + ngx_stream_lua_main_conf_t *lmcf) +{ + ngx_queue_t *q; + ngx_event_t *ev; + ngx_msec_int_t remaining; + + while (!ngx_queue_empty(&lmcf->deferred_timers)) { + q = ngx_queue_head(&lmcf->deferred_timers); + ngx_queue_remove(q); + + ev = ngx_queue_data(q, ngx_event_t, queue); + + /* timer.key stores the absolute expiry (registration now + delay) */ + remaining = (ngx_msec_int_t) (ev->timer.key - ngx_current_msec); + + if (remaining <= 0) { + /* already expired: dispatch as the delay==0 path would */ +#ifdef HAVE_POSTED_DELAYED_EVENTS_PATCH + ngx_post_event(ev, &ngx_posted_delayed_events); +#else + ngx_add_timer(ev, 0); +#endif + continue; + } + + ngx_add_timer(ev, (ngx_msec_t) remaining); + } +} + + +static void +ngx_stream_lua_init_worker_pump_loop(ngx_cycle_t *cycle, + ngx_stream_lua_main_conf_t *lmcf, + ngx_stream_lua_init_worker_state_t *st) +{ + ngx_msec_t deadline = 0, budget; + ngx_uint_t unlimited; + + unlimited = (lmcf->init_worker_timeout == 0); + + if (!unlimited) { + ngx_time_update(); + deadline = ngx_current_msec + lmcf->init_worker_timeout; + } + + ngx_stream_lua_init_worker_toggle_accept(cycle, 0); + + for ( ;; ) { + if (st->done) { + break; + } + + if (ngx_terminate || ngx_quit || ngx_exiting) { + ngx_log_error(NGX_LOG_WARN, cycle->log, 0, + "init_worker_by_lua* aborted by signal while " + "waiting for a yielded operation"); + break; + } + + if (unlimited) { + budget = NGX_TIMER_INFINITE; + + } else { + if ((ngx_msec_int_t) (deadline - ngx_current_msec) <= 0) { + st->timeout = 1; + ngx_log_error(NGX_LOG_ERR, cycle->log, 0, + "init_worker_by_lua* timed out after %M ms", + lmcf->init_worker_timeout); + break; + } + + budget = deadline - ngx_current_msec; + } + + ngx_stream_lua_init_worker_pump(cycle, budget); + } + + ngx_stream_lua_init_worker_toggle_accept(cycle, 1); +} + ngx_int_t ngx_stream_lua_init_worker(ngx_cycle_t *cycle) @@ -54,6 +292,14 @@ ngx_stream_lua_init_worker(ngx_cycle_t *cycle) ngx_stream_core_srv_conf_t *cscf, *top_cscf; ngx_stream_lua_srv_conf_t *lscf, *top_lscf; + ngx_pool_cleanup_t *cln; + ngx_stream_lua_cleanup_t *scln; + lua_State *co; + ngx_int_t rc; + int co_ref = LUA_NOREF; + + ngx_stream_lua_init_worker_state_t st; + lmcf = ngx_stream_cycle_get_module_main_conf(cycle, ngx_stream_lua_module); if (lmcf == NULL || lmcf->lua == NULL) { @@ -301,16 +547,117 @@ ngx_stream_lua_init_worker(ngx_cycle_t *cycle) r = ctx->request; ctx->context = NGX_STREAM_LUA_CONTEXT_INIT_WORKER; - ctx->cur_co_ctx = NULL; + ctx->cur_co_ctx = &ctx->entry_co_ctx; r->read_event_handler = ngx_stream_lua_block_reading; - ngx_stream_lua_set_req(lmcf->lua, r); + /* compile the chunk; run it on a coroutine below */ + if (lmcf->init_worker_handler(cycle->log, lmcf, lmcf->lua) != NGX_OK) { + /* compile-time error: already logged by the handler */ + ngx_stream_lua_set_req(lmcf->lua, NULL); + ngx_stream_lua_finalize_request(r, NGX_ERROR); + return NGX_OK; + } + + /* lmcf->lua stack top: chunk_func */ + + co = ngx_stream_lua_new_thread(r, lmcf->lua, &co_ref); + + if (co == NULL) { + ngx_log_error(NGX_LOG_ERR, cycle->log, 0, + "lua: failed to create new coroutine for " + "init_worker_by_lua*"); - (void) lmcf->init_worker_handler(cycle->log, lmcf, lmcf->lua); + goto runner_failed; + } + + /* move code closure to new coroutine */ + lua_xmove(lmcf->lua, co, 1); + +#ifndef OPENRESTY_LUAJIT + /* set closure's env table to new coroutine's globals table */ + ngx_stream_lua_get_globals_table(co); + lua_setfenv(co, -2); +#endif + + ngx_stream_lua_set_req(co, r); + + ctx->cur_co_ctx->co = co; + ctx->cur_co_ctx->co_ref = co_ref; + +#ifdef NGX_LUA_USE_ASSERT + ctx->cur_co_ctx->co_top = 1; +#endif + + ngx_stream_lua_attach_co_ctx_to_L(co, ctx->cur_co_ctx); + + ctx->entered_content_phase = 1; + + /* {{{ register request cleanup hooks */ + scln = ngx_stream_lua_cleanup_add(r, 0); + if (scln == NULL) { + goto runner_failed; + } + + scln->handler = ngx_stream_lua_request_cleanup_handler; + scln->data = ctx; + ctx->cleanup = &scln->handler; + /* }}} */ + + ngx_memzero(&st, sizeof(st)); + + cln = ngx_pool_cleanup_add(r->pool, 0); + if (cln == NULL) { + goto runner_failed; + } + + cln->handler = ngx_stream_lua_init_worker_done; + cln->data = &st; + + rc = ngx_stream_lua_run_thread(lmcf->lua, r, ctx, 0); + + if (rc == NGX_AGAIN || rc == NGX_DONE) { + ngx_stream_lua_init_worker_pump_loop(cycle, lmcf, &st); + + } else { + st.done = 1; + st.failed = (rc == NGX_ERROR || rc >= NGX_STREAM_BAD_REQUEST); + } + + if (!st.done) { + ngx_log_error(NGX_LOG_ERR, cycle->log, 0, + "init_worker_by_lua* failed to complete"); + + ngx_stream_lua_request_cleanup(ctx, 1); + ngx_stream_lua_finalize_request(r, NGX_ERROR); + + } else if (rc != NGX_AGAIN && rc != NGX_DONE) { + ngx_stream_lua_finalize_request(r, rc); + } + + ngx_stream_lua_init_worker_flush_timers(lmcf); + ngx_stream_lua_set_req(lmcf->lua, NULL); + + if (st.failed || st.timeout) { + if (lmcf->init_worker_abort_on_error) { + return NGX_ERROR; + } + } - ngx_destroy_pool(c->pool); return NGX_OK; +runner_failed: + + /* free the coroutine before the pool holding r is destroyed */ + if (ctx != NULL && ctx->entry_co_ctx.co_ref != LUA_NOREF) { + ngx_stream_lua_del_thread(r, lmcf->lua, ctx, &ctx->entry_co_ctx); + } + + if (c != NULL) { + ngx_stream_lua_close_fake_connection(c); + } + + return NGX_ERROR; + failed: if (conf.temp_pool) { @@ -332,8 +679,7 @@ ngx_stream_lua_init_worker_by_inline(ngx_log_t *log, int status; status = luaL_loadbuffer(L, (char *) lmcf->init_worker_src.data, - lmcf->init_worker_src.len, "=init_worker_by_lua") - || ngx_stream_lua_do_call(log, L); + lmcf->init_worker_src.len, "=init_worker_by_lua"); return ngx_stream_lua_report(log, L, status, "init_worker_by_lua"); } @@ -345,8 +691,7 @@ ngx_stream_lua_init_worker_by_file(ngx_log_t *log, { int status; - status = luaL_loadfile(L, (char *) lmcf->init_worker_src.data) - || ngx_stream_lua_do_call(log, L); + status = luaL_loadfile(L, (char *) lmcf->init_worker_src.data); return ngx_stream_lua_report(log, L, status, "init_worker_by_lua_file"); } diff --git a/src/ngx_stream_lua_module.c b/src/ngx_stream_lua_module.c index 772e8c55..2470c196 100644 --- a/src/ngx_stream_lua_module.c +++ b/src/ngx_stream_lua_module.c @@ -240,6 +240,20 @@ static ngx_command_t ngx_stream_lua_cmds[] = { 0, (void *) ngx_stream_lua_init_worker_by_file }, + { ngx_string("lua_init_worker_timeout"), + NGX_STREAM_MAIN_CONF|NGX_CONF_TAKE1, + ngx_conf_set_msec_slot, + NGX_STREAM_MAIN_CONF_OFFSET, + offsetof(ngx_stream_lua_main_conf_t, init_worker_timeout), + NULL }, + + { ngx_string("lua_init_worker_abort_on_error"), + NGX_STREAM_MAIN_CONF|NGX_CONF_FLAG, + ngx_conf_set_flag_slot, + NGX_STREAM_MAIN_CONF_OFFSET, + offsetof(ngx_stream_lua_main_conf_t, init_worker_abort_on_error), + NULL }, + /* preread_by_lua_file rel/or/abs/path/to/script */ { ngx_string("preread_by_lua_file"), NGX_STREAM_MAIN_CONF|NGX_STREAM_SRV_CONF|NGX_CONF_TAKE1, @@ -796,6 +810,9 @@ ngx_stream_lua_create_main_conf(ngx_conf_t *cf) lmcf->pool = cf->pool; lmcf->max_pending_timers = NGX_CONF_UNSET; lmcf->max_running_timers = NGX_CONF_UNSET; + lmcf->init_worker_timeout = NGX_CONF_UNSET_MSEC; + lmcf->init_worker_abort_on_error = NGX_CONF_UNSET; + ngx_queue_init(&lmcf->deferred_timers); #if (NGX_PCRE) lmcf->regex_cache_max_entries = NGX_CONF_UNSET; lmcf->regex_match_limit = NGX_CONF_UNSET; @@ -843,6 +860,14 @@ ngx_stream_lua_init_main_conf(ngx_conf_t *cf, void *conf) lmcf->max_running_timers = 256; } + if (lmcf->init_worker_timeout == NGX_CONF_UNSET_MSEC) { + lmcf->init_worker_timeout = 0; + } + + if (lmcf->init_worker_abort_on_error == NGX_CONF_UNSET) { + lmcf->init_worker_abort_on_error = 0; + } + #if (NGX_STREAM_LUA_HAVE_SA_RESTART) if (lmcf->set_sa_restart == NGX_CONF_UNSET) { lmcf->set_sa_restart = 1; diff --git a/src/ngx_stream_lua_timer.c b/src/ngx_stream_lua_timer.c index 8dcc124e..d5439326 100644 --- a/src/ngx_stream_lua_timer.c +++ b/src/ngx_stream_lua_timer.c @@ -362,6 +362,19 @@ ngx_stream_lua_ngx_timer_helper(lua_State *L, int every) lmcf->pending_timers++; + if (ctx->context == NGX_STREAM_LUA_CONTEXT_INIT_WORKER) { + /* + * Freeze: user timers registered during init_worker are deferred + * until the runner flushes them at exit. ev->timer.key stores the + * absolute expiry (now + delay), matching ngx_add_timer's convention. + */ + ev->timer.key = ngx_current_msec + delay; + ngx_queue_insert_tail(&lmcf->deferred_timers, &ev->queue); + + lua_pushinteger(L, 1); + return 1; + } + ngx_add_timer(ev, delay); ngx_log_debug2(NGX_LOG_DEBUG_STREAM, ngx_cycle->log, 0, diff --git a/src/ngx_stream_lua_util.h b/src/ngx_stream_lua_util.h index e395ecc1..0aaf90e4 100644 --- a/src/ngx_stream_lua_util.h +++ b/src/ngx_stream_lua_util.h @@ -72,7 +72,7 @@ extern char ngx_stream_lua_headers_metatable_key; (str)->len = sizeof(text) - 1; (str)->data = (u_char *) text #endif -#define NGX_STREAM_LUA_CONTEXT_YIELDABLE (NGX_STREAM_LUA_CONTEXT_PREREAD \ +#define NGX_STREAM_LUA_CONTEXT_YIELDABLE (NGX_STREAM_LUA_CONTEXT_INIT_WORKER \ | NGX_STREAM_LUA_CONTEXT_CONTENT \ | NGX_STREAM_LUA_CONTEXT_TIMER \ | NGX_STREAM_LUA_CONTEXT_SSL_CLIENT_HELLO \ diff --git a/t/172-init-worker-cosocket.t b/t/172-init-worker-cosocket.t new file mode 100644 index 00000000..ad2a747c --- /dev/null +++ b/t/172-init-worker-cosocket.t @@ -0,0 +1,469 @@ +# vim:set ft= ts=4 sw=4 et fdm=marker: + +use Test::Nginx::Socket::Lua::Stream; + +$ENV{TEST_NGINX_MEMCACHED_PORT} ||= 11211; +$ENV{TEST_NGINX_RESOLVER} ||= '8.8.8.8'; + +master_on(); +repeat_each(1); + +plan tests => repeat_each() * (blocks() * 3 + 3); + +no_long_string(); + +# UDP echo server for the UDP cosocket test: memcached 1.6+ dropped the +# text-over-UDP protocol, so we echo datagrams ourselves +my $udp_echo_pid = fork(); +if (!$udp_echo_pid) { + require IO::Socket::INET; + my $srv = IO::Socket::INET->new( + LocalAddr => '127.0.0.1', LocalPort => 19849, Proto => 'udp') + or die "udp echo bind: $!"; + while (my $peer = $srv->recv(my $buf, 4096)) { + $srv->send($buf, 0, $peer); + } + exit 0; +} +END { kill 'KILL', $udp_echo_pid if $udp_echo_pid; } + +our $HtmlDir = html_dir; + +run_tests(); + +__DATA__ + + + +=== TEST 1: TCP cosocket connect/send/receive in init_worker_by_lua +--- stream_config eval +qq{ + lua_init_worker_timeout 10s; + init_worker_by_lua_block { + local sock = ngx.socket.tcp() + local ok, err = sock:connect("127.0.0.1", $ENV{TEST_NGINX_MEMCACHED_PORT}) + if not ok then + iw_result = "connect failed: " .. (err or "unknown") + return + end + + local bytes, err = sock:send("flush_all\\r\\n") + if not bytes then + iw_result = "send failed: " .. (err or "unknown") + return + end + + local line, err = sock:receive() + if not line then + iw_result = "receive failed: " .. (err or "unknown") + return + end + + sock:close() + iw_result = "received: " .. line + } +} +--- stream_server_config + content_by_lua_block { + ngx.say(iw_result or "no result") + } +--- stream_response +received: OK +--- timeout: 15 +--- no_error_log +[error] + + + +=== TEST 2: ngx.sleep in init_worker_by_lua +--- stream_config + lua_init_worker_timeout 5s; + init_worker_by_lua_block { + local t0 = ngx.now() + ngx.sleep(0.1) + local t1 = ngx.now() + ngx.sleep(0) + local t2 = ngx.now() + + iw_done = true + iw_slept = (t1 - t0) >= 0.09 + iw_zero_ok = (t2 - t1) < 1 + } +--- stream_server_config + content_by_lua_block { + ngx.say(iw_done and "done" or "not done") + ngx.say(iw_slept and "slept OK" or "sleep too short") + ngx.say(iw_zero_ok and "zero OK" or "zero hung") + } +--- stream_response +done +slept OK +zero OK +--- no_error_log +[error] + + + +=== TEST 3: semaphore wait in init_worker, posted by a spawned uthread +--- stream_config eval +qq{ + lua_init_worker_timeout 5s; + init_worker_by_lua_block { + local sema = require("ngx.semaphore").new(0) + + ngx.log(ngx.DEBUG, "iw: before spawn") + local child, serr = ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + local ok, err = sock:connect("127.0.0.1", + $ENV{TEST_NGINX_MEMCACHED_PORT}) + if not ok then + ngx.log(ngx.ERR, "child connect failed: ", err) + return + end + sock:send("flush_all\\r\\n") + sock:receive() + sock:close() + ngx.log(ngx.DEBUG, "iw: child posting") + sema:post(1) + end) + iw_spawn_ok = child and true or false + + ngx.log(ngx.DEBUG, "iw: before wait") + + local wok, werr = sema:wait(3) + + ngx.log(ngx.DEBUG, "iw: after wait") + iw_wait_ok = wok and true or false + iw_wait_err = werr or "none" + + local jok = ngx.thread.wait(child) + iw_join_ok = jok and true or false + } +} +--- stream_server_config + content_by_lua_block { + ngx.say(iw_spawn_ok and "spawn OK" or "spawn failed") + ngx.say(iw_wait_ok and "wait OK" or "wait failed: " + .. (iw_wait_err or "")) + ngx.say(iw_join_ok and "join OK" or "join failed") + } +--- stream_response +spawn OK +wait OK +join OK +--- log_level: debug +--- grep_error_log eval: qr/iw: [^,\n]*/ +--- grep_error_log_out +iw: before spawn +iw: before wait +iw: child posting +iw: after wait +--- no_error_log +[error] + + + +=== TEST 4: non-yielding init_worker code behaves unchanged +--- stream_config + init_worker_by_lua_block { + foo = ngx.md5("hello world") + } +--- stream_server_config + content_by_lua_block { + ngx.say("foo = ", foo) + } +--- stream_response +foo = 5eb63bbbe01eeed093cb22bb8f5acdc3 +--- no_error_log +[error] + + + +=== TEST 5: timer.at registered in init_worker fires only after init finishes +--- stream_config + lua_shared_dict iw_state 1m; + init_worker_by_lua_block { + local shdict = ngx.shared.iw_state + shdict:set("chunk_done", 0) + shdict:set("fires", 0) + + local ok, err = ngx.timer.at(0, function(premature) + if premature then + return + end + shdict:set("fires", (shdict:get("fires") or 0) + 1) + if shdict:get("chunk_done") ~= 1 then + shdict:set("bad_early", 1) + end + end) + shdict:set("timer_ok", ok and 1 or 0) + + ngx.sleep(0.2) + + shdict:set("chunk_done", 1) + } +--- stream_server_config + content_by_lua_block { + local shdict = ngx.shared.iw_state + for i = 1, 100 do + if (shdict:get("fires") or 0) >= 1 then + break + end + ngx.sleep(0.02) + end + ngx.say("timer_ok: " .. (shdict:get("timer_ok") == 1 and "OK" or "FAIL")) + ngx.say("fired: " .. ((shdict:get("fires") or 0) >= 1 and "OK" or "FAIL")) + ngx.say("no early fire: " .. (shdict:get("bad_early") and "FAIL" or "OK")) + } +--- stream_response +timer_ok: OK +fired: OK +no early fire: OK +--- timeout: 10 +--- no_error_log +[error] + + + +=== TEST 6: timer.every registered in init_worker fires only after init and renews +--- stream_config + lua_shared_dict iw_state 1m; + init_worker_by_lua_block { + local shdict = ngx.shared.iw_state + shdict:set("chunk_done", 0) + shdict:set("fires", 0) + + local tok, err = ngx.timer.every(0.05, function(premature) + if premature then + return + end + shdict:set("fires", shdict:get("fires") + 1) + if shdict:get("chunk_done") ~= 1 then + shdict:set("bad_early", 1) + end + end) + shdict:set("timer_ok", tok and 1 or 0) + + ngx.sleep(0.2) + + shdict:set("chunk_done", 1) + } +--- stream_server_config + content_by_lua_block { + local shdict = ngx.shared.iw_state + for i = 1, 100 do + if (shdict:get("fires") or 0) >= 3 then + break + end + ngx.sleep(0.05) + end + ngx.say("timer_ok: " .. (shdict:get("timer_ok") == 1 and "OK" or "FAIL")) + ngx.say("renewal: " .. ((shdict:get("fires") or 0) >= 3 and "OK" or "FAIL")) + ngx.say("no early fire: " .. (shdict:get("bad_early") and "FAIL" or "OK")) + } +--- stream_response +timer_ok: OK +renewal: OK +no early fire: OK +--- timeout: 10 +--- no_error_log +[error] + + + +=== TEST 7: ngx.thread.spawn + wait, child does a cosocket roundtrip +--- stream_config eval +qq{ + lua_init_worker_timeout 5s; + init_worker_by_lua_block { + local child, err = ngx.thread.spawn(function() + local sock = ngx.socket.tcp() + local ok, err = sock:connect("127.0.0.1", + $ENV{TEST_NGINX_MEMCACHED_PORT}) + if not ok then + return "connect failed: " .. (err or "?") + end + sock:send("flush_all\\r\\n") + local line = sock:receive() + sock:close() + return "child: " .. line + end) + iw_spawn_ok = child and true or false + local ok, res = ngx.thread.wait(child) + iw_wait_ok = ok and true or false + iw_child_ret = res or "none" + } +} +--- stream_server_config + content_by_lua_block { + ngx.say(iw_spawn_ok and "spawn OK" or "spawn failed") + ngx.say(iw_wait_ok and "wait OK" or "wait failed") + ngx.say(iw_child_ret) + } +--- stream_response +spawn OK +wait OK +child: OK +--- timeout: 15 +--- no_error_log +[error] + + + +=== TEST 8: UDP cosocket roundtrip in init_worker +--- stream_config + lua_init_worker_timeout 5s; + init_worker_by_lua_block { + local sock = ngx.socket.udp() + local ok, err = sock:setpeername("127.0.0.1", 19849) + if not ok then + iw_udp_result = "setpeername failed: " .. (err or "?") + return + end + sock:settimeout(2000) + local bytes, err = sock:send("ping") + if not bytes then + iw_udp_result = "send failed: " .. (err or "?") + return + end + local data, err = sock:receive() + if not data then + iw_udp_result = "receive failed: " .. (err or "?") + return + end + sock:close() + iw_udp_result = "received: " .. data + } +--- stream_server_config + content_by_lua_block { + ngx.say(iw_udp_result or "no result") + } +--- stream_response +received: ping +--- timeout: 10 +--- no_error_log +[error] + + + +=== TEST 9: cosocket connect via resolver in init_worker +--- stream_config + resolver $TEST_NGINX_RESOLVER ipv6=off; + lua_init_worker_timeout 10s; + init_worker_by_lua_block { + local sock = ngx.socket.tcp() + sock:settimeout(5000) + local ok, err = sock:connect("openresty.org", 80) + if not ok then + iw_resolver_result = "connect failed: " .. (err or "?") + return + end + sock:close() + iw_resolver_result = "connected" + } +--- stream_server_config + content_by_lua_block { + ngx.say(iw_resolver_result or "no result") + } +--- stream_response +connected +--- timeout: 15 +--- no_error_log +[error] + + + +=== TEST 10: SSL cosocket handshake in init_worker +--- stream_config + resolver $TEST_NGINX_RESOLVER ipv6=off; + lua_init_worker_timeout 10s; + init_worker_by_lua_block { + local sock = ngx.socket.tcp() + sock:settimeout(5000) + local ok, err = sock:connect("openresty.org", 443) + if not ok then + iw_ssl_result = "connect failed: " .. (err or "?") + return + end + + local session, err = sock:sslhandshake(nil, "openresty.org", false) + if not session then + iw_ssl_result = "ssl handshake failed: " .. (err or "?") + return + end + + sock:close() + iw_ssl_result = "handshake OK" + } +--- stream_server_config + content_by_lua_block { + ngx.say(iw_ssl_result or "no result") + } +--- stream_response +handshake OK +--- timeout: 15 +--- no_error_log +[error] + + + +=== TEST 11: stalled remote read times out via lua_init_worker_timeout, worker starts +--- stream_config eval +qq{ + lua_init_worker_timeout 500ms; + init_worker_by_lua_block { + local sock = ngx.socket.tcp() + local ok, err = sock:connect("127.0.0.1", $ENV{TEST_NGINX_MEMCACHED_PORT}) + if not ok then + return + end + sock:send("get") -- incomplete command: memcached waits forever + sock:receive() -- hangs here; the 500ms budget aborts mid-read + } +} +--- stream_server_config + content_by_lua_block { + ngx.say("still serving") + } +--- stream_response +still serving +--- timeout: 10 +--- error_log +init_worker_by_lua* timed out + + + +=== TEST 12: compile error keeps the legacy log prefix, worker still starts +--- stream_config + init_worker_by_lua_block { + local x = + -- nothing after the "=": syntax error at load time + } +--- stream_server_config + content_by_lua_block { + ngx.say("still serving") + } +--- stream_response +still serving +--- error_log +init_worker_by_lua error: +--- no_error_log +lua run thread returned: + + + +=== TEST 13: runtime error with default abort_on_error off, worker still starts +--- stream_config + init_worker_by_lua_block { + error("boom") -- runtime error before any yield + } +--- stream_server_config + content_by_lua_block { + ngx.say("still serving") + } +--- stream_response +still serving +--- error_log +lua entry thread aborted: +--- no_error_log +init_worker_by_lua error: diff --git a/t/173-init-worker-accept.t b/t/173-init-worker-accept.t new file mode 100644 index 00000000..3384eb5a --- /dev/null +++ b/t/173-init-worker-accept.t @@ -0,0 +1,71 @@ +# vim:set ft= ts=4 sw=4 et fdm=marker: + +use Test::Nginx::Socket::Lua::Stream; + +master_on(); +repeat_each(1); + +plan tests => repeat_each() * (blocks() * 3 + 2); + +no_long_string(); + +run_tests(); + +__DATA__ + + + +=== TEST 1: accept deferred, pump stays idle (1 worker, level-triggered) +--- stream_config + init_worker_by_lua_block { + ngx.log(ngx.DEBUG, "iw: pump begin") + ngx.sleep(1) + ngx.log(ngx.DEBUG, "iw: pump done") + iw_done = true + } +--- stream_server_config + content_by_lua_block { + ngx.say(iw_done and "response ok" or "init_worker unfinished") + } +--- stream_response +response ok +--- timeout: 15 +--- log_level: debug +--- grep_error_log eval: qr/iw: pump [a-z]+/ +--- grep_error_log_out +iw: pump begin +iw: pump done +--- no_error_log +[error] +exited on signal + + + +=== TEST 2: cosocket in init_worker, then server accepts normally +--- stream_config eval +qq{ + lua_init_worker_timeout 10s; + init_worker_by_lua_block { + local sock = ngx.socket.tcp() + local ok, err = sock:connect("127.0.0.1", $ENV{TEST_NGINX_MEMCACHED_PORT}) + if not ok then + iw_result = "connect failed: " .. (err or "?") + return + end + sock:send("flush_all\\r\\n") + local line = sock:receive() + sock:close() + iw_result = "cosocket: " .. line + } +} +--- stream_server_config + content_by_lua_block { + ngx.say(iw_result or "no result") + ngx.say("accept ok") + } +--- stream_response +cosocket: OK +accept ok +--- timeout: 15 +--- no_error_log +[error] diff --git a/t/174-init-worker-accept-workers.t b/t/174-init-worker-accept-workers.t new file mode 100644 index 00000000..622a728d --- /dev/null +++ b/t/174-init-worker-accept-workers.t @@ -0,0 +1,63 @@ +# vim:set ft= ts=4 sw=4 et fdm=marker: + +use Test::Nginx::Socket::Lua::Stream; + +master_on(); +repeat_each(1); +workers(2); + +plan tests => repeat_each() * (blocks() * 3 + 1); + +no_long_string(); + +run_tests(); + +__DATA__ + + + +=== TEST 1: accept deferred, no crash, no spin (2 workers) +--- stream_config + init_worker_by_lua_block { + ngx.sleep(1) + iw_done = true + } +--- stream_server_config + content_by_lua_block { + ngx.say(iw_done and "response ok" or "init_worker unfinished") + } +--- stream_response +response ok +--- timeout: 15 +--- no_error_log +[error] +exited on signal + + + +=== TEST 2: cosocket in init_worker under 2 workers +--- stream_config eval +qq{ + lua_init_worker_timeout 10s; + init_worker_by_lua_block { + local sock = ngx.socket.tcp() + local ok, err = sock:connect("127.0.0.1", $ENV{TEST_NGINX_MEMCACHED_PORT}) + if not ok then + iw_result = "connect failed: " .. (err or "?") + return + end + sock:send("flush_all\\r\\n") + local line = sock:receive() + sock:close() + iw_result = "cosocket: " .. line + } +} +--- stream_server_config + content_by_lua_block { + ngx.say(iw_result or "no result") + } +--- stream_response +cosocket: OK +--- timeout: 15 +--- no_error_log +[error] diff --git a/t/175-init-worker-abort.t b/t/175-init-worker-abort.t new file mode 100644 index 00000000..131dd95f --- /dev/null +++ b/t/175-init-worker-abort.t @@ -0,0 +1,58 @@ +# vim:set ft= ts=4 sw=4 et fdm=marker: + +use Test::Nginx::Socket::Lua::Stream; + +$ENV{TEST_NGINX_MEMCACHED_PORT} ||= 11211; + +# single-process mode (no master_on): an init_process failure exits +# the whole nginx with code 2, which --- must_die asserts + +repeat_each(1); + +$Test::Nginx::Util::DaemonEnabled = 'off'; + +plan tests => repeat_each() * (blocks() * 2); + +no_long_string(); +run_tests(); + +__DATA__ + +=== TEST 1: abort_on_error on + runtime error before yield exits with code 2 +--- stream_config + lua_init_worker_abort_on_error on; + init_worker_by_lua_block { + error("boom") -- runtime error before any yield + } +--- stream_server_config + content_by_lua_block { + ngx.say("never reached") + } +--- must_die: 2 +--- error_log +lua entry thread aborted: + + + +=== TEST 2: abort_on_error on + init_worker timeout exits with code 2 +--- stream_config eval +qq{ + lua_init_worker_abort_on_error on; + lua_init_worker_timeout 500ms; + init_worker_by_lua_block { + local sock = ngx.socket.tcp() + local ok, err = sock:connect("127.0.0.1", $ENV{TEST_NGINX_MEMCACHED_PORT}) + if not ok then + return + end + sock:send("get") -- incomplete command: memcached waits forever + sock:receive() -- hangs; the 500ms budget aborts mid-read + } +} +--- stream_server_config + content_by_lua_block { + ngx.say("never reached") + } +--- must_die: 2 +--- error_log +init_worker_by_lua* timed out diff --git a/t/176-init-worker-accept-reuseport.t b/t/176-init-worker-accept-reuseport.t new file mode 100644 index 00000000..6304fbb3 --- /dev/null +++ b/t/176-init-worker-accept-reuseport.t @@ -0,0 +1,69 @@ +# vim:set ft= ts=4 sw=4 et fdm=marker: + +use Test::Nginx::Socket::Lua::Stream; + +plan(skip_all => "TEST_NGINX_REUSE_PORT=1 is required " + . "(TEST_NGINX_REUSE_PORT=1 prove t/176-init-worker-accept-reuseport.t)") + unless $ENV{TEST_NGINX_REUSE_PORT}; + +master_on(); +repeat_each(1); +workers(2); + +plan tests => repeat_each() * (blocks() * 3 + 1); + +no_long_string(); + +our $cpu_clock = <<'_EOC_'; + local ffi = require("ffi") + ffi.cdef[[ + typedef long time_t; + typedef struct timeval { time_t tv_sec; time_t tv_usec; } timeval; + typedef struct { + struct timeval ru_utime; + struct timeval ru_stime; + long pad[14]; + } rusage_t; + int getrusage(int who, rusage_t *usage); + ]] + local ru = ffi.new("rusage_t") + function cpu_us() + ffi.C.getrusage(0, ru) -- 0 = RUSAGE_SELF + return tonumber(ru.ru_utime.tv_sec) * 1000000 + + tonumber(ru.ru_utime.tv_usec) + + tonumber(ru.ru_stime.tv_sec) * 1000000 + + tonumber(ru.ru_stime.tv_usec) + end +_EOC_ + +run_tests(); + +__DATA__ + +=== TEST 1: reuseport re-arm branch (2 workers, pump stays idle) +--- stream_config eval +qq{ + init_by_lua_block { +$main::cpu_clock + } + init_worker_by_lua_block { + local c0 = cpu_us() + ngx.sleep(1) + local c1 = cpu_us() + iw_cpu_us = c1 - c0 + } +} +--- steam_listen_option: reuseport +--- stream_server_config + content_by_lua_block { + ngx.say("response ok") + ngx.say(iw_cpu_us and iw_cpu_us < 50000 + and "cpu idle OK" or "cpu busy: " .. (iw_cpu_us or "?")) + } +--- stream_response +response ok +cpu idle OK +--- timeout: 15 +--- no_error_log +[error] +exited on signal