feat(proxy): serve from thread-per-core workers with per-worker upstream pools

.github/workflows/ci.yml

@@ -218,9 +218,25 @@ jobs:
           retention-days: 1
 
   e2e:
-    name: e2e (vitest) + coverage
+    name: e2e (vitest, ${{ matrix.serving }}) + coverage
     needs: [build-bin]
     runs-on: ubuntu-latest
+    strategy:
+      # The two serving modes are separate products of the same binary:
+      # thread-per-core answers a request on the thread that accepted it
+      # and dispatches on that thread's own upstream pool, work-stealing
+      # shares one runtime and one pool. A failure in one says nothing
+      # about the other, so neither leg cancels the other.
+      fail-fast: false
+      matrix:
+        include:
+          # The Linux default, and so the one production runs.
+          - serving: thread-per-core
+            thread_per_core: "true"
+          # The documented fallback. Kept green so it stays a real
+          # option when a deployment needs it.
+          - serving: work-stealing
+            thread_per_core: "false"
     services:
       etcd:
         image: quay.io/coreos/etcd:v3.5.15
@@ -270,11 +286,19 @@ jobs:
       - name: run e2e
         if: steps.harness.outputs.present == 'true'
         working-directory: tests/e2e
+        # Read by the harness, which writes it into every spawned
+        # gateway's `proxy.thread_per_core`. The binary's own default is
+        # platform-derived, so without this both legs would run the same
+        # mode and the second would prove nothing.
+        env:
+          E2E_THREAD_PER_CORE: ${{ matrix.thread_per_core }}
         run: pnpm test
       - uses: actions/upload-artifact@v7
         if: always()
         with:
-          name: coverage-e2e
+          # Per leg: two uploads under one name is a hard conflict the
+          # moment the suite starts emitting coverage.
+          name: coverage-e2e-${{ matrix.serving }}
           path: tests/e2e/coverage/lcov.info
           if-no-files-found: ignore
           retention-days: 7
@@ -296,7 +320,7 @@ jobs:
         # follow-up), this download must remain soft — otherwise the
         # coverage-gate hard-fails every run on `Artifact not found`.
         continue-on-error: true
-        with: { name: coverage-e2e, path: cov/e2e }
+        with: { pattern: coverage-e2e-*, path: cov/e2e, merge-multiple: true }
       - name: merge + threshold
         run: |
           npm i -g lcov-result-merger @lcov-viewer/cli

config.example.yaml

@@ -51,6 +51,16 @@ proxy:
   #   trusted_proxies: ["10.0.0.0/8", "127.0.0.1/32"]
   #   recursive: true
   #   header: x-forwarded-for
+  # Serve from independent worker threads, each with its own runtime,
+  # its own SO_REUSEPORT listener on addr, and its own upstream
+  # connection pool, so a request is handled end to end on the thread
+  # that accepted it. Omitted, it is on for Linux and off elsewhere.
+  # Set false to serve from one shared runtime. Applied at startup.
+  # thread_per_core: true
+  # Number of proxy worker threads. Omitted, it follows the parallelism
+  # available to the process, so a container CPU limit or a taskset
+  # affinity mask sizes it. Applied at startup.
+  # workers: 4
   # Entry-level URL rewriting: map legacy URL shapes onto AISIX endpoints
   # before routing. The first rule whose `match` regex matches the request
   # path rewrites it (once, no cascading); the request then flows through

config.managed.yaml

@@ -45,6 +45,11 @@ proxy:
   # 0 = no request-body cap (the default); set a value to bound
   # per-request memory.
   # request_body_limit_bytes: 0
+  # Serving topology; on for Linux by default. A managed deployment
+  # reaches these through the environment rather than by editing this
+  # file: AISIX_PROXY__THREAD_PER_CORE, AISIX_PROXY__WORKERS.
+  # thread_per_core: true
+  # workers: 4
   # Entry-level URL rewriting (first matching rule wins; `match` runs on
   # the raw, percent-encoded path; `rewrite` replaces the matched portion,
   # $1 = capture group; the query string is preserved). Lets clients keep

crates/aisix-admin/src/playground_handler.rs

