bugfix: coroutine.wrap: propagate errors to the parent coroutine.

Author: thibaultcha

src/ngx_http_lua_common.h

@@ -431,6 +431,13 @@ struct ngx_http_lua_co_ctx_s {
                                                         the ngx.thread.spawn()
                                                         call */
     unsigned                 sem_resume_status:1;
+
+    unsigned                 is_wrap; /* set when creating coroutines via
+                                         coroutine.wrap */
+
+    unsigned                 propagate_error; /* set when propagating an error
+                                                 from a coroutine to its
+                                                 parent */
 };
 
 

src/ngx_http_lua_coroutine.c

@@ -25,6 +25,7 @@
 
 
 static int ngx_http_lua_coroutine_create(lua_State *L);
+static int ngx_http_lua_coroutine_wrap(lua_State *L);
 static int ngx_http_lua_coroutine_resume(lua_State *L);
 static int ngx_http_lua_coroutine_yield(lua_State *L);
 static int ngx_http_lua_coroutine_status(lua_State *L);
@@ -62,6 +63,45 @@ ngx_http_lua_coroutine_create(lua_State *L)
 }
 
 
+static int
+ngx_http_lua_coroutine_wrap_runner(lua_State *L)
+{
+    /* retrieve closure and insert it at the bottom of
+     * the stack for coroutine.resume() */
+    lua_pushvalue(L, lua_upvalueindex(1));
+    lua_insert(L, 1);
+
+    return ngx_http_lua_coroutine_resume(L);
+}
+
+
+static int
+ngx_http_lua_coroutine_wrap(lua_State *L)
+{
+    ngx_http_request_t          *r;
+    ngx_http_lua_ctx_t          *ctx;
+    ngx_http_lua_co_ctx_t       *coctx = NULL;
+
+    r = ngx_http_lua_get_req(L);
+    if (r == NULL) {
+        return luaL_error(L, "no request found");
+    }
+
+    ctx = ngx_http_get_module_ctx(r, ngx_http_lua_module);
+    if (ctx == NULL) {
+        return luaL_error(L, "no request ctx found");
+    }
+
+    ngx_http_lua_coroutine_create_helper(L, r, ctx, &coctx);
+
+    coctx->is_wrap = 1;
+
+    lua_pushcclosure(L, ngx_http_lua_coroutine_wrap_runner, 1);
+
+    return 1;
+}
+
+
 int
 ngx_http_lua_coroutine_create_helper(lua_State *L, ngx_http_request_t *r,
     ngx_http_lua_ctx_t *ctx, ngx_http_lua_co_ctx_t **pcoctx)
@@ -246,7 +286,7 @@ ngx_http_lua_inject_coroutine_api(ngx_log_t *log, lua_State *L)
     int         rc;
 
     /* new coroutine table */
-    lua_createtable(L, 0 /* narr */, 14 /* nrec */);
+    lua_createtable(L, 0 /* narr */, 16 /* nrec */);
 
     /* get old coroutine table */
     lua_getglobal(L, "coroutine");
@@ -258,6 +298,9 @@ ngx_http_lua_inject_coroutine_api(ngx_log_t *log, lua_State *L)
     lua_getfield(L, -1, "create");
     lua_setfield(L, -3, "_create");
 
+    lua_getfield(L, -1, "wrap");
+    lua_setfield(L, -3, "_wrap");
+
     lua_getfield(L, -1, "resume");
     lua_setfield(L, -3, "_resume");
 
@@ -273,6 +316,9 @@ ngx_http_lua_inject_coroutine_api(ngx_log_t *log, lua_State *L)
     lua_pushcfunction(L, ngx_http_lua_coroutine_create);
     lua_setfield(L, -2, "__create");
 
+    lua_pushcfunction(L, ngx_http_lua_coroutine_wrap);
+    lua_setfield(L, -2, "__wrap");
+
     lua_pushcfunction(L, ngx_http_lua_coroutine_resume);
     lua_setfield(L, -2, "__resume");
 
