在主体过滤器中的 ngx_thread_task_post 之后，nginx 任务完成处理程序无法响应

Posted 2023-03-15

【中文标题】在主体过滤器中的 ngx_thread_task_post 之后，nginx 任务完成处理程序无法响应

【英文标题】：nginx task completion handler cannot respond after ngx_thread_task_post in a body filter

【发布时间】：2020-11-09 15:07:30

【问题描述】：

我正在开发一个启用多线程的 nginx (1.19.0) body filter 模块（--with-threads 使 NGINX 能够使用线程池。有关详细信息，请参阅 NGINX 博客上的Thread Pools in NGINX Boost Performance 9x!。），其目标是在上游服务器的响应中保存acess_token。

我提到了development guide - Threads、How to make Nginx wait for a thread pool task和nginx HTTP module with Thread Pools and Tasks，

我的代码 sn-p 如下：

typedef struct 
    int status;
    cJSON *oauth2_rsp;
    ngx_http_request_t *req;
    ngx_chain_t *chain;
 redis_thread_ctx_t;

/* This function is executed in a separate thread */
static void redis_thread_func(void *data, ngx_log_t *log) 
    ngx_logd("SAM_DEBUG: redis_thread_func");
    redis_thread_ctx_t *ctx = data;
    cJSON *oauth2_access_token = cJSON_GetObjectItemCaseSensitive(ctx->oauth2_rsp, OAUTH2_PARAM_NAME_ACCESS_TOKEN);
    cJSON *oauth2_token_type = cJSON_GetObjectItemCaseSensitive(ctx->oauth2_rsp, OAUTH2_PARAM_NAME_TOKEN_TYPE);
    cJSON *oauth2_expires_in = cJSON_GetObjectItemCaseSensitive(ctx->oauth2_rsp, OAUTH2_PARAM_NAME_EXPIRES_IN);
    if (0 == cache_token(ctx->req, oauth2_access_token->valuestring,
                         cJSON_IsString(oauth2_token_type) ? oauth2_token_type->valuestring : "Bear",
                         cJSON_IsNumber(oauth2_expires_in) ? oauth2_expires_in->valueint : 3600)) 
        ctx->status = NGX_HTTP_OK;
     else 
        ngx_log_error(NGX_LOG_ERR, log, 0, "cache_token failed");
    
    ngx_logd("SAM_DEBUG: after cache_token");
    cJSON_free(ctx->oauth2_rsp);
    ngx_logd("SAM_DEBUG: after cJSON_free");


/*
 * The task completion handler executes on the main event loop, and is pretty straightforward: Mark the background
 * processing complete, and call the nginx HTTP function to resume processing of the request.
 */
static void redis_thread_completion(ngx_event_t *ev) 
    redis_thread_ctx_t *ctx = ev->data;
    ngx_http_request_t *req = ctx->req;
    ngx_connection_t *con = req->connection;
    ngx_log_t *log = con->log;
    ngx_http_set_log_request(log, req);
    ngx_logd("SAM_DEBUG: redis_thread_completion: \"%V?%V\"", &req->uri, &req->args);

    req->main->blocked--;
    req->aio = 0;
    //ngx_http_handler(req);
    ngx_http_next_body_filter(req, ctx->chain);
    //ngx_http_finalize_request(req, NGX_DONE);
    ngx_logd("SAM_DEBUG: after ngx_http_next_body_filter");


