bugfix: buffer bloat and CPU 100% when download large file was filtered by body_filter_by_lua.

When http body was change by lua script, ngx_http_lua_body_filter_param_set will
mark the origin buf as consumed with code "cl->buf->pos = cl->buf->last". Because
the buf was marked as consumed, the function ngx_output_chain will continue to
write data to the output filter.

When kernel socket send buffer is full, write to the socket will return EAGAIN.
And the output buf will be cached in output chain. The output chain becomes longer and longer.
Function ngx_chain_update_chains do an linear iteration of the output chain to
put the new buf to the end of the chain. At the end, there are thounsands of bufs
in the output chain which cause the CPU utilization become 100% and memory continue to grow up.

nginx.conf:
    location /download {
        alias download;
        header_filter_by_lua_block {
            ngx.header.content_length = nil
        }

        body_filter_by_lua_block {
            ngx.arg[1]=ngx.arg[1]
        }
    }

create a 2G file with the following cmd:
  mkdir -p /usr/local/openresty/nginx/download
  dd if=/dev/zero of=/usr/local/openresty/nginx/download/2G count=4096000

reproduce with the following cmd:
  curl -o /dev/null http://127.0.0.1/download/2G

typical call stacks:

42706f: ngx_chain_update_chains[14]
47e993: ngx_http_chunked_body_filter[14]
483024: ngx_http_gzip_body_filter[14]
4864d3: ngx_http_ssi_body_filter[14]
489668: ngx_http_charset_body_filter[14]
48b86c: ngx_http_addition_body_filter[14]
48bf9c: ngx_http_gunzip_body_filter[14]
48d75e: ngx_http_trailers_filter[14]
4fb63c: ngx_http_lua_capture_body_filter[14]
509cb9: ngx_http_lua_body_filter[14]
42739e: ngx_output_chain[14]
48e1aa: ngx_http_copy_filter[14]
45ea2b: ngx_http_output_filter[14]
496569: ngx_http_static_handler[14]
45ec6e: ngx_http_core_content_phase[14]
459a25: ngx_http_core_run_phases[14]
463ed3: ngx_http_process_request[14]
4643ff: ngx_http_process_request_headers[14]
4647a6: ngx_http_process_request_line[14]
44c1fe: ngx_epoll_process_events[14]
4431c3: ngx_process_events_and_timers[14]
44a65a: ngx_worker_process_cycle[14]
449074: ngx_spawn_process[14]
44aaac: ngx_start_worker_processes[14]
44b213: ngx_master_process_cycle[14]
422df2: main[14]
23493: __libc_start_main[2]
422e5e: _start[14]

42706f: ngx_chain_update_chains[14]
509ce7: ngx_http_lua_body_filter[14]
42739e: ngx_output_chain[14]
48e1aa: ngx_http_copy_filter[14]
45ea2b: ngx_http_output_filter[14]
496569: ngx_http_static_handler[14]
45ec6e: ngx_http_core_content_phase[14]
459a25: ngx_http_core_run_phases[14]
463ed3: ngx_http_process_request[14]
4643ff: ngx_http_process_request_headers[14]
4647a6: ngx_http_process_request_line[14]
44c1fe: ngx_epoll_process_events[14]
4431c3: ngx_process_events_and_timers[14]
44a65a: ngx_worker_process_cycle[14]
449074: ngx_spawn_process[14]
44aaac: ngx_start_worker_processes[14]
44b213: ngx_master_process_cycle[14]
422df2: main[14]
23493: __libc_start_main[2]
422e5e: _start[14]

