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

Author: thibaultcha

src/ngx_http_lua_common.h

@@ -482,6 +482,13 @@ struct ngx_http_lua_co_ctx_s {
                                                         the ngx.thread.spawn()
                                                         call */
     unsigned                 sem_resume_status:1;
+
+    unsigned                 is_wrap:1; /* set when creating coroutines via
+                                           coroutine.wrap */
+
+    unsigned                 propagate_error:1; /* 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)
@@ -250,7 +290,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");
@@ -262,6 +302,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");
 
@@ -277,6 +320,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");
 
@@ -291,7 +337,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"
 #ifdef OPENRESTY_LUAJIT
             "local get_req = require 'thread.exdata'\n"
 #else
@@ -321,24 +367,18 @@ 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";
-
+            "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);
@@ -348,7 +388,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

@@ -1018,8 +1018,6 @@ ngx_http_lua_run_thread(lua_State *L, ngx_http_request_t *r,
     /* set Lua VM panic handler */
     lua_atpanic(L, ngx_http_lua_atpanic);
 
-    dd("ctx = %p", ctx);
-
     NGX_LUA_EXCEPTION_TRY {
 
         if (ctx->cur_co_ctx->thread_spawn_yielded) {
@@ -1031,19 +1029,15 @@ ngx_http_lua_run_thread(lua_State *L, ngx_http_request_t *r,
 
         for ( ;; ) {
 
-            dd("calling lua_resume: vm %p, nret %d", ctx->cur_co_ctx->co,
-               (int) nrets);
+            dd("ctx: %p, co: %p, co status: %d, co is_wrap: %d",
+               ctx, ctx->cur_co_ctx->co, ctx->cur_co_ctx->co_status,
+               ctx->cur_co_ctx->is_wrap);
 
 #if (NGX_PCRE)
             /* XXX: work-around to nginx regex subsystem */
             old_pool = ngx_http_lua_pcre_malloc_init(r->pool);
 #endif
 
-            /*  run code */
-            dd("ctx: %p", ctx);
-            dd("cur co: %p", ctx->cur_co_ctx->co);
-            dd("cur co status: %d", ctx->cur_co_ctx->co_status);
-
             orig_coctx = ctx->cur_co_ctx;
 
 #ifdef NGX_LUA_USE_ASSERT
@@ -1055,10 +1049,19 @@ ngx_http_lua_run_thread(lua_State *L, ngx_http_request_t *r,
 
 #if DDEBUG
             if (lua_gettop(orig_coctx->co) > 0) {
-                dd("top elem: %s", luaL_typename(orig_coctx->co, -1));
+                dd("co top elem: %s", luaL_typename(orig_coctx->co, -1));
+            }
+
+            if (orig_coctx->propagate_error) {
+                dd("co propagate_error: %d", orig_coctx->propagate_error);
             }
 #endif
 
+            if (orig_coctx->propagate_error) {
+                orig_coctx->propagate_error = 0;
+                goto propagate_error;
+            }
+
             ngx_http_lua_assert(orig_coctx->co_top + nrets
                                 == lua_gettop(orig_coctx->co));
 
@@ -1203,12 +1206,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);
@@ -1217,7 +1214,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;
 
@@ -1328,12 +1333,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);
                 }
@@ -1343,7 +1342,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");
@@ -1399,6 +1407,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,
@@ -1488,16 +1500,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]