feat(runtime): automatically scale CPU work and prepare 0.89.0

Cargo.toml

@@ -61,7 +61,7 @@ categories = ["network-programming", "web-programming::http-server"]
 
 # 二进制走 CalVer，crate 走 SemVer —— 两者是两回事，卖给不同的人。
 # 这里是 crate 的版本。
-version = "0.88.0"
+version = "0.89.0"
 
 [workspace.dependencies]
 # ── 内部 crate ───────────────────────────────────────────
@@ -89,6 +89,8 @@ tw-handover = { path = "crates/tw-handover" }
 
 # ── 版本与企业版 workspace 对齐，便于反向依赖时不打架 ────
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "time", "sync", "io-util", "signal"] }
+# 只读内存容量和 Linux cgroup 限额，不扫描进程，也不开采集线程。
+sysinfo = { version = "0.36", default-features = false, features = ["system"] }
 axum = { version = "0.8", features = ["macros", "ws"] }
 tower = "0.5"
 hyper = { version = "1", features = ["full"] }

bin/twcore/Cargo.toml

@@ -13,7 +13,14 @@ description = "Self-contained AI gateway: the local engine behind ThinkWatch Lit
 keywords.workspace = true
 categories.workspace = true
 
+[features]
+default = ["mimalloc"]
+# 系统分配器的构建只用于对照诊断；发布包默认启用，无需用户设置。
+mimalloc = ["dep:mimalloc"]
+
 [dependencies]
+# Linux 不主动申请 THP，避免小对象/线程堆按大页驻留；不要求用户设置内核或环境变量。
+mimalloc = { version = "0.1.52", optional = true, features = ["no_thp"] }
 tw-config = { workspace = true }
 tw-control = { workspace = true }
 # 升级不断线：新旧 core 之间的交接连接

bin/twcore/src/main.rs

@@ -18,12 +18,17 @@ mod lockfile;
 mod pack_cli;
 mod proc;
 mod record;
+mod runtime;
 mod sync_cmd;
 mod upgrade;
 
 use background::Background;
 use lockfile::{LockFile, LockOutcome};
 
+#[cfg(feature = "mimalloc")]
+#[global_allocator]
+static ALLOCATOR: mimalloc::MiMalloc = mimalloc::MiMalloc;
+
 #[derive(Parser)]
 // `about` 不写值就是 Cargo.toml 里的 description：一句话只写一处，两边不会各说各的
 #[command(name = "twcore", version, about)]
@@ -256,6 +261,7 @@ fn main() -> Result<()> {
                 }),
         )
         .init();
+    runtime::init_allocator();
     // 网关的代码 panic 时，那个请求的结局写得出哪一行、为什么（stderr 在桌面版里没人看）
     tw_gateway::panic_note::install();
 