47d51f: ngx_http_write_filter[14]
47e973: ngx_http_chunked_body_filter[14]
483024: ngx_http_gzip_body_filter[14]
4864d3: ngx_http_ssi_body_filter[14]
489668: ngx_http_charset_body_filter[14]
48b86c: ngx_http_addition_body_filter[14]
48bf9c: ngx_http_gunzip_body_filter[14]
48d75e: ngx_http_trailers_filter[14]
4fb63c: ngx_http_lua_capture_body_filter[14]
509cb9: ngx_http_lua_body_filter[14]
42739e: ngx_output_chain[14]
48e1aa: ngx_http_copy_filter[14]
45ea2b: ngx_http_output_filter[14]
496569: ngx_http_static_handler[14]
45ec6e: ngx_http_core_content_phase[14]
459a25: ngx_http_core_run_phases[14]
463ed3: ngx_http_process_request[14]
4643ff: ngx_http_process_request_headers[14]
4647a6: ngx_http_process_request_line[14]
44c1fe: ngx_epoll_process_events[14]
4431c3: ngx_process_events_and_timers[14]
44a65a: ngx_worker_process_cycle[14]
449074: ngx_spawn_process[14]
44aaac: ngx_start_worker_processes[14]
44b213: ngx_master_process_cycle[14]
422df2: main[14]
23493: __libc_start_main[2]
422e5e: _start[14]

Author: zhuizhuhaomeng

.travis.yml

@@ -1,5 +1,9 @@
 dist: xenial
 
+branches:
+  only:
+    - "master"
+
 os: linux
 
 language: c

src/ngx_http_lua_bodyfilterby.c

@@ -235,18 +235,15 @@ ngx_http_lua_body_filter(ngx_http_request_t *r, ngx_chain_t *in)
     uint16_t                     old_context;
     ngx_http_cleanup_t          *cln;
     ngx_chain_t                 *out;
+    ngx_chain_t                 *cl, *ln;
     ngx_http_lua_main_conf_t    *lmcf;
 
     ngx_log_debug1(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,
                    "lua body filter for user lua code, uri \"%V\"", &r->uri);
 
-    if (in == NULL) {
-        return ngx_http_next_body_filter(r, in);
-    }
-
     llcf = ngx_http_get_module_loc_conf(r, ngx_http_lua_module);
 
-    if (llcf->body_filter_handler == NULL) {
+    if (llcf->body_filter_handler == NULL || r->header_only) {
         dd("no body filter handler found");
         return ngx_http_next_body_filter(r, in);
     }
@@ -269,7 +266,48 @@ ngx_http_lua_body_filter(ngx_http_request_t *r, ngx_chain_t *in)
             in->buf->file_pos = in->buf->file_last;
         }
 
-        return NGX_OK;
+        in = NULL;
+    }
+
+    if (in != NULL) {
+        if (ngx_chain_add_copy(r->pool, &ctx->filter_in_bufs, in) != NGX_OK) {
+            return NGX_ERROR;
+        }
+    }
+
+    if (ctx->filter_busy_bufs != NULL
+        && (r->connection->buffered
+            & (NGX_HTTP_LOWLEVEL_BUFFERED | NGX_LOWLEVEL_BUFFERED)))
+    {
+        /* Socket write buffer was full on last write.
+         * Try to write the remain data, if still can not write
+         * do not execute body_filter_by_lua otherwise the `in` chain will be
+         * replaced by new content from lua and buf of `in` mark as consumed.
+         * And then ngx_output_chain will call the filter chain again which
+         * make all the data cached in the memory and long ngx_chain_t link
+         * cause CPU 100%.
+         */
+        rc = ngx_http_next_body_filter(r, NULL);
+
+        if (rc == NGX_ERROR) {
+            return rc;
+        }
+
+        out = NULL;
+        ngx_chain_update_chains(r->pool,
+                                &ctx->free_bufs, &ctx->filter_busy_bufs, &out,
+                                (ngx_buf_tag_t) &ngx_http_lua_module);
+        if (rc != NGX_OK) {
+            if (ctx->filter_busy_bufs != NULL
+                && (r->connection->buffered
+                    & (NGX_HTTP_LOWLEVEL_BUFFERED | NGX_LOWLEVEL_BUFFERED))) {
+                ngx_log_debug0(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,
+                               "waiting body filter busy buffer to be sent");
+                return NGX_AGAIN;
+            }
+        }
+
+        /* continue to process bufs in ctx->filter_in_bufs */
     }
 
     if (ctx->cleanup == NULL) {
@@ -286,6 +324,9 @@ ngx_http_lua_body_filter(ngx_http_request_t *r, ngx_chain_t *in)
     old_context = ctx->context;
     ctx->context = NGX_HTTP_LUA_CONTEXT_BODY_FILTER;
 
+    in = ctx->filter_in_bufs;
+    ctx->filter_in_bufs = NULL;
+
     dd("calling body filter handler");
     rc = llcf->body_filter_handler(r, in);
 
@@ -298,26 +339,36 @@ ngx_http_lua_body_filter(ngx_http_request_t *r, ngx_chain_t *in)
     }
 
     lmcf = ngx_http_get_module_main_conf(r, ngx_http_lua_module);
