ngx_lua中的協程調度(六)之ngx_http_lua_run_posted_thread

ngx_http_lua_run_posted_thread

這個函數主要是爲了ngx.thread.spawn的處理，ngx.thread.spawn生成新的"light thread"，這個"light thread"運行優先級比它的父協程高，會優先運行，父協程被迫暫停。"light thread"運行結束或者yield後，再由ngx_http_lua_run_posted_threads去運行父協程。

ngx.thread.spawn中建立"light thread"後， 調用ngx_http_lua_post_thread。

if (ngx_http_lua_post_thread(r, ctx, ctx->cur_co_ctx) != NGX_OK) {
        return luaL_error(L, "no memory");
    }

ngx_http_lua_post_thread函數將父協程放在了ctx->posted_threads指向的鏈表中。

ngx_int_t
ngx_http_lua_post_thread(ngx_http_request_t *r, ngx_http_lua_ctx_t *ctx,
    ngx_http_lua_co_ctx_t *coctx)
{
    ngx_http_lua_posted_thread_t  **p;
    ngx_http_lua_posted_thread_t   *pt;

    pt = ngx_palloc(r->pool, sizeof(ngx_http_lua_posted_thread_t));
    if (pt == NULL) {
        return NGX_ERROR;
    }

    pt->co_ctx = coctx;
    pt->next = NULL;

    for (p = &ctx->posted_threads; *p; p = &(*p)->next) { /* void */ }

    *p = pt;

    return NGX_OK;
}

ngx_http_lua_run_posted_threads從ctx->posted_threads指向的鏈表中依次取出每一個元素，調用ngx_http_lua_run_thread運行。

/* this is for callers other than the content handler */
ngx_int_t
ngx_http_lua_run_posted_threads(ngx_connection_t *c, lua_State *L,
    ngx_http_request_t *r, ngx_http_lua_ctx_t *ctx)
{
    ngx_int_t                        rc;
    ngx_http_lua_posted_thread_t    *pt;

    for ( ;; ) {
        if (c->destroyed) {
            return NGX_DONE;
        }

        pt = ctx->posted_threads;
        if (pt == NULL) {
            return NGX_DONE;
        }

        ctx->posted_threads = pt->next;

        ngx_http_lua_probe_run_posted_thread(r, pt->co_ctx->co,
                                             (int) pt->co_ctx->co_status);

        if (pt->co_ctx->co_status != NGX_HTTP_LUA_CO_RUNNING) {
            continue;
        }

        ctx->cur_co_ctx = pt->co_ctx;

        rc = ngx_http_lua_run_thread(L, r, ctx, 0);

        if (rc == NGX_AGAIN) {
            continue;
        }

        if (rc == NGX_DONE) {
            ngx_http_lua_finalize_request(r, NGX_DONE);
            continue;
        }

        /* rc == NGX_ERROR || rc >= NGX_OK */

        if (ctx->entered_content_phase) {
            ngx_http_lua_finalize_request(r, rc);
        }

        return rc;
    }

    /* impossible to reach here */
}

ngx_http_lua_run_thread使用方式

以lua-nginx-module的Access階段的處理爲例，實際的執行工做由ngx_http_lua_access_by_chunk函數中實現。 以下面的代碼，調用ngx_http_lua_run_thread後根據返回值繼續處理。

static ngx_int_t
ngx_http_lua_access_by_chunk(lua_State *L, ngx_http_request_t *r)
{
    /* 此處省去了建立協程的部分，只關注協程的運行  */


    rc = ngx_http_lua_run_thread(L, r, ctx, 0);

    dd("returned %d", (int) rc);

    if (rc == NGX_ERROR || rc > NGX_OK) {
        return rc;
    }

    c = r->connection;

    if (rc == NGX_AGAIN) {
        rc = ngx_http_lua_run_posted_threads(c, L, r, ctx);

        if (rc == NGX_ERROR || rc == NGX_DONE || rc > NGX_OK) {
            return rc;
        }

        if (rc != NGX_OK) {
            return NGX_DECLINED;
        }

    } else if (rc == NGX_DONE) {
        ngx_http_lua_finalize_request(r, NGX_DONE);

        rc = ngx_http_lua_run_posted_threads(c, L, r, ctx);

        if (rc == NGX_ERROR || rc == NGX_DONE || rc > NGX_OK) {
            return rc;
        }

        if (rc != NGX_OK) {
            return NGX_DECLINED;
        }
    }

#if 1
    if (rc == NGX_OK) {
        if (r->header_sent) {
            dd("header already sent");

            /* response header was already generated in access_by_lua*,
             * so it is no longer safe to proceed to later phases
             * which may generate responses again */

            if (!ctx->eof) {
                dd("eof not yet sent");

                rc = ngx_http_lua_send_chain_link(r, ctx, NULL
                                                  /* indicate last_buf */);
                if (rc == NGX_ERROR || rc > NGX_OK) {
                    return rc;
                }
            }

            return NGX_HTTP_OK;
        }

        return NGX_OK;
    }
#endif

    return NGX_DECLINED;
}

ngx_http_lua_run_thread的返回值

函數ngx_http_lua_run_thread的返回值可分爲下面幾種

• NGX_OK
• NGX_AGAIN
• NGX_DONE
• NGX_ERROR: 執行出錯
• 大於200: 響應的HTTP狀態碼

按照Nginx的處理規則，返回NGX_ERROR或大於200的HTTP狀態碼時，將會無條件結束當前請求的處理。 返回NGX_OK代表當前階段處理完成，此時只須要調用ngx_http_lua_send_chain_link發送響應便可。重點關注的是NGX_AGAIN和NGX_DONE這兩個。返回這兩個值時都要調用ngx_http_lua_run_posted_thread來處理。

NGX_AGAIN

ngx_http_lua_run_thread何時會返回NGX_AGAIN?

• 1. ngx.sleep或ngx.socket等致使協程的yield
• 2. 調用ngx.thread致使當前請求對應的一個父協程和一個或多個"light thread"沒有所有退出。 因爲狀況2的存在，須要調用ngx_http_lua_run_posted_thread進行處理。

NGX_DONE

在Nginx中，NGX_DONE表示對當前請求的處理已經告一段落了，可是請求尚未處理完成，以後的工做會有其餘的模塊進行。主要出如今三個地方

• 調用ngx_http_read_client_request_body讀取請求包體時，因爲從socket讀取數據是異步的，會返回NGX_DONE。讀取數據後對請求的處理由設置的回調函數執行
• 調用ngx_http_internal_redirect執行了內部跳轉
• 建立子請求後，等待子請求完成後繼續處理

在ngx_http_request_t中有一個做爲引用計數的成員count。每次調用ngx_http_finalize_requet(r, NGX_DONE)時會將r的引用計數減一，減爲0時纔會真正結束當前請求。與此對應的模塊返回NGX_DONE時都會有r->count++的操做。

在函數ngx_http_lua_access_by_chunk中當ngx_http_lua_run_thread返回NGX_DONE時(相比於返回值爲NGX_AGAIN的狀況)增長了一次ngx_http_finalize_request(r, NGX_DONE)的操做，就是爲了將r的引用計數減一。

若是這裏調用ngx_http_finalize_request(r, NGX_DONE)致使r的引用計數爲0,將請求結束了，此時c->destory爲true，再調用ngx_http_lua_run_posted_thread會直接返回。