@@ -81,6 +81,8 @@ mod tests {
             real_ip: Default::default(),
             url_rewrites: Vec::new(),
             tls: None,
+            thread_per_core: None,
+            workers: None,
         }
     }
 

crates/aisix-core/src/config.rs

@@ -412,6 +412,35 @@ pub struct ProxyConfig {
     /// an L7 LB / ingress that sets `x-forwarded-for`.
     #[serde(default)]
     pub real_ip: RealIpConfig,
+    /// Serve the proxy from independent worker threads — each with its
+    /// own runtime, its own `SO_REUSEPORT` listener on `addr`, and its
+    /// own upstream connection pool — instead of one shared runtime
+    /// whose threads hand work to each other.
+    ///
+    /// Omitted, the default, enables it on Linux and disables it
+    /// elsewhere: the kernel spreads incoming connections across
+    /// same-port listeners on Linux, and other platforms do not.
+    /// Set `false` to serve from one shared runtime on any platform.
+    ///
+    /// A request is handled end to end on the thread that accepted it,
+    /// which removes a cross-thread handoff per request. On a small
+    /// number of client connections (fewer than about four per worker)
+    /// the kernel's per-connection spreading can leave workers unevenly
+    /// loaded; throughput at that size may be lower than with a shared
+    /// runtime.
+    ///
+    /// Applied at startup. Changing it requires a restart.
+    #[serde(default)]
+    pub thread_per_core: Option<bool>,
+    /// Number of proxy worker threads.
+    ///
+    /// Omitted, the default, uses the parallelism available to the
+    /// process, which follows the CPU limits applied by a container
+    /// runtime, cgroup, or `taskset`. Must be at least 1.
+    ///
+    /// Applied at startup. Changing it requires a restart.
+    #[serde(default)]
+    pub workers: Option<usize>,
     /// Entry-level URL rewrite rules, applied to every proxy-listener
     /// request **before** routing (the admin and metrics listeners are
     /// unaffected). The first rule whose `match` regex matches the request
@@ -433,6 +462,23 @@ impl ProxyConfig {
     const fn default_body_limit() -> usize {
         0
     }
+
+    /// Whether the proxy serves from thread-per-core workers, resolving
+    /// the platform default when unset.
+    pub fn thread_per_core_enabled(&self) -> bool {
+        self.thread_per_core.unwrap_or(cfg!(target_os = "linux"))
+    }
+
+    /// Proxy worker-thread count, resolving the default when unset.
+    ///
+    /// `available_parallelism` reports the CPUs this process may actually
+    /// run on, so a cgroup CPU limit or a `taskset` affinity mask sizes
+    /// the pool correctly without the operator restating it here. Falls
+    /// back to 1 on the platforms that cannot report it.
+    pub fn worker_threads(&self) -> usize {
+        self.workers
+            .unwrap_or_else(|| std::thread::available_parallelism().map_or(1, |n| n.get()))
+    }
 }
 
 /// One entry-level URL rewrite rule (see [`ProxyConfig::url_rewrites`]).
@@ -1349,6 +1395,15 @@ impl Config {
                 "proxy.real_ip.trusted_proxies invalid CIDR/IP: {bad}"
             )));
         }
+        // Zero workers would bind no listener at all: the proxy would
+        // boot, report healthy, and refuse every connection.
+        if self.proxy.workers == Some(0) {
+            return Err(BootstrapError::Config(
+                "proxy.workers must be at least 1 (omit it to use the \
+                 parallelism available to the process)"
+                    .into(),
+            ));
+        }
         // The dedicated metrics listener address must be a bindable
         // socket address — it is always bound when prometheus is enabled.
         let metrics_addr = &self.observability.metrics.prometheus.addr;
@@ -2651,4 +2706,81 @@ admin:
         assert_eq!(tls.client_key_file, "/c.key");
         assert_eq!(tls.domain_name.as_deref(), Some("etcd.aisix.cloud"));
     }