@@ -394,7 +400,9 @@ fn fmt_time(ms: u64) -> String {
 /// 数字准不准。
 fn cmd_speed(path: &Path, provider: Option<String>, proxy: Option<String>) -> Result<()> {
     let cfg = tw_config::load(path).with_context(|| format!("reading {}", path.display()))?;
-    let rt = tokio::runtime::Runtime::new()?;
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()?;
     rt.block_on(async move {
         if let Some(name) = proxy {
             let px = cfg
@@ -792,9 +800,7 @@ fn cmd_serve(path: &Path, port: Option<u16>, safe: bool, parent: Option<u32>) ->
     // 端口文件）都会顺手 `create_dir_all`，谁先到谁建 —— 而 Windows 上「谁
     // 建的」决定了它的 ACL，也就决定了那份明文密钥同机的其他用户读不读得到。
     tw_config::private_dir::create(&dir).with_context(|| format!("creating {}", dir.display()))?;
-    let rt = tokio::runtime::Builder::new_multi_thread()
-        .enable_all()
-        .build()?;
+    let rt = runtime::build()?;
     // **锁被一个活着的 core 占着时，先问它是不是在等接替者**（升级不断线，见 `handover`）。
     // 是的话接管它；不是的话照旧报「已有实例在运行」
     let (mut lock, mut incoming) = match LockFile::acquire(&dir)? {

bin/twcore/src/runtime.rs

@@ -0,0 +1,95 @@
+//! 网关的运行时按机器自动定规模；任务由 Tokio 调度，用户无需写 runtime 配置。
+
+use anyhow::{Context, Result};
+
+/// Linux 的 THP 会把小对象所在的稀疏堆按 2 MiB 驻留，抵消线程回收的收益。
+/// 只约束本进程，不改系统设置，也不要求用户提供分配器环境变量。
+pub fn init_allocator() {
+    #[cfg(all(target_os = "linux", feature = "mimalloc"))]
+    {
+        // SAFETY: PR_SET_THP_DISABLE 只接受整数参数，不读写 Rust 的内存。
+        // 必须在创建工作线程之前调用，使其继承该进程的策略。
+        let result = unsafe {
+            libc::prctl(
+                libc::PR_SET_THP_DISABLE,
+                1 as libc::c_ulong,
+                0 as libc::c_ulong,
+                0 as libc::c_ulong,
+                0 as libc::c_ulong,
+            )
+        };
+        if result != 0 {
+            tracing::debug!(error = %std::io::Error::last_os_error(), "could not disable transparent huge pages");
+        }
+    }
+}
+
+pub fn build() -> Result<tokio::runtime::Runtime> {
+    let policy = tw_gateway::resources::policy();
+    // 保留 Tokio 原有的显式覆盖，但自动值本身就适用于大核服务器。
+    let workers = workers(
+        policy.workers,
+        std::env::var("TOKIO_WORKER_THREADS").ok().as_deref(),
+    )?;
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(workers)
+        .thread_keep_alive(tw_gateway::resources::CPU_KEEP_ALIVE)
+        .enable_all()
+        .build()?;
+    tracing::info!(
+        workers,
+        heavy_jobs = policy.heavy_jobs,
+        cpu_idle_seconds = tw_gateway::resources::CPU_KEEP_ALIVE.as_secs(),
+        body_budget_mib = policy.body_budget / (1024 * 1024),
+        allocator = if cfg!(feature = "mimalloc") {
+            "mimalloc"
+        } else {
+            "system"
+        },
+        "automatically sized gateway runtime"
+    );
+    Ok(rt)
+}
+
+fn workers(automatic: usize, explicit: Option<&str>) -> Result<usize> {
+    match explicit {
+        None => Ok(automatic),
+        Some(value) => value
+            .parse::<std::num::NonZeroUsize>()
+            .map(usize::from)
+            .with_context(|| {
+                format!("TOKIO_WORKER_THREADS must be a positive integer, got `{value}`")
+            }),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[cfg(all(target_os = "linux", feature = "mimalloc"))]
+    #[test]
+    fn allocator_disables_transparent_huge_pages_for_this_process() {
+        init_allocator();
+        // SAFETY: PR_GET_THP_DISABLE returns the process flag and takes integer arguments.
+        let disabled = unsafe {
+            libc::prctl(
+                libc::PR_GET_THP_DISABLE,
+                0 as libc::c_ulong,
+                0 as libc::c_ulong,
+                0 as libc::c_ulong,
+                0 as libc::c_ulong,
+            )
+        };
+        assert_eq!(disabled, 1);
+    }
+
+    #[test]
+    fn explicit_tokio_override_is_optional_and_validated() {
+        assert_eq!(workers(8, None).unwrap(), 8);
+        assert_eq!(workers(8, Some("16")).unwrap(), 16);
+        for bad in ["0", "", "-1", "eight"] {
+            assert!(workers(8, Some(bad)).is_err());
+        }
+    }
+}

crates/tw-api/msg-codes.txt

@@ -425,6 +425,7 @@ gw.request.body_declared_too_large
 gw.request.body_over_limit
 gw.request.body_unreadable
 gw.request.decoded_body_over_limit
+gw.request.memory_busy
 gw.request.unsupported_encoding
 gw.route.all_selected_disabled
 gw.route.denied

crates/tw-gateway/Cargo.toml

@@ -51,6 +51,7 @@ tw-dialect = { workspace = true }
 tw-plugin = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
+sysinfo = { workspace = true }
 tracing = { workspace = true }
 zstd = { workspace = true }
 # 压缩过的请求体和上游回答：gzip、deflate 用它解，zstd 用上面那个

crates/tw-gateway/src/content.rs

@@ -947,7 +947,13 @@ async fn redact(raw: Bytes, r: Redaction) -> String {
     let heavy = raw.len() >= HEAVY;
     let go = move || r.apply(&String::from_utf8_lossy(&raw));
     if heavy {
-        tokio::task::spawn_blocking(go).await.unwrap_or_default()
+        let slot = crate::resources::heavy_slot().await;
+        tokio::task::spawn_blocking(move || {
+            let _slot = slot;
+            go()
+        })
+        .await
+        .unwrap_or_default()
     } else {
         go()
     }

crates/tw-gateway/src/inflate.rs

@@ -30,6 +30,8 @@ pub enum Coding {
 pub enum Decode {
     /// 解出来的超过了上限
     OverLimit,
+    /// 进程的正文缓冲预算暂时用满了，客户端可以稍后重试
+    Busy,
     /// 数据和编码对不上、解到一半断了
     Failed(String),
 }
@@ -57,6 +59,16 @@ impl Coding {
     /// 解开 `raw`，解出来的最多 `max` 字节。**读到第 `max + 1` 个字节就停**：解出来的有多大，
     /// 压缩数据自己说了不算（zstd 的帧头可以写一个假的长度），只信实际解出来的
     pub fn decode(self, raw: &[u8], max: usize) -> Result<Bytes, Decode> {
+        self.decode_with(raw, max, |_| Ok(()))
+    }
+
+    /// 解压输出每次增长前预留预算，不信压缩帧自报的长度。
+    pub(crate) fn decode_with(
+        self,
+        raw: &[u8],
+        max: usize,
+        mut reserve: impl FnMut(usize) -> Result<(), Decode>,
+    ) -> Result<Bytes, Decode> {
         let reader: Box<dyn Read> = match self {
             Self::Gzip => Box::new(flate2::read::MultiGzDecoder::new(raw)),
             // HTTP 的 deflate 是 zlib 封装的（RFC 9110 §8.4.1.2）
@@ -66,12 +78,21 @@ impl Coding {
             ),
         };
         let mut out = Vec::new();
-        reader
-            .take(max as u64 + 1)
-            .read_to_end(&mut out)
-            .map_err(|e| Decode::Failed(e.to_string()))?;
-        if out.len() > max {
-            return Err(Decode::OverLimit);
+        let mut reader = reader.take(max as u64 + 1);
+        let mut chunk = [0; 16 * 1024];
+        loop {
+            let n = reader
+                .read(&mut chunk)
+                .map_err(|e| Decode::Failed(e.to_string()))?;
+            if n == 0 {
+                break;
+            }
+            let next = out.len().saturating_add(n);
+            if next > max {
+                return Err(Decode::OverLimit);
+            }
+            reserve(next)?;
+            out.extend_from_slice(&chunk[..n]);
         }
         Ok(Bytes::from(out))
     }

crates/tw-gateway/src/lib.rs

@@ -51,6 +51,7 @@ pub mod probe;
 mod pulse;
 pub mod quota;
 pub mod quote;
+pub mod resources;
 pub mod retry;
 pub mod seal;
 pub mod sent;