diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..9a4fe4b --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,51 @@ +name: CI + +on: + push: + branches: [ master ] + pull_request: + branches: [ master ] + +jobs: + build: + strategy: + fail-fast: false + matrix: + runtime: + - name: apisix-runtime-or-1.31.8.1 + script_url: "https://raw.githubusercontent.com/api7/apisix-build-tools/master/build-apisix-runtime.sh" + script_name: "build-apisix-runtime.sh" + # build-tools master still pins an apisix-nginx-module release without OpenResty 1.31.8.1 support + build_env: "OPENRESTY_VERSION=1.31.8.1 apisix_nginx_module_ver=support-openresty-1.31.8.1" + + runs-on: "ubuntu-24.04" + name: build (${{ matrix.runtime.name }}) + + env: + OPENRESTY_PREFIX: "/usr/local/openresty" + + steps: + - name: Check out code + uses: actions/checkout@v4 + + - name: Get dependencies + run: | + sudo apt update + sudo apt install -y git wget build-essential cpanminus libncurses5-dev libreadline-dev libssl-dev perl libpcre3 libpcre3-dev zlib1g-dev + + - name: Before install + run: | + sudo cpanm --notest Test::Nginx > build.log 2>&1 || (cat build.log && exit 1) + git clone https://github.com/openresty/test-nginx.git test-nginx + + - name: Install + run: | + wget ${{ matrix.runtime.script_url }} + chmod +x ${{ matrix.runtime.script_name }} + env ${{ matrix.runtime.build_env }} OR_PREFIX=$OPENRESTY_PREFIX CC="gcc -fsanitize=address -fdiagnostics-color=always -Werror -Wno-unused-but-set-variable -Wno-unused-parameter" \ + ./${{ matrix.runtime.script_name }} latest + + - name: Script + run: | + export PATH=$OPENRESTY_PREFIX/nginx/sbin:$PATH + prove -I. -Itest-nginx/lib -r t/ diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..080fcd1 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +t/servroot/ diff --git a/nginx-1.31.8.patch b/nginx-1.31.8.patch new file mode 100644 index 0000000..0c92503 --- /dev/null +++ b/nginx-1.31.8.patch @@ -0,0 +1,273 @@ +diff --git src/core/ngx_connection.h src/core/ngx_connection.h +index 271ac24..7dae4fc 100644 +--- src/core/ngx_connection.h ++++ src/core/ngx_connection.h +@@ -203,6 +203,9 @@ struct ngx_connection_s { + #if (NGX_THREADS || NGX_COMPAT) + ngx_thread_task_t *sendfile_task; + #endif ++#if (T_NGX_MULTI_UPSTREAM) ++ void *multi_c; ++#endif + }; + + +diff --git src/http/ngx_http_request.h src/http/ngx_http_request.h +index 6b62b21..9b519fe 100644 +--- src/http/ngx_http_request.h ++++ src/http/ngx_http_request.h +@@ -614,6 +614,13 @@ struct ngx_http_request_s { + + unsigned http_minor:16; + unsigned http_major:16; ++ ++#if (T_NGX_MULTI_UPSTREAM) ++ ngx_queue_t *multi_item; ++ ngx_queue_t *backend_r; ++ ngx_queue_t waiting_queue; ++ ngx_flag_t waiting; ++#endif + }; + + +diff --git src/http/ngx_http_upstream.c src/http/ngx_http_upstream.c +index 1c7a9f1..67a15d2 100644 +--- src/http/ngx_http_upstream.c ++++ src/http/ngx_http_upstream.c +@@ -9,6 +9,10 @@ + #include + #include + ++#if (T_NGX_MULTI_UPSTREAM) ++#include ++#endif ++ + + #if (NGX_HTTP_CACHE) + static ngx_int_t ngx_http_upstream_cache(ngx_http_request_t *r, +@@ -1582,6 +1586,12 @@ ngx_http_upstream_check_broken_connection(ngx_http_request_t *r, + } + + ++#if (T_NGX_MULTI_UPSTREAM) ++ ++#include "ngx_http_multi_upstream.c" ++ ++#endif /* T_NGX_MULTI_UPSTREAM */ ++ + static void + ngx_http_upstream_connect(ngx_http_request_t *r, ngx_http_upstream_t *u) + { +@@ -1663,6 +1673,58 @@ ngx_http_upstream_connect(ngx_http_request_t *r, ngx_http_upstream_t *u) + + c = u->peer.connection; + ++#if (T_NGX_MULTI_UPSTREAM) ++ if (u->multi) { ++ if (!(u->multi_mode & NGX_MULTI_UPS_SUPPORT_MULTI)) { ++ ngx_log_error(NGX_LOG_ERR, c->log, 0, ++ "multi: upstream configured multi, but handler does not support it"); ++ ngx_http_multi_upstream_finalize_request(c, ++ NGX_HTTP_INTERNAL_SERVER_ERROR); ++ return; ++ } ++ ++ if (rc == NGX_AGAIN) { //first real connect ++ c->read->handler = ngx_http_multi_upstream_connect_handler; ++ c->write->handler = ngx_http_multi_upstream_connect_handler; ++ ngx_add_timer(c->write, u->conf->connect_timeout); ++ ngx_log_error(NGX_LOG_INFO, c->log, 0, ++ "multi: connect new to backend %p", c); ++ } else if (rc == NGX_DONE) { ++ if (!u->peer.cached) { //new connection, connected immediately ++ c->read->handler = ngx_http_multi_upstream_connect_handler; ++ c->write->handler = ngx_http_multi_upstream_connect_handler; ++ ngx_add_timer(c->write, u->conf->connect_timeout); ++ ngx_log_error(NGX_LOG_INFO, c->log, 0, ++ "multi: connect new to backend immediately %p", c); ++ ngx_http_multi_upstream_connect_handler(c->write); ++ } else if (ngx_multi_connected(c)) { //use exist connection ++ ngx_log_error(NGX_LOG_INFO, c->log, 0, "multi: connect reuse %p", c); ++ ++ ngx_http_multi_upstream_init_request(c, r); ++ ngx_http_multi_upstream_process(c, 1); ++ } else { ++ /* queued on the connection, connect_init will start it */ ++ ngx_log_error(NGX_LOG_INFO, c->log, 0, ++ "multi: connect reuse unfinished %p, wait", c); ++ } ++ } else { ++ ngx_log_error(NGX_LOG_ERR, c->log, 0, ++ "multi: connect return %i error", rc); ++ ngx_http_multi_upstream_finalize_request(c, ++ NGX_HTTP_INTERNAL_SERVER_ERROR); ++ } ++ ++ return; ++ } else if ((u->multi_mode & NGX_MULTI_UPS_NEED_MULTI) == NGX_MULTI_UPS_NEED_MULTI) { ++ ngx_log_error(NGX_LOG_ERR, c->log, 0, ++ "multi: need multi, but upstream does not support it, " ++ "maybe need configuration 'multi' in upstream"); ++ ngx_http_upstream_finalize_request(r, u, ++ NGX_HTTP_INTERNAL_SERVER_ERROR); ++ return; ++ } ++#endif ++ + c->requests++; + + c->data = r; +@@ -1915,6 +1977,13 @@ ngx_http_upstream_ssl_handshake(ngx_http_request_t *r, ngx_http_upstream_t *u, + u->output.sendfile = 0; + } + ++#if (T_NGX_MULTI_UPSTREAM) ++ if (u->multi) { ++ ngx_http_multi_upstream_connect_init(c); ++ return; ++ } ++#endif ++ + c->write->handler = ngx_http_upstream_handler; + c->read->handler = ngx_http_upstream_handler; + +@@ -2243,6 +2312,14 @@ ngx_http_upstream_send_request(ngx_http_request_t *r, ngx_http_upstream_t *u, + ngx_post_event(c->read, &ngx_posted_events); + } + ++#if (T_NGX_MULTI_UPSTREAM) ++ if (u->multi && r->connection != u->peer.connection && !r->waiting) { ++ ngx_multi_connection_t *multi_c = ngx_get_multi_connection(c); ++ ngx_queue_insert_tail(&multi_c->waiting_list, &r->waiting_queue); ++ r->waiting = 1; ++ } ++#endif ++ + return; + } + +@@ -2289,6 +2366,12 @@ ngx_http_upstream_send_request(ngx_http_request_t *r, ngx_http_upstream_t *u, + ngx_add_timer(c->read, u->read_timeout); + + if (c->read->ready) { ++#if (T_NGX_MULTI_UPSTREAM) ++ if (u->multi) { ++ ngx_http_multi_upstream_read_handler(c); ++ return; ++ } ++#endif + ngx_http_upstream_process_header(r, u); + return; + } +@@ -2326,6 +2409,11 @@ ngx_http_upstream_send_request_body(ngx_http_request_t *r, + u->request_body_blocked = 1; + + } else { ++#if T_NGX_MULTI_UPSTREAM ++ if (u->multi && rc == NGX_OK) { ++ ngx_multi_clean_leak(u->peer.connection); ++ } ++#endif + u->request_body_blocked = 0; + } + +@@ -3937,6 +4025,13 @@ ngx_http_upstream_process_non_buffered_downstream(ngx_http_request_t *r) + return; + } + ++#if (T_NGX_MULTI_UPSTREAM) ++ if (u->multi) { ++ ngx_http_multi_upstream_process_non_buffered_request(r); ++ return; ++ } ++#endif ++ + ngx_http_upstream_process_non_buffered_request(r, 1); + } + +@@ -4631,6 +4726,16 @@ ngx_http_upstream_next(ngx_http_request_t *r, ngx_http_upstream_t *u, + ngx_log_debug1(NGX_LOG_DEBUG_HTTP, r->connection->log, 0, + "http next upstream, %xi", ft_type); + ++#if (T_NGX_MULTI_UPSTREAM) ++ if (u->multi && ngx_http_multi_connection_fake(r)) { ++ ngx_log_error(NGX_LOG_INFO, r->connection->log, 0, ++ "multi: http next upstream fake_r %p", r); ++ ++ ngx_http_multi_upstream_next(r->connection, ft_type); ++ return; ++ } ++#endif ++ + if (u->peer.sockaddr) { + + if (u->peer.connection) { +@@ -4805,6 +4910,16 @@ ngx_http_upstream_finalize_request(ngx_http_request_t *r, + ngx_log_debug1(NGX_LOG_DEBUG_HTTP, r->connection->log, 0, + "finalize http upstream request: %i", rc); + ++#if (T_NGX_MULTI_UPSTREAM) ++ if (u->multi && ngx_http_multi_connection_fake(r)) { ++ ngx_log_error(NGX_LOG_INFO, r->connection->log, 0, ++ "http finalize upstream fake_r %p", r); ++ ++ ngx_http_multi_upstream_finalize_request(r->connection, rc); ++ return; ++ } ++#endif ++ + if (u->cleanup == NULL) { + /* the request was already finalized */ + ngx_http_finalize_request(r, NGX_DONE); +diff --git src/http/ngx_http_upstream.h src/http/ngx_http_upstream.h +index 8f0422b..09bd6b3 100644 +--- src/http/ngx_http_upstream.h ++++ src/http/ngx_http_upstream.h +@@ -430,6 +430,14 @@ struct ngx_http_upstream_s { + unsigned request_body_blocked:1; + unsigned header_sent:1; + unsigned response_received:1; ++ ++#if (T_NGX_MULTI_UPSTREAM) ++ unsigned multi:1; ++ void *multi_init; ++ ngx_pool_t *send_pool; ++ ngx_flag_t multi_mode; ++#endif ++ + }; + + +diff --git src/stream/ngx_stream.h src/stream/ngx_stream.h +index 40db2a6..3b74c1c 100644 +--- src/stream/ngx_stream.h ++++ src/stream/ngx_stream.h +@@ -297,6 +297,13 @@ struct ngx_stream_session_s { + unsigned health_check:1; + + unsigned limit_conn_status:2; ++ ++#if (T_NGX_MULTI_UPSTREAM) ++ ngx_queue_t *multi_item; ++ ngx_queue_t *backend_r; ++ ngx_queue_t waiting_queue; ++ ngx_flag_t waiting; ++#endif + }; + + +diff --git src/stream/ngx_stream_upstream.h src/stream/ngx_stream_upstream.h +index 4bc6f10..5904998 100644 +--- src/stream/ngx_stream_upstream.h ++++ src/stream/ngx_stream_upstream.h +@@ -152,6 +152,10 @@ typedef struct { + unsigned connected:1; + unsigned proxy_protocol:2; + unsigned half_closed:1; ++ ++#if (T_NGX_MULTI_UPSTREAM) ++ unsigned multi:1; ++#endif + } ngx_stream_upstream_t; + + diff --git a/ngx_http_multi_upstream.c b/ngx_http_multi_upstream.c index cd52423..206978b 100644 --- a/ngx_http_multi_upstream.c +++ b/ngx_http_multi_upstream.c @@ -451,7 +451,12 @@ ngx_http_multi_upstream_read_handler(ngx_connection_t *pc) } } - if (ngx_http_upstream_process_headers(real_r, real_u) != NGX_OK) { + rc = ngx_http_upstream_process_headers(real_r, real_u); + if (rc != NGX_OK) { + /* nginx 1.31.8+ leaves NGX_ERROR to the caller */ + if (rc == NGX_ERROR) { + ngx_http_upstream_finalize_request(real_r, real_u, NGX_ERROR); + } continue; } @@ -472,7 +477,11 @@ ngx_http_multi_upstream_read_handler(ngx_connection_t *pc) } } - if (ngx_http_upstream_process_headers(real_r, real_u) != NGX_OK) { + rc = ngx_http_upstream_process_headers(real_r, real_u); + if (rc != NGX_OK) { + if (rc == NGX_ERROR) { + ngx_http_upstream_finalize_request(real_r, real_u, NGX_ERROR); + } continue; } diff --git a/ngx_http_multi_upstream_module.c b/ngx_http_multi_upstream_module.c index 597965c..2fd6d98 100644 --- a/ngx_http_multi_upstream_module.c +++ b/ngx_http_multi_upstream_module.c @@ -321,6 +321,9 @@ ngx_http_multi_upstream_get_peer(ngx_peer_connection_t *pc, void *data) } c = pc->connection; + /* tells ngx_http_upstream_connect() this is a new connection */ + pc->cached = 0; + if (ngx_http_multi_upstream_init_connection(c, pc, data) != NGX_OK) { ngx_log_error(NGX_LOG_ERR, pc->log, 0, "multi: init new connection failed, c: %p", c); diff --git a/patch.sh b/patch.sh index 44b51bc..8e5fd2f 100755 --- a/patch.sh +++ b/patch.sh @@ -42,6 +42,9 @@ elif [[ "$1" == *openresty-1.29.2.* ]]; then elif [[ "$1" == *openresty-1.31.1.* ]]; then patch="$script_dir/nginx-1.31.1.patch" dir="$1/bundle/nginx-1.31.1" +elif [[ "$1" == *openresty-1.31.8.* ]]; then + patch="$script_dir/nginx-1.31.8.patch" + dir="$1/bundle/nginx-1.31.8" else err "can't detect OpenResty version" exit 1 diff --git a/t/multi.t b/t/multi.t new file mode 100644 index 0000000..d868afd --- /dev/null +++ b/t/multi.t @@ -0,0 +1,209 @@ +use Test::Nginx::Socket::Lua 'no_plan'; + +log_level('info'); +no_long_string(); +no_shuffle(); + +$ENV{TEST_NGINX_DUBBO_SOCK} ||= html_dir() . "/dubbo.sock"; + +# a stub dubbo backend: answers every request with status 200 and body "dubbo ok" +my $stub = <<'_EOC_'; +local bit = require("bit") + +local function be32(s, i) + local a, b, c, d = s:byte(i, i + 3) + return ((a * 256 + b) * 256 + c) * 256 + d +end + +local function u32(n) + return string.char(bit.band(bit.rshift(n, 24), 0xff), bit.band(bit.rshift(n, 16), 0xff), + bit.band(bit.rshift(n, 8), 0xff), bit.band(n, 0xff)) +end + +local function str(s) + return string.char(#s) .. s +end + +local function bin(s) + return string.char(0x20 + #s) .. s +end + +local sock = assert(ngx.req.socket(true)) +while true do + local hdr = sock:receive(16) + if not hdr then + return + end + + local len = be32(hdr, 13) + if len > 0 and not sock:receive(len) then + return + end + + local flag, reqid = hdr:byte(3), hdr:sub(5, 12) + local typ, payload + if bit.band(flag, 0x20) ~= 0 then + typ, payload = 0x22, "N" + else + typ = 0x02 + payload = "\x91H" .. str("status") .. str("200") + .. str("body") .. bin("dubbo ok") .. "Z" + end + + sock:send("\xda\xbb" .. string.char(typ, 20) .. reqid .. u32(#payload) .. payload) +end +_EOC_ + +add_block_preprocessor(sub { + my ($block) = @_; + + # a slow shutdown can leave the socket behind + unlink $ENV{TEST_NGINX_DUBBO_SOCK}; + my $listen_unix = ''; + if (($block->http_config // '') =~ /TEST_NGINX_DUBBO_SOCK/) { + $listen_unix = "listen unix:$ENV{TEST_NGINX_DUBBO_SOCK};"; + } + + my $main_config = $block->main_config // ''; + $main_config .= <<_EOC_; +stream { + server { + listen 1986; + $listen_unix + content_by_lua_block { +$stub + } + } +} +_EOC_ + $block->set_value("main_config", $main_config); + + if (!$block->request) { + $block->set_value("request", "GET /t"); + } + + if (!$block->no_error_log && !$block->error_log) { + $block->set_value("no_error_log", "[error]\n[alert]"); + } +}); + +run_tests(); + +__DATA__ + +=== TEST 1: multiplexed dubbo request +--- http_config +upstream dubbo_backend { + server 127.0.0.1:1986; + multi 4; +} +--- config +location /t { + dubbo_pass org.apache.dubbo.Demo 1.0.0 hello dubbo_backend; +} +--- response_body chomp +dubbo ok +--- error_log +multi: connect new to backend + + + +=== TEST 2: later requests reuse the multi connection +--- http_config +upstream dubbo_backend { + server 127.0.0.1:1986; + multi 1; +} +--- config +location /dubbo { + dubbo_pass org.apache.dubbo.Demo 1.0.0 hello dubbo_backend; +} +location /t { + content_by_lua_block { + for i = 1, 2 do + local res = ngx.location.capture("/dubbo") + ngx.say(res.status, " ", res.body) + end + } +} +--- response_body +200 dubbo ok +200 dubbo ok +--- error_log +multi: connect reuse + + + +=== TEST 3: requests on a connection that is still connecting wait for it +--- http_config +upstream dubbo_backend { + server 127.0.0.1:1986; + multi 1; +} +--- config +location /dubbo { + dubbo_pass org.apache.dubbo.Demo 1.0.0 hello dubbo_backend; +} +location /t { + content_by_lua_block { + -- both start before the connect completes, so the second one + -- lands on the connection the first one is still opening + local r1, r2 = ngx.location.capture_multi({ {"/dubbo"}, {"/dubbo"} }) + ngx.say(r1.status, " ", r1.body) + ngx.say(r2.status, " ", r2.body) + } +} +--- response_body +200 dubbo ok +200 dubbo ok +--- error_log +multi: connect reuse unfinished + + + +=== TEST 4: a new connection that connects immediately +unix sockets connect synchronously +--- http_config +upstream dubbo_backend { + server unix:$TEST_NGINX_DUBBO_SOCK; + multi 4; +} +--- config +location /t { + dubbo_pass org.apache.dubbo.Demo 1.0.0 hello dubbo_backend; +} +--- response_body chomp +dubbo ok +--- error_log +multi: connect new to backend immediately + + + +=== TEST 5: multi upstream used by a handler that can not multiplex +--- http_config +upstream backend { + server 127.0.0.1:1986; + multi 4; +} +--- config +location /t { + proxy_pass http://backend; +} +--- error_code: 500 +--- error_log +multi: upstream configured multi, but handler does not support it + + + +=== TEST 6: dubbo_pass to an upstream without multi +--- http_config +upstream dubbo_backend { + server 127.0.0.1:1986; +} +--- config +location /t { + dubbo_pass org.apache.dubbo.Demo 1.0.0 hello dubbo_backend; +} +--- error_code: 500 +--- error_log +multi: need multi, but upstream does not support it