@@ -287,7 +333,7 @@ ngx_http_lua_inject_coroutine_api(ngx_log_t *log, lua_State *L)
     /* inject coroutine APIs */
     {
         const char buf[] =
-            "local keys = {'create', 'yield', 'resume', 'status'}\n"
+            "local keys = {'create', 'yield', 'resume', 'status', 'wrap'}\n"
             "local getfenv = getfenv\n"
             "for _, key in ipairs(keys) do\n"
                "local std = coroutine['_' .. key]\n"
@@ -305,24 +351,19 @@ ngx_http_lua_inject_coroutine_api(ngx_log_t *log, lua_State *L)
                     "return std(...)\n"
                 "end\n"
             "end\n"
-            "local create, resume = coroutine.create, coroutine.resume\n"
-            "coroutine.wrap = function(f)\n"
-               "local co = create(f)\n"
-               "return function(...) return select(2, resume(co, ...)) end\n"
-            "end\n"
             "package.loaded.coroutine = coroutine";
 
 #if 0
             "debug.sethook(function () collectgarbage() end, 'rl', 1)"
 #endif
             ;
 
-        rc = luaL_loadbuffer(L, buf, sizeof(buf) - 1, "=coroutine.wrap");
+        rc = luaL_loadbuffer(L, buf, sizeof(buf) - 1, "=coroutine_api");
     }
 
     if (rc != 0) {
         ngx_log_error(NGX_LOG_ERR, log, 0,
-                      "failed to load Lua code for coroutine.wrap(): %i: %s",
+                      "failed to load Lua code for coroutine_api: %i: %s",
                       rc, lua_tostring(L, -1));
 
         lua_pop(L, 1);
@@ -332,7 +373,7 @@ ngx_http_lua_inject_coroutine_api(ngx_log_t *log, lua_State *L)
     rc = lua_pcall(L, 0, 0, 0);
     if (rc != 0) {
         ngx_log_error(NGX_LOG_ERR, log, 0,
-                      "failed to run the Lua code for coroutine.wrap(): %i: %s",
+                      "failed to run the Lua code for coroutine_api: %i: %s",
                       rc, lua_tostring(L, -1));
         lua_pop(L, 1);
     }

src/ngx_http_lua_util.c

@@ -950,7 +950,7 @@ ngx_http_lua_run_thread(lua_State *L, ngx_http_request_t *r,
     int                      rv, success = 1;
     lua_State               *next_co;
     lua_State               *old_co;
-    const char              *err, *msg, *trace;
+    const char              *err = NULL, *msg = NULL, *trace = NULL;
     ngx_int_t                rc;
 #if (NGX_PCRE)
     ngx_pool_t              *old_pool = NULL;
@@ -988,6 +988,8 @@ ngx_http_lua_run_thread(lua_State *L, ngx_http_request_t *r,
             dd("ctx: %p", ctx);
             dd("cur co: %p", ctx->cur_co_ctx->co);
             dd("cur co status: %d", ctx->cur_co_ctx->co_status);
+            dd("cur co has propagated error: %d",
+               ctx->cur_co_ctx->propagate_error);
 
             orig_coctx = ctx->cur_co_ctx;
 
@@ -1004,6 +1006,11 @@ ngx_http_lua_run_thread(lua_State *L, ngx_http_request_t *r,
             }
 #endif
 
+            if (ctx->cur_co_ctx->propagate_error) {
+                ctx->cur_co_ctx->propagate_error = 0;
+                goto propagate_error;
+            }
+
             ngx_http_lua_assert(orig_coctx->co_top + nrets
                                 == lua_gettop(orig_coctx->co));
 
@@ -1148,12 +1155,6 @@ ngx_http_lua_run_thread(lua_State *L, ngx_http_request_t *r,
                     next_coctx = ctx->cur_co_ctx->parent_co_ctx;
                     next_co = next_coctx->co;
 
-                    /*
-                     * prepare return values for coroutine.resume
-                     * (true plus any retvals)
-                     */
-                    lua_pushboolean(next_co, 1);
-
                     if (nrets) {
                         dd("moving %d return values to next co", nrets);
                         lua_xmove(ctx->cur_co_ctx->co, next_co, nrets);
@@ -1162,7 +1163,15 @@ ngx_http_lua_run_thread(lua_State *L, ngx_http_request_t *r,
 #endif
                     }
 
-                    nrets++;  /* add the true boolean value */
+                    if (!ctx->cur_co_ctx->is_wrap) {
+                        /*
+                         * prepare return values for coroutine.resume
+                         * (true plus any retvals)
+                         */
+                        lua_pushboolean(next_co, 1);
+                        lua_insert(next_co, 1);
+                        nrets++;  /* add the true boolean value */
+                    }
 
                     ctx->cur_co_ctx = next_coctx;
 
@@ -1273,12 +1282,6 @@ ngx_http_lua_run_thread(lua_State *L, ngx_http_request_t *r,
 
                 next_co = next_coctx->co;
 
-                /*
-                 * ended successful, coroutine.resume returns true plus
-                 * any return values
-                 */
-                lua_pushboolean(next_co, success);
-
                 if (nrets) {
                     lua_xmove(ctx->cur_co_ctx->co, next_co, nrets);
                 }
@@ -1288,7 +1291,16 @@ ngx_http_lua_run_thread(lua_State *L, ngx_http_request_t *r,
                     ctx->uthreads--;
                 }
 
-                nrets++;
+                if (!ctx->cur_co_ctx->is_wrap) {
+                    /*
+                     * ended successful, coroutine.resume returns true plus
+                     * any return values
+                     */
+                    lua_pushboolean(next_co, success);
+                    lua_insert(next_co, 1);
+                    nrets++;
+                }
+
                 ctx->cur_co_ctx = next_coctx;
 
                 ngx_http_lua_probe_info("set parent running");
@@ -1344,6 +1356,10 @@ ngx_http_lua_run_thread(lua_State *L, ngx_http_request_t *r,
                                           ctx->cur_co_ctx);
             trace = lua_tostring(L, -1);
 
+propagate_error:
+
+            ngx_http_lua_assert(err != NULL && msg != NULL && trace != NULL);
+
             if (ctx->cur_co_ctx->is_uthread) {
 
                 ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,
@@ -1433,16 +1449,25 @@ ngx_http_lua_run_thread(lua_State *L, ngx_http_request_t *r,
 
             next_coctx->co_status = NGX_HTTP_LUA_CO_RUNNING;
 
+            ctx->cur_co_ctx = next_coctx;
+
+            if (orig_coctx->is_wrap) {
+                /*
+                 * coroutine.wrap propagates errors
+                 * to the parent
+                 */
+                next_coctx->propagate_error = 1;
+                continue;
+            }
+
             /*
              * ended with error, coroutine.resume returns false plus
              * err msg
              */
             lua_pushboolean(next_co, 0);
-            lua_xmove(ctx->cur_co_ctx->co, next_co, 1);
+            lua_xmove(orig_coctx->co, next_co, 1);
             nrets = 2;
 
-            ctx->cur_co_ctx = next_coctx;
-
             ngx_log_error(NGX_LOG_ERR, r->connection->log, 0,
                           "lua coroutine: %s: %s\n%s", err, msg, trace);
 

t/062-count.t

@@ -391,7 +391,7 @@ probe process("$LIBLUA_PATH").function("rehashtab") {
 --- stap_out2
 3
 --- response_body
-coroutine: 14
+coroutine: 16
 --- no_error_log
 [error]