保存响应缓冲区并稍后在 NGINX 输出正文过滤器中发送它们

【问题标题】：Save response buffers and send them later in NGINX output body filter保存响应缓冲区并稍后在 NGINX 输出正文过滤器中发送它们
【发布时间】：2020-12-23 11:14:45
【问题描述】：

我知道在NGINX中可能会多次调用输出（响应）体过滤器，所以我想使用累积缓冲区来保存整个响应体并通过修改第二个输入参数每次写入一个空缓冲区链直到收到最后一个缓冲区（参见下面的代码 sn-p）。

结果发现客户端可以收到reponse header（chunked response），但是收不到reponse body。

Server: nginx/1.19.3
Connection: keep-alive
Transfer-Encoding: chunked

为什么或如何做（保存它们并在以后发送它们，即使在调用ngx_http_output_body_filter_by_test 之后）？

// Use the placement new constructor to all a constructor on an already allocated memory
// buffer = malloc(/*...*/);
// object = new (buffer) MyClass(/*...*/);
// But later you have to call the destructor explicitly too :
// object->~MyClass();
// free(buffer);
// https://en.cppreference.com/w/cpp/language/new
typedef struct ngx_http_test_ctx_s {
    std::string buffer;
} ngx_http_test_ctx_t;

std::string cumulative_buffer;

//@param chain nullable, e.g. rc = ngx_http_output_filter(r, NULL) in nginx\src\http\ngx_http_request.c#ngx_http_writer 
ngx_int_t ngx_http_output_body_filter_by_test(ngx_http_request_t *req, ngx_chain_t *chain) {
    // TODO Auto-generated method stub
    logdf("%.*s|%d,%d", ARGS_NGX_STR(req->unparsed_uri), req->headers_out.content_length_n, req->chunked);
    int last = 0;
    for (ngx_chain_t *cl = chain; cl; cl = cl->next) {
        ngx_buf_t *buf = cl->buf;
        // 0/0 if ngx_http_send_special(req, NGX_HTTP_LAST);
        logdf("chain@%p %d%d %" PRIdMAX " / %" PRIdMAX, chain, buf->last_buf, buf->last_in_chain, buf->last - buf->pos, buf->end - buf->start);
        cumulative_buffer.append(buf->pos, buf->last);
        // ngx_chain_update_chains ngx_event_pipe_write_to_downstream
        buf->pos = buf->last; // avoid being write_to_downstream
        buf->flush = 0; // avoid being write_to_downstream
        if (buf->last_buf || buf->last_in_chain) {
            buf->last_in_chain = 0; // avoid being considered as the last buffer by NGINX
            buf->last_buf = 0; //  avoid being considered as the last buffer by NGINX
            last = 1;
        }
    }
    if (last) {
        logdf("cumulative_buffer=%s", cumulative_buffer.c_str());
        // do something use the entire reponse body
        // auto newbuf = reinterpret_cast<ngx_buf_t*>(ngx_alloc_buf(req->pool));
        auto newbuf = ngx_create_temp_buf(req->pool, cumulative_buffer.length());
        newbuf->last = ngx_cpymem(newbuf->last, cumulative_buffer.data(), cumulative_buffer.length());
        newbuf->last_buf = 1;
        newbuf->last_in_chain = 1;
#ifdef ENABLE_NEW_CHAIN_IN_REPONSE_BODY_FILTER
        auto newchain = reinterpret_cast<ngx_chain_t*>(ngx_alloc_chain_link(req->pool));
        newchain->buf = newbuf;
        newchain->next = NULL;
        ngx_int_t rc = ngx_http_next_output_body_filter(req, newchain);
        req->write_event_handler(req); //req->write_event_handler is ngx_http_writer
        return rc;
#else
        // ngx_free_chain(req->pool, chain);
        chain->buf = newbuf;
        chain->next = NULL;

        return ngx_http_next_output_body_filter(req, chain);
        //req->buffered |= NGX_HTTP_SUB_BUFFERED;
#endif
    }
    return ngx_http_next_output_body_filter(req, chain);
    //if (chain) {
    //  req->buffered &= ~NGX_HTTP_SUB_BUFFERED;
    //}

    // Delay the call to the ngx_http_next_output_body_filter function because response may need to be revised.
    // return NGX_OK;
}