+
+    /// Serving topology is a startup decision, so every existing config
+    /// — none of which names it — has to keep loading and resolve to the
+    /// platform's answer.
+    #[test]
+    fn serving_topology_defaults_to_the_platform_answer() {
+        let f = write_yaml(
+            r#"
+etcd:
+  endpoints: ["http://localhost:2379"]
+proxy:
+  addr: "0.0.0.0:3000"
+admin:
+  addr: "127.0.0.1:3001"
+  admin_keys: ["k1"]
+"#,
+        );
+        let cfg = Config::load_from_path(Some(f.path())).unwrap();
+        assert_eq!(cfg.proxy.thread_per_core, None);
+        assert_eq!(cfg.proxy.workers, None);
+        assert_eq!(
+            cfg.proxy.thread_per_core_enabled(),
+            cfg!(target_os = "linux"),
+            "thread-per-core is the default where the kernel spreads \
+             connections across same-port listeners, and only there"
+        );
+        assert_eq!(
+            cfg.proxy.worker_threads(),
+            std::thread::available_parallelism().map_or(1, |n| n.get()),
+        );
+    }
+
+    /// The fallback an operator reaches for when thread-per-core is the
+    /// wrong shape for their traffic. It has to win on every platform,
+    /// including the one where it is also the default.
+    #[test]
+    fn explicit_serving_topology_overrides_the_platform_default() {
+        let f = write_yaml(
+            r#"
+etcd:
+  endpoints: ["http://localhost:2379"]
+proxy:
+  addr: "0.0.0.0:3000"
+  thread_per_core: false
+  workers: 3
+admin:
+  addr: "127.0.0.1:3001"
+  admin_keys: ["k1"]
+"#,
+        );
+        let cfg = Config::load_from_path(Some(f.path())).unwrap();
+        assert_eq!(cfg.proxy.thread_per_core, Some(false));
+        assert!(!cfg.proxy.thread_per_core_enabled());
+        assert_eq!(cfg.proxy.worker_threads(), 3);
+    }
+
+    /// Zero workers would bind no listener and still report a healthy
+    /// boot, so it has to fail at load naming the field to fix.
+    #[test]
+    fn rejects_zero_proxy_workers() {
+        let f = write_yaml(
+            r#"
+etcd:
+  endpoints: ["http://localhost:2379"]
+proxy:
+  addr: "0.0.0.0:3000"
+  workers: 0
+admin:
+  addr: "127.0.0.1:3001"
+  admin_keys: ["k1"]
+"#,
+        );
+        let err = Config::load_from_path(Some(f.path()))
+            .unwrap_err()
+            .to_string();
+        assert!(err.contains("proxy.workers"), "unexpected error: {err}");
+    }
 }

crates/aisix-gateway/src/upstream_http.rs

@@ -418,6 +418,48 @@ mod tests {
         assert!(!production.contains("fn t()"), "{production}");
     }
 