//https://serverfault.com/questions/480352/modify-data-being-proxied-by-nginx-on-the-fly
static ngx_int_t ngx_http_pep_body_filter(ngx_http_request_t *req, ngx_chain_t *chain) 
    // ... omitted for brevity
    cJSON *oauth2_rsp_json = NULL;
    //#if (NGX_THREADS)
    ngx_thread_task_t *task = ngx_thread_task_alloc(req->pool, sizeof(redis_thread_ctx_t));
    if (NULL == task) 
        return NGX_ERROR;
    
    ngx_logd("SAM_DEBUG: after ngx_thread_task_alloc");
    redis_thread_ctx_t *redis_ctx = task->ctx;
    redis_ctx->status = NGX_HTTP_BAD_GATEWAY;
    redis_ctx->req = req;
    redis_ctx->oauth2_rsp = oauth2_rsp_json;
    redis_ctx->chain = chain;

    task->handler = redis_thread_func;
    task->event.handler = redis_thread_completion;
    task->event.data = redis_ctx;
    ngx_http_core_loc_conf_t *clcf = ngx_http_get_module_loc_conf(req, ngx_http_core_module);
    //subrequests=51, count=1, blocked=1, aio=0
    ngx_logd("SAM_DEBUG: subrequests=%d, count=%d, blocked=%d, aio=%d", req->subrequests, req->count, req->blocked, req->aio);
    if (NGX_OK != ngx_thread_task_post(clcf->thread_pool, task)) 
        req->main->blocked--;
        cJSON_free(oauth2_rsp_json);
        ngx_log_error(NGX_LOG_ERR, log, 0, "ngx_thread_task_post failed");
        return NGX_ERROR; //NGX_HTTP_INTERNAL_SERVER_ERROR
    
    //Note: increment `req->main->blocked` so nginx won't finalize request (req)
    req->main->blocked++;
    req->aio = 1;
    ngx_logd("SAM_DEBUG: after ngx_thread_task_post");
//#else
#if defined(USE_REDIS_TO_CACHE_TOKEN) && (NGX_THREADS)
    return NGX_OK; //NGX_AGAIN
#else
    return ngx_http_next_body_filter ? ngx_http_next_body_filter(req, chain) : NGX_OK;
#endif

经过测试，不幸的是，我发现客户端无法收到响应。邮递员向我显示“错误：套接字挂断”或wireshak 中没有相应的HTTP 响应包。另外error.log如下，

2020/07/20 18:38:55 [debug] 461#461: *3 redis_thread_completion|772|SAM_DEBUG: after ngx_http_next_body_filter
2020/07/20 18:38:55 [debug] 461#461: timer delta: 3
2020/07/20 18:38:55 [debug] 461#461: worker cycle
2020/07/20 18:38:55 [debug] 461#461: epoll timer: 59997


2020/07/20 18:39:55 [debug] 461#461: timer delta: 59998
2020/07/20 18:39:55 [debug] 461#461: *3 event timer del: 3: 18416868
2020/07/20 18:39:55 [debug] 461#461: *3 http empty handler
2020/07/20 18:39:55 [debug] 461#461: worker cycle
2020/07/20 18:39:55 [debug] 461#461: epoll timer: 5002
2020/07/20 18:40:00 [debug] 461#461: timer delta: 5003
2020/07/20 18:40:00 [debug] 461#461: *3 event timer del: 3: 18421871
2020/07/20 18:40:00 [debug] 461#461: *3 http keepalive handler
2020/07/20 18:40:00 [debug] 461#461: *3 close http connection: 3
2020/07/20 18:40:00 [debug] 461#461: *3 reusable connection: 0
2020/07/20 18:40:00 [debug] 461#461: *3 free: 0000000000000000
2020/07/20 18:40:00 [debug] 461#461: *3 free: 00007FFFEC420BF0, unused: 136
2020/07/20 18:40:00 [debug] 461#461: worker cycle
2020/07/20 18:40:00 [debug] 461#461: epoll timer: -1

我哪里出错了？

【参考方案1】：

看了nginx\src\http\ngx_http_upstream.c中static void ngx_http_upstream_thread_event_handler(ngx_event_t *ev)的代码后，我把static void redis_thread_completion(ngx_event_t *ev)中的ngx_http_next_body_filter(req, ctx->chain);改成了req->write_event_handler(req);和

static ngx_int_t ngx_http_pep_body_filter(ngx_http_request_t *req, ngx_chain_t *chain) 
    // ... omitted for brevity
    return ngx_http_next_body_filter(req, chain);

因此，nginx 可以将 HTTP 响应发送到客户端。