static void ngx_http_test_cleanup(void *data) {
    auto ctx = reinterpret_cast<ngx_http_em_subrequest_enforcer_ctx_t*>(data);
    ctx->~ngx_http_test_ctx_t();
    // delete ctx->xx;
    logdf("ctx@%p", ctx);
}

将第二个return ngx_http_next_output_body_filter(req, chain); 替换为return NGX_OK; 后，客户端可以收到响应正文。但是，有时，last 总是 0（即没有 buf 其 buf-&gt;last_buf || buf-&gt;last_in_chain 是 true）（内容长度与分块？）。

ngx_int_t ngx_http_output_body_filter_by_test(ngx_http_request_t *req, ngx_chain_t *chain) {
    //// TODO Auto-generated method stub
    logdf("%.*s|content_length_n=%" PRIdMAX ", chunked=%d", ARGS_NGX_STR(req->unparsed_uri), req->headers_out.content_length_n, req->chunked);
    if (req->upstream) {
        logdf("upstream: length = %" PRIdMAX  ", buffer@%p.size=%" PRIdMAX ", bytes_received=%" PRIdMAX ", bytes_sent=%" PRIdMAX
            ", response_length=%" PRIdMAX , req->upstream->length // pending bytes to send to the downstream client
            , req->upstream->buffer.pos, req->upstream->buffer.last - req->upstream->buffer.pos
            , req->upstream->state->bytes_received, req->upstream->state->bytes_sent
            , req->upstream->state->response_length); // req->upstream->pipe->preread_size
    }

    if (NGX_STR_STARTS_WITH(req->unparsed_uri, "/stream/")) {
#if 0
        for (ngx_chain_t *cl = chain; cl; cl = cl->next) {
            ngx_buf_t *buf = cl->buf;
            logdf("chain@%p buf@%p %d%d %" PRIdMAX "/%" PRIdMAX, chain, buf->pos,
                buf->last_buf, buf->last_in_chain, buf->last - buf->pos, buf->end - buf->start);
        }
        return ngx_http_next_output_body_filter(req, chain);
#endif
        if (0 == (intptr_t)ngx_http_get_module_ctx(req, ngx_http_em_module)) {
            ngx_http_set_ctx(req, (void*)1, ngx_http_em_module);
            cumulative_buffer.clear();
        }
        int last = 0;
        for (ngx_chain_t *cl = chain; cl; cl = cl->next) {
            ngx_buf_t *buf = cl->buf;
            // 0/0 if ngx_http_send_special(req, NGX_HTTP_LAST);
            logdf("chain@%p buf@%p %d%d %" PRIdMAX "/%" PRIdMAX, chain, buf->pos,
                buf->last_buf, buf->last_in_chain, buf->last - buf->pos, buf->end - buf->start);
            cumulative_buffer.append(buf->pos, buf->last);
            if (buf->last_buf || buf->last_in_chain) {
                //buf->last_in_chain = 0; // avoid being considered as the last buffer by NGINX
                //buf->last_buf = 0; //  avoid being considered as the last buffer by NGINX
                last = 1;
            }
            //// ngx_chain_update_chains ngx_event_pipe_write_to_downstream
            //buf->pos = buf->last; // avoid being write_to_downstream
            //buf->flush = 0; // avoid being write_to_downstream
        }
        if (last) {
            logdf("cumulative_buffer(length=%" PRIdMAX ")=%.16s", cumulative_buffer.length(), cumulative_buffer.c_str());
            // do something use the entire reponse body
            // auto newbuf = reinterpret_cast<ngx_buf_t*>(ngx_alloc_buf(req->pool));
            auto len = ngx_min(1024, cumulative_buffer.length());
            auto newbuf = ngx_create_temp_buf(req->pool, len);
            newbuf->last = ngx_cpymem(newbuf->last, cumulative_buffer.data(), len);
            newbuf->last_buf = 1;
            newbuf->last_in_chain = 1;
#ifdef ENABLE_NEW_CHAIN_IN_REPONSE_BODY_FILTER
            auto newchain = reinterpret_cast<ngx_chain_t*>(ngx_alloc_chain_link(req->pool));
            newchain->buf = newbuf;
            newchain->next = NULL;
            ngx_int_t rc = ngx_http_next_output_body_filter(req, newchain);
            req->write_event_handler(req); //req->write_event_handler is ngx_http_writer
            return rc;
#else
            // ngx_free_chain(req->pool, chain);
            chain->buf = newbuf;
            chain->next = NULL;

            return ngx_http_next_output_body_filter(req, chain);
            //req->buffered |= NGX_HTTP_SUB_BUFFERED;
#endif
        }
        return NGX_OK; // ngx_http_next_output_body_filter(req, chain);
    }
}