+    /// On a thread-per-core worker every dispatch runs on that worker's
+    /// own pool, and that one pool stands in for all of the clients
+    /// below — so it is built with a single user agent
+    /// (`upstream_tls::DISPATCH_USER_AGENT`).
+    ///
+    /// That substitution is only invisible while the clients it replaces
+    /// agree on the user agent. Give one bridge its own and the header
+    /// it sends changes depending on which serving mode the deployment
+    /// runs, which is not something an upstream-facing identity is
+    /// allowed to do. Whoever wants a distinct user agent has to give
+    /// that client a distinct pool as well.
+    #[test]
+    fn every_dispatch_client_presents_the_same_user_agent() {
+        let crates_dir = std::path::Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/.."));
+        // Only literals: the telemetry, heartbeat, and OTLP clients build
+        // theirs from a version string or a named constant, and none of
+        // them talks to a model provider or reaches the per-worker pool.
+        const NEEDLE: &str = ".user_agent(\"";
+        let mut offenders = Vec::new();
+        for file in rust_sources(crates_dir) {
+            let src = std::fs::read_to_string(&file).expect("read source");
+            for (n, line) in production_half(&src).lines().enumerate() {
+                let Some(rest) = line.split_once(NEEDLE).map(|(_, r)| r) else {
+                    continue;
+                };
+                let Some((agent, _)) = rest.split_once('"') else {
+                    continue;
+                };
+                if agent != crate::upstream_tls::DISPATCH_USER_AGENT {
+                    offenders.push(format!("{}:{}: {agent}", file.display(), n + 1));
+                }
+            }
+        }
+        assert!(
+            offenders.is_empty(),
+            "these dispatch clients present a user agent the per-worker \
+             pool would replace with `{}`:\n{}",
+            crate::upstream_tls::DISPATCH_USER_AGENT,
+            offenders.join("\n"),
+        );
+    }
+
     fn rust_sources(dir: &std::path::Path) -> Vec<std::path::PathBuf> {
         let mut out = Vec::new();
         let Ok(entries) = std::fs::read_dir(dir) else {

crates/aisix-gateway/src/upstream_tls.rs

@@ -210,6 +210,72 @@ pub fn reqwest_material() -> &'static ReqwestTlsMaterial {
 /// idle connection pool.
 static PK_CLIENTS: OnceLock<dashmap::DashMap<ProviderKeyTls, reqwest::Client>> = OnceLock::new();
 
+// ─── per-worker pools ────────────────────────────────────────────────
+
+/// The user agent every dispatch-path client is built with.
+///
+/// A worker's pool stands in for those clients, so it has to present the
+/// same identity upstream. `every_dispatch_client_presents_the_same_user_agent`
+/// holds them in step.
+pub(crate) const DISPATCH_USER_AGENT: &str = "aisix/0.1";
+
+thread_local! {
+    /// Whether this thread serves proxy traffic on its own runtime.
+    static IS_WORKER_THREAD: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
+
+    /// This worker's upstream pool, built on first dispatch. `None` once
+    /// a build failure has been reported for this thread.
+    static WORKER_CLIENT: std::cell::OnceCell<Option<reqwest::Client>> =
+        const { std::cell::OnceCell::new() };
+}
+
+/// Declares the calling thread a proxy worker with its own runtime, so
+/// its dispatches use a pool this thread alone polls.
+///
+/// Called once per worker in thread-per-core serving. Left unset
+/// everywhere else — the shared runtime's threads, the blocking pool,
+/// background tasks — so those keep dispatching on the process-wide
+/// pools exactly as before.
+pub fn mark_worker_thread() {
+    IS_WORKER_THREAD.set(true);
+}
+
+/// This worker's pool, or `None` when the thread is not a worker or the
+/// pool could not be built.
+///
+/// One pool per worker rather than per client: the dispatch clients are
+/// all built from the same recipe, so merging them costs nothing but the
+/// per-owner split of `pool_max_idle_per_host`, and keeping them split
+/// would multiply idle connections by the worker count for no gain.
+///
+/// A build failure falls back to the shared pool — which carries the
+/// deployment's trust settings — and is cached so one broken
+/// configuration cannot log per request.
+fn worker_client() -> Option<reqwest::Client> {
+    if !IS_WORKER_THREAD.get() {
+        return None;
+    }
+    WORKER_CLIENT.with(|cell| {
+        cell.get_or_init(|| {
+            match crate::upstream_http::client_builder()
+                .user_agent(DISPATCH_USER_AGENT)
+                .build()
+            {
+                Ok(client) => Some(client),
+                Err(e) => {
+                    tracing::error!(
+                        error = %e,
+                        "per-worker upstream pool could not be built; this worker \
+                         dispatches on the shared pool"
+                    );
+                    None
+                }
+            }
+        })
+        .clone()
+    })
+}
+
 /// The client to dispatch this Provider Key's request on.
 ///
 /// Returns `shared` unchanged whenever the key sets no override, which
@@ -226,7 +292,12 @@ pub fn client_for_provider_key(
     tls: Option<&ProviderKeyTls>,
 ) -> reqwest::Client {
     let Some(tls) = tls.filter(|t| !t.is_noop()) else {
-        return shared.clone();
+        // On a thread-per-core worker, dispatch on that worker's own
+        // pool: the upstream connection is then read by the same runtime
+        // that is waiting for the response, instead of waking a thread
+        // that has to hand it back. Everywhere else this is `None` and
+        // the shared pool is used, as it always was.
+        return worker_client().unwrap_or_else(|| shared.clone());
     };
     let cache = PK_CLIENTS.get_or_init(dashmap::DashMap::new);
     if let Some(existing) = cache.get(tls) {