+
+    /* lmcf->body_filter_chain is the new buffer chain if body_filter_by_lua*
+     * set new body content via ngx.arg[1] = new_content otherwise it is the
+     * original `in` buffer chain.
+     */
     out = lmcf->body_filter_chain;
 
-    if (in == out) {
-        return ngx_http_next_body_filter(r, in);
-    }
+    if (in != out) {
+        /* content of body was replaced in ngx_http_lua_body_filter_param_set
+         * and the buffers was marked as consumed.
+         */
+        for (cl = in; cl != NULL; cl = ln) {
+            ln = cl->next;
+            ngx_free_chain(r->pool, cl);
+        }
 
-    if (out == NULL) {
-        /* do not forward NULL to the next filters because the input is
-         * not NULL */
-        return NGX_OK;
+        if (out == NULL) {
+            /* do not forward NULL to the next filters because the input is
+             * not NULL */
+            return NGX_OK;
+        }
     }
 
-    /* in != out */
     rc = ngx_http_next_body_filter(r, out);
     if (rc == NGX_ERROR) {
         return NGX_ERROR;
     }
 
     ngx_chain_update_chains(r->pool,
-                            &ctx->free_bufs, &ctx->busy_bufs, &out,
+                            &ctx->free_bufs, &ctx->filter_busy_bufs, &out,
                             (ngx_buf_tag_t) &ngx_http_lua_module);
 
     return rc;

src/ngx_http_lua_common.h

@@ -548,6 +548,9 @@ typedef struct ngx_http_lua_ctx_s {
     ngx_chain_t             *busy_bufs;
     ngx_chain_t             *free_recv_bufs;
 
+    ngx_chain_t             *filter_in_bufs;  /* for the body filter */
+    ngx_chain_t             *filter_busy_bufs;  /* for the body filter */
+
     ngx_http_cleanup_pt     *cleanup;
 
     ngx_http_cleanup_t      *free_cleanup; /* free list of cleanup records */

src/ngx_http_lua_output.c

@@ -244,6 +244,11 @@ ngx_http_lua_ngx_echo(lua_State *L, unsigned newline)
         return 2;
     }
 
+    ngx_log_debug2(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,
+                   "%s has %sbusy bufs",
+                   newline ? "lua say response" : "lua print response",
+                   ctx->busy_bufs != NULL ? "" : "not ");
+
     dd("downstream write: %d, buf len: %d", (int) rc,
        (int) (b->last - b->pos));
 

t/082-body-filter-2.t

@@ -0,0 +1,57 @@
+# vim:set ft= ts=4 sw=4 et fdm=marker:
+
+BEGIN {
+    $ENV{TEST_NGINX_POSTPONE_OUTPUT} = 1;
+    $ENV{TEST_NGINX_EVENT_TYPE} = 'poll';
+    $ENV{MOCKEAGAIN}='world'
+}
+
+use Test::Nginx::Socket::Lua;
+
+#worker_connections(1014);
+#master_process_enabled(1);
+#log_level('warn');
+
+log_level('debug');
+
+repeat_each(2);
+
+plan tests => repeat_each() * (blocks() * 5);
+
+#no_diff();
+no_long_string();
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: check ctx->busy_bufs
+--- config
+    location /t {
+        postpone_output 1;
+        content_by_lua_block {
+            for i = 1, 5 do
+                ngx.say(i, ": Hello World!")
+            end
+        }
+
+        body_filter_by_lua_block {
+            ngx.arg[1] = ngx.arg[1]
+        }
+    }
+--- request
+GET /t
+--- response_body
+1: Hello World!
+2: Hello World!
+3: Hello World!
+4: Hello World!
+5: Hello World!
+
+--- error_log
+waiting body filter busy buffer to be sent
+lua say response has busy bufs
+--- no_error_log
+[error]
+
+