nginx.conf sn-p

        # /stream/{n} Stream n JSON responses, max n is 100, 59.6 kB, 58690 bytes = 57.314453125 kB
        location /stream/ {
            proxy_buffer_size   4k;
            proxy_buffers   4 8k;
            proxy_busy_buffers_size 16k;
            # proxy_pass http://httpbin.org;
            proxy_pass http://127.0.0.1:8080;
        }

我发现 NGINX 在SSL_ERROR_WANT_WRITE 之后没有重试（“SSL_write: -1”和“SSL_get_error: 3”）。

>   NGINX.exe!ngx_ssl_write(ngx_connection_s * c, unsigned char * data, unsigned __int64 size) Line 2576    C
    NGINX.exe!ngx_ssl_send_chain(ngx_connection_s * c, ngx_chain_s * in, __int64 limit) Line 2495   C
    NGINX.exe!ngx_http_write_filter(ngx_http_request_s * r, ngx_chain_s * in) Line 294  C
    NGINX.exe!ngx_http_chunked_body_filter(ngx_http_request_s * r, ngx_chain_s * in) Line 219   C
    NGINX.exe!ngx_http_gzip_body_filter(ngx_http_request_s * r, ngx_chain_s * in) Line 308  C
    NGINX.exe!ngx_http_postpone_filter(ngx_http_request_s * r, ngx_chain_s * in) Line 91    C
    NGINX.exe!ngx_http_ssi_body_filter(ngx_http_request_s * r, ngx_chain_s * in) Line 413   C
    NGINX.exe!ngx_http_charset_body_filter(ngx_http_request_s * r, ngx_chain_s * in) Line 557   C
    NGINX.exe!ngx_http_trailers_filter(ngx_http_request_s * r, ngx_chain_s * in) Line 264   C
    NGINX.exe!ngx_output_chain(ngx_output_chain_ctx_s * ctx, ngx_chain_s * in) Line 74  C
    NGINX.exe!ngx_http_copy_filter(ngx_http_request_s * r, ngx_chain_s * in) Line 152   C
    NGINX.exe!ngx_http_range_body_filter(ngx_http_request_s * r, ngx_chain_s * in) Line 635 C
    ngx_http_test_module.dll!ngx_http_output_body_filter_by_test(ngx_http_request_s * req, ngx_chain_s * chain) Line 1148   C++
    NGINX.exe!ngx_http_output_filter(ngx_http_request_s * r, ngx_chain_s * in) Line 1846    C
    NGINX.exe!ngx_http_send_special(ngx_http_request_s * r, unsigned __int64 flags) Line 3515   C
    NGINX.exe!ngx_http_upstream_finalize_request(ngx_http_request_s * r, ngx_http_upstream_s * u, __int64 rc) Line 4487 C
    NGINX.exe!ngx_http_upstream_process_request(ngx_http_request_s * r, ngx_http_upstream_s * u) Line 4051  C
    NGINX.exe!ngx_http_upstream_process_upstream(ngx_http_request_s * r, ngx_http_upstream_s * u) Line 3963 C
    NGINX.exe!ngx_http_upstream_handler(ngx_event_s * ev) Line 1286 C
    NGINX.exe!ngx_event_process_posted(ngx_cycle_s * cycle, ngx_queue_s * posted) Line 35   C
    NGINX.exe!ngx_process_events_and_timers(ngx_cycle_s * cycle) Line 265   C
    NGINX.exe!ngx_worker_thread(void * data) Line 795   C

我预计（在断点的情况下来自error.log），

[debug] 10512#20076: *1 reusable connection: 0
[debug] 10512#20076: *1 http upstream temp fd: -1
[debug] 10512#20076: *1 http output filter "/stream/21?"
[debug] 10512#20076: *1 malloc: 0000024099030A50:307242
[debug] 10512#20076: *1 http copy filter: "/stream/21?"
[debug] 10512#20076: *1 http postpone filter "/stream/21?" 000000B5F25FEDE8
[debug] 10512#20076: *1 http chunk: 307242
[debug] 10512#20076: *1 write old buf t:1 f:0 000002409902B468, pos 000002409902B468, size: 156 file: 0, size: 0
[debug] 10512#20076: *1 write new buf t:1 f:0 000002409902CB78, pos 000002409902CB78, size: 7 file: 0, size: 0
[debug] 10512#20076: *1 write new buf t:1 f:0 0000024099030A50, pos 0000024099030A50, size: 307242 file: 0, size: 0
[debug] 10512#20076: *1 write new buf t:0 f:0 0000000000000000, pos 00007FF78CC708F8, size: 7 file: 0, size: 0
[debug] 10512#20076: *1 http write filter: l:1 f:0 s:307412
[debug] 10512#20076: *1 http write filter limit 0
[debug] 10512#20076: *1 malloc: 0000024098E6EB50:512
[debug] 10512#20076: *1 malloc: 0000024098E67EC0:16384
[debug] 10512#20076: *1 SSL buf copy: 156
[debug] 10512#20076: *1 SSL buf copy: 7
[debug] 10512#20076: *1 SSL buf copy: 16221
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 16384
[debug] 10512#20076: *1 SSL to write: 16384
[debug] 10512#20076: *1 SSL_write: 16384
[debug] 10512#20076: *1 SSL buf copy: 12493
[debug] 10512#20076: *1 SSL buf copy: 7
[debug] 10512#20076: *1 SSL to write: 12500
[debug] 10512#20076: *1 SSL_write: 12500
[debug] 10512#20076: *1 http write filter 0000000000000000
[debug] 10512#20076: *1 http copy filter: 0 "/stream/21?"
[debug] 10512#20076: *1 http finalize request: 0, "/stream/21?" a:1, c:1
[debug] 10512#20076: *1 set http keepalive handler
[debug] 10512#20076: *1 http close request
[debug] 10512#20076: *1 http log handler
[debug] 10512#20076: *1 free: 000002409902EA10
[debug] 10512#20076: *1 free: 0000024099030A50
[debug] 10512#20076: *1 free: 0000000000000000
[debug] 10512#20076: *1 free: 0000000000000000
[debug] 10512#20076: *1 free: 0000000000000000
[debug] 10512#20076: *1 free: 00000240990298B0, unused: 2
[debug] 10512#20076: *1 free: 000002409902A8F0, unused: 12
[debug] 10512#20076: *1 free: 000002409902C970, unused: 3279
[debug] 10512#20076: *1 free: 0000024098E6E710
[debug] 10512#20076: *1 hc free: 0000000000000000
[debug] 10512#20076: *1 hc busy: 0000000000000000 0
[debug] 10512#20076: *1 free: 0000024098E67EC0
[debug] 10512#20076: *1 reusable connection: 1
[debug] 10512#20076: *1 event timer add: 576: 65000:1608876752566
[debug] 10512#20076: worker cycle

而不是（真正运行时没有任何调试断点），

[debug] 10512#20076: *3 reusable connection: 0
[debug] 10512#20076: *3 http upstream temp fd: -1
[debug] 10512#20076: *3 http output filter "/stream/22?"
[debug] 10512#20076: *3 malloc: 0000024099030A50:307242
[debug] 10512#20076: *3 http copy filter: "/stream/22?"
[debug] 10512#20076: *3 http postpone filter "/stream/22?" 000000B5F25FEDE8
[debug] 10512#20076: *3 http chunk: 307242
[debug] 10512#20076: *3 write old buf t:1 f:0 0000024098E0C878, pos 0000024098E0C878, size: 156 file: 0, size: 0
[debug] 10512#20076: *3 write new buf t:1 f:0 0000024098E6E188, pos 0000024098E6E188, size: 7 file: 0, size: 0
[debug] 10512#20076: *3 write new buf t:1 f:0 0000024099030A50, pos 0000024099030A50, size: 307242 file: 0, size: 0
[debug] 10512#20076: *3 write new buf t:0 f:0 0000000000000000, pos 00007FF78CC708F8, size: 7 file: 0, size: 0
[debug] 10512#20076: *3 http write filter: l:1 f:0 s:307412
[debug] 10512#20076: *3 http write filter limit 0
[debug] 10512#20076: *3 malloc: 0000024098DD79C0:512
[debug] 10512#20076: *3 malloc: 0000024098E67EC0:16384
[debug] 10512#20076: *3 SSL buf copy: 156
[debug] 10512#20076: *3 SSL buf copy: 7
[debug] 10512#20076: *3 SSL buf copy: 16221
[debug] 10512#20076: *3 SSL to write: 16384
[debug] 10512#20076: *3 SSL_write: 16384
[debug] 10512#20076: *3 SSL buf copy: 16384
[debug] 10512#20076: *3 SSL to write: 16384
[debug] 10512#20076: *3 SSL_write: 16384
[debug] 10512#20076: *3 SSL buf copy: 16384
[debug] 10512#20076: *3 SSL to write: 16384
[debug] 10512#20076: *3 SSL_write: 16384
[debug] 10512#20076: *3 SSL buf copy: 16384
[debug] 10512#20076: *3 SSL to write: 16384
[debug] 10512#20076: *3 SSL_write: 16384
[debug] 10512#20076: *3 SSL buf copy: 16384
[debug] 10512#20076: *3 SSL to write: 16384
[debug] 10512#20076: *3 SSL_write: -1
[debug] 10512#20076: *3 SSL_get_error: 3
[debug] 10512#20076: *3 http write filter 0000024098E6E20C
[debug] 10512#20076: *3 http copy filter: -2 "/stream/22?"
[debug] 10512#20076: *3 http finalize request: -2, "/stream/22?" a:1, c:1
[debug] 10512#20076: *3 event timer add: 576: 60000:1608876912668
[debug] 10512#20076: *3 select add event fd:576 ev:1
[debug] 10512#20076: worker cycle

参考文献

NGINX 参考资料

【问题讨论】：

  • 你在构建什么，一个 nginx 模块？它的目标是什么？稍微介绍一下就好了。
  • nginx 模块。将链式缓冲区合并到一个连续的内存（std::string），然后找到两个关键字之间的内容并将其替换为新内容。

标签： c++ c nginx


【解决方案1】：

我已经找到了解决方案，通过在过滤器函数中的其他代码前面添加，让 NGINX 再次将剩余响应内容发送到客户端（因为 NGINX 只在下一个过滤器中写入一部分，然后套接字变为不可写） .

现在看来

ngx_int_t ngx_http_output_body_filter_by_test(ngx_http_request_t *req, ngx_chain_t *chain) {
    if (NULL == chain) {
        return ngx_http_next_output_body_filter(req, chain);
    }
    int last = 0;
    for (ngx_chain_t *cl = chain; cl; cl = cl->next) {
        ngx_buf_t *buf = cl->buf;
        cumulative_buffer.append(buf->pos, buf->last);
        if (buf->last_buf || buf->last_in_chain) {
            last = 1;
        }
        // ngx_chain_update_chains ngx_event_pipe_write_to_downstream
        buf->pos = buf->last; // avoid being write_to_downstream
        buf->flush = 0; // avoid being write_to_downstream
    }
    if (last) {
        // TODO modify chain using cumulative_buffer
        return ngx_http_next_output_body_filter(req, chain);
    }
    return NGX_OK; // ngx_http_next_output_body_filter(req, chain);
}

【讨论】：

  • 如果响应足够大，你不会用完缓冲区吗？
猜你喜欢
  • 1970-01-01
  • 2013-07-30
  • 1970-01-01
  • 2021-07-14
  • 1970-01-01
  • 1970-01-01
  • 2015-11-04
  • 2012-01-19
  • 1970-01-01
相关资源
最近更新 更多
热门标签
Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode