Merge f-152-users (2e111557b) into f-split-int: CPU worker counts per physical core with a cap of 32, host socket, quota; warning above the recommendation; setup caps what it writes

CMakeLists.txt

@@ -2091,6 +2091,10 @@ if(STRATA_BUILD_TESTS)
   add_executable(pool_test src/kernels/cpu/pool_test.cpp)
   target_link_libraries(pool_test PRIVATE strata_kernels_cpu)
   add_test(NAME pool_test COMMAND pool_test --selftest WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR})
+  # the default CPU expert worker count on mock CPUs (8c/16t, 16c/32t, 64c/128t, 2-socket 128c/256t, ...)
+  add_executable(pool_plan_test src/kernels/cpu/pool_plan_test.cpp)
+  target_include_directories(pool_plan_test PRIVATE include)
+  add_test(NAME pool_plan_test COMMAND pool_plan_test)
   # issue #29: tiny batches with sleeping workers, as on a big-VRAM card; a hang fails it
   add_executable(pool_stress src/kernels/cpu/pool_stress.cpp)
   target_link_libraries(pool_stress PRIVATE strata_kernels_cpu)

docs/DETAILS.md

@@ -596,6 +596,20 @@ the same worker count and workload. This does not change kernels, phase barriers
 does not affect the legacy single-token/oracle fallback. Setup's `--calibrate` does not tune it yet.
 For the server, add `"--pool-tasks", "192"` to the existing `args` list in its configuration, then restart it.
 
+**How many CPU threads the engine starts (`--pool-workers`).** Without the setting the engine picks the number of
+CPU expert workers from the CPU, not from the count of logical threads: one worker per *physical* core (never an SMT
+sibling) except the core of the main thread, one core less on a CPU with 17 or more cores, only the cores of the main
+thread's CPU socket on a multi-socket machine, at most 32, no more than a container's CPU limit (docker `--cpus`) allows,
+and on a hybrid CPU with more efficiency than performance cores the performance cores but one plus half of the efficiency
+cores. When the OS does not say which logical processors share a core (some containers), it counts half of them. The start
+line says what bounded the number (`CPU pool: 31 workers by default (32 physical cores, 64 logical processors; bounded by ...)`).
+The workers wait by spinning, so a worker without a core of its own takes time from the others: on a 20-core CPU 24
+workers decoded 44% slower than the default, on a 32-core EPYC 63 workers 35% slower. A count in the config
+(`--pool-workers N`) is always used as given; when it is above what the CPU uses well the engine prints a plain warning
+with the recommended number at start (`strata cpu pool: warning: ...`). Setup writes no worker count except the hybrid
+CPU's, capped the same way, and gives the CPU vision encoder at most 16 threads (`--threads` in the `vision` entry of the
+config changes that).
+
 ### Running it at startup (Task Scheduler)
 
 To have the model up at logon, people start the serve from **Task Scheduler** (or a service). Beware: Windows

include/strata/kernels/cpu/pool.hpp

@@ -29,6 +29,7 @@
 
 #include "strata/kernels/cpu/expert.hpp"
 #include "strata/kernels/cpu/native_expert.hpp"
+#include "strata/kernels/cpu/pool_plan.hpp"
 
 #include <atomic>
 #include <cstdio>
@@ -82,8 +83,22 @@ struct CpuTopology {
     std::vector<int> worker_cores;  ///< Ordered CPU IDs; on Windows, group * 64 + processor within the group
     int host_core = -1;             ///< Logical core reserved for host thread
     int host_sibling = -1;          ///< The host core's SMT sibling (-1: none), the logical processor `Sibling` takes
+    /// For the default worker count (pool_plan.hpp): the CPU package (socket) of each worker core (parallel to
+    /// `worker_cores`, -1: not known) and of the host core, the logical processors the process may use, whether the
+    /// OS told physical cores from SMT siblings, and the container's CPU quota (0: none).  On a multi-socket CPU the
+    /// host's package comes first in `worker_cores`.
+    std::vector<int> worker_package;
+    int host_package = -1;
+    int logical = 0;
+    bool layout_known = true;
+    double cpu_quota = 0;
 };
 
+/// The description plan_pool_workers() takes, from a detected topology.
+PoolPlanInput pool_plan_input(const CpuTopology& topo);
+/// The CPUs a container quota allows (cgroup v2 cpu.max / v1 cfs quota), 0 when none or not Linux.
+double container_cpu_quota();
+
 /// Which core the host thread - the layer loop that spins on the GPU's flags - takes when `skip_first` reserves one
 /// (--host-core first|last|sibling, STRATA_HOST_CORE).  `First` is the layout the pool has always used.  `Last`
 /// (eddoursul's fork, F12) puts the host on the last physical core and lets the workers have the first: Windows sends
@@ -165,6 +180,11 @@ public:
     ExpertPool& operator=(const ExpertPool&) = delete;
 
     int workers() const { return n_; }
+    /// What bounded the default worker count (pool_plan.hpp; "" when a count was configured), the physical cores
+    /// and logical processors counted - for the start-up line.
+    const char* plan_limit() const { return plan_limit_; }
+    int plan_cores() const { return plan_cores_; }
+    int plan_logical() const { return plan_logical_; }
     /// Whether the host thread also drains.  Reported at startup, because "the engine adapts to the machine it
     /// is on" is only true if the engine says which adaptation it took.
     bool host_works() const { return host_works_; }
@@ -242,6 +262,9 @@ private:
     void publish();
 
     int n_ = 0;
+    const char* plan_limit_ = "";
+    int plan_cores_ = 0;
+    int plan_logical_ = 0;
     bool host_works_ = true;
     ExpertJob* jobs_ = nullptr;
     int njobs_ = 0;

include/strata/kernels/cpu/pool_plan.hpp

@@ -0,0 +1,126 @@
+// include/strata/kernels/cpu/pool_plan.hpp - how many CPU expert workers the engine starts by default, and when a
+// configured count deserves a warning.  Pure functions of a CPU description (no OS calls), so the choice is unit
+// tested on mock topologies (pool_plan_test.cpp: 8c/16t, 16c/32t, 64c/128t, a 2-socket 128c/256t box, ...).
+//
+// Why a plan and not "every physical core but the host's":
+//   * the expert pool's workers spin while they wait.  A worker without a core of its own (more workers than physical
+//     cores) takes cycles from a worker that has one, so a big count is slower than a small one, not just useless.
+//     A report from a 256-thread machine: ~256 workers, speed collapsed.
+//   * the kernel is DRAM-bound; past the memory system's width more cores add nothing, only the barrier at the end of
+//     every layer waits for more threads (kPoolWorkersCap, measured: phaseA-tests/notes-152-users.md).
+//   * on a 2-socket box the workers of the other socket read the experts across the interconnect (35 workers measured
+//     slower than 17-18 on 2-socket Xeons, bench #674 #707): the default stays on the host thread's socket.
+//   * a container limited to a few CPUs by a CFS quota (docker --cpus) still shows every core of the machine; spinning
+//     threads beyond the quota are throttled together.
+//
+// A count the user configures (--pool-workers) always runs as given; `pool_workers_warning` says in plain words when it
+// is above what this CPU can use well (recommend, never force).
+#pragma once
+
+#include <algorithm>
+#include <string>
+#include <vector>
+
+namespace strata::kernels::cpu {
+
+/// The most default workers: past this the layer barrier costs more than the extra cores read (see the header comment).
+/// Measured on 6c/12t (z440), 20c (P2003), 32c/64t (r9700) and 64c (l620): notes-152-users.md.
+inline constexpr int kPoolWorkersCap = 32;
+/// Cores kept free for the other threads of the engine (copy / IO threads, the GPU driver's, the OS) on a CPU with at
+/// least `kPoolReserveFrom` cores that the pool may use (besides the host's).  Small CPUs keep every core.
+inline constexpr int kPoolReserveFrom = 16;
+inline constexpr int kPoolReserve = 1;
+
+struct PoolPlanInput {
+    int logical = 0;                    ///< logical processors this process may run on
+    /// physical cores this process may use per CPU package (socket), the host thread's package FIRST; the host
+    /// thread's own core is counted in it.  One entry on a single-socket CPU.  Empty: the topology is not known.
+    std::vector<int> cores_per_package;
+    double cpu_quota = 0;               ///< CPUs a container quota allows (cgroup cpu.max); 0: no quota
+    /// Physical performance / efficiency cores of a hybrid CPU (both 0 on an ordinary one).  With more E-cores than
+    /// P-cores the default is the P-cores but the host's plus half of the E-cores (#642: i9-14900KF 8P+16E, 15 workers
+    /// 165 / 116 tok/s against 106 / 84 with all 23; Core Ultra 7 265, 8P+12E, RTX 3060 decode at 4K: 12-16 workers
+    /// +12.6% over the 19 of one-per-core, 24 workers -44%: notes-152-users.md).
+    int p_cores = 0;
+    int e_cores = 0;
+};
+
+struct PoolPlan {
+    int workers = 1;                    ///< the default count of pool workers (the host thread works beside them)
+    int cores = 0;                      ///< physical cores counted (all packages)
+    int cores_local = 0;                ///< ... on the host's package
+    int packages = 1;
+    bool topology_known = true;
+    const char* limit = "";             ///< what bounded the count (for the start-up line)
+};
+
+inline PoolPlan plan_pool_workers(const PoolPlanInput& in) {
+    PoolPlan p;
+    int cores = 0;
+    for (int c : in.cores_per_package) cores += std::max(0, c);
+    int local = in.cores_per_package.empty() ? 0 : std::max(0, in.cores_per_package.front());
+    p.packages = std::max<int>(1, (int) in.cores_per_package.size());
+    p.topology_known = !in.cores_per_package.empty() && cores > 0;
+    if (!p.topology_known) {
+        // The OS did not say which logical processors share a core.  Counting every one as a core is what put a
+        // 256-thread machine at 255 workers; a 2-way SMT is the usual case on a many-core CPU.
+        cores = in.logical >= 16 ? in.logical / 2 : in.logical;
+        local = cores;
+        p.packages = 1;
+    }
+    p.cores = cores;
+    p.cores_local = local;
+    p.limit = "every physical core but the host's";
+    int n = std::max(1, local - 1);                       // the host thread has the first core
+    if (!p.topology_known) p.limit = "half the logical processors (the OS gave no core layout)";
+    else if (p.packages > 1) p.limit = "the host's CPU socket";
+    if (n >= kPoolReserveFrom) {                          // a big CPU keeps a core for the engine's other threads
+        n -= kPoolReserve;
+        if (p.topology_known && p.packages == 1) p.limit = "physical cores, one kept free";
+    }
+    if (in.p_cores > 0 && in.e_cores > in.p_cores) {      // a hybrid CPU with more E-cores than P-cores
+        const int h = std::max(1, in.p_cores - 1 + in.e_cores / 2);
+        if (h < n) {
+            n = h;
+            p.limit = "the performance cores and half of the efficiency cores";
+        }
+    }
+    if (in.cpu_quota > 0.0) {
+        const int q = std::max(1, (int) in.cpu_quota - 1);
+        if (q < n) {
+            n = q;
+            p.limit = "the container's CPU quota";
+        }
+    }
+    if (n > kPoolWorkersCap) {
+        n = kPoolWorkersCap;
+        p.limit = "the measured cap";
+    }
+    p.workers = std::max(1, n);
+    return p;
+}
+
+/// The warning for a configured worker count `requested` (> 0), or "" when it is fine.  `plan` is plan_pool_workers(in).
+inline std::string pool_workers_warning(int requested, const PoolPlanInput& in, const PoolPlan& plan) {
+    if (requested <= 0 || requested <= plan.workers) return "";
+    const std::string rec = "Recommended: --pool-workers " + std::to_string(plan.workers) +
+                            " (or leave it out: the engine picks that by itself). It still runs as configured.";
+    const std::string asked = std::to_string(requested) + " CPU expert workers are configured";
+    if (requested + 1 > in.logical && in.logical > 0)
+        return asked + ", but this process can run only " + std::to_string(in.logical) +
+               " CPU threads. The extra workers spin and take time from the others. " + rec;
+    if (plan.topology_known && requested + 1 > plan.cores)
+        return asked + ", but this CPU has " + std::to_string(plan.cores) +
+               " physical cores. Two workers on one core share its memory bandwidth and slow each other down. " + rec;
+    if (plan.packages > 1 && requested + 1 > plan.cores_local)
+        return asked + ", more than the " + std::to_string(plan.cores_local) +
+               " cores of the CPU socket the engine's main thread runs on. Workers on the other socket read the experts " +
+               "across the link and were measured slower than fewer local ones. " + rec;
+    if (in.cpu_quota > 0.0 && (double) requested + 1 > in.cpu_quota)
+        return asked + ", but a container limit allows about " + std::to_string((int) in.cpu_quota) +
+               " CPUs. Threads beyond the limit are throttled together. " + rec;
+    return asked + ", more than this CPU can use well (the speed stops rising at about " +
+           std::to_string(plan.workers) + " workers and the end of each layer waits for every one). " + rec;
+}
+
+}  // namespace strata::kernels::cpu

setup.py

@@ -941,6 +941,7 @@ def recommend_pool_workers(args: list) -> list:
     n = hybrid_pool_workers(cpu_cores())
     if n is None or "--pool-workers" in args:
         return args
+    n = min(n, POOL_WORKERS_CAP)   # setup never writes a big worker count (see plan_pool_workers)
     p, e = cpu_cores()
     ok(f"hybrid CPU ({p} performance + {e} efficiency cores): {n} CPU expert workers - the performance cores and half "
        "of the efficiency cores (--pool-workers in the config; START-HERE --calibrate measures it on this PC)")
@@ -999,6 +1000,70 @@ def two_socket_note(sockets) -> list[str]:
             "the config's args (START-HERE --calibrate measures it on this PC)"]
 
 
+# The engine's default count of CPU expert workers (include/strata/kernels/cpu/pool_plan.hpp, which tools/test_pool_plan.py
+# keeps in step with these numbers).  Setup writes no worker count on an ordinary CPU: the engine picks it, from the
+# physical cores (not the SMT siblings), minus the host thread's core and a reserve on a big CPU, on the host thread's
+# socket, under a cap.  What setup does write (a hybrid CPU's count, the vision encoder's threads) is capped the same way.
+POOL_WORKERS_CAP = 32
+POOL_RESERVE_FROM = 16
+POOL_RESERVE = 1
+VISION_THREADS_CAP = 16
+
+
+def plan_pool_workers(logical: int, cores_per_package=None, quota: float = 0.0, hybrid=None) -> int:
+    """The engine's default pool worker count for a CPU of `logical` threads whose physical cores per socket are
+    `cores_per_package` (the host thread's socket first; None/empty: the layout is not known) and a container CPU
+    quota (0: none)."""
+    cores = list(cores_per_package or [])
+    if cores and sum(cores) > 0:
+        local = cores[0]
+    else:                                              # no layout: half the logical processors on a many-thread CPU
+        local = logical // 2 if logical >= 16 else logical
+    n = max(1, local - 1)                              # the host thread has the first core
+    if n >= POOL_RESERVE_FROM:
+        n -= POOL_RESERVE
+    if hybrid and hybrid[0] > 0 and hybrid[1] > hybrid[0]:   # (P-cores, E-cores): the P-cores but the host's + half the E-cores
+        n = min(n, max(1, hybrid[0] - 1 + hybrid[1] // 2))
+    if quota > 0:
+        n = min(n, max(1, int(quota) - 1))
+    return max(1, min(n, POOL_WORKERS_CAP))
+
+
+def cpu_core_layout():
+    """(logical threads, [physical cores per socket]); the cores list is empty when the OS does not say."""
+    logical = os.cpu_count() or 1
+    sockets = cpu_sockets()
+    return logical, ([sockets[1]] * sockets[0] if sockets else [])
+
+
+def pool_workers_note(args, layout=None) -> list[str]:
+    """A plain warning when the config's `--pool-workers` is above what this CPU uses well.  Recommend, never force: the
+    count stays as it is and the engine runs with it."""
+    if "--pool-workers" not in args:
+        return []
+    try:
+        asked = int(args[args.index("--pool-workers") + 1])
+    except (IndexError, ValueError):
+        return []
+    hybrid = cpu_cores() if layout is None else None
+    logical, cores = layout or cpu_core_layout()
+    rec = plan_pool_workers(logical, cores, hybrid=hybrid)
+    if asked <= rec:
+        return []
+    why = ("more than this PC has CPU threads" if asked + 1 > logical else
+           "more than its physical cores" if cores and asked + 1 > sum(cores) else
+           "more than the cores of one CPU socket" if len(cores) > 1 and asked + 1 > cores[0] else
+           "more than the memory bandwidth can feed")
+    return [f"the config asks for {asked} CPU expert workers, {why}: extra workers can make Strata slower. Recommended: "
+            f"--pool-workers {rec} (or remove the setting: the engine picks that by itself). It stays as you set it."]
+
+
+def vision_threads(logical: int | None = None) -> int:
+    """The CPU vision encoder's threads: one per physical core as before (half the logical threads), at most
+    VISION_THREADS_CAP - a 256-thread server does not need 128 of them."""
+    return max(1, min((logical if logical is not None else (os.cpu_count() or 8)) // 2, VISION_THREADS_CAP))
+
+
 def cpu_info():
     """(name, avx2, avx512): avx512 means everything Strata's fast AVX-512 kernels use (F, BW, VL, VNNI, VBMI),
     the same test the engine makes (cpu_avx512_ok), not just AVX-512F."""
@@ -6195,6 +6260,8 @@ def install_own_path(a, family, gpu, hip, amd, cuda_tk, multi, port, ram, adopte
     if a.browser is not None:
         cfg["open_browser"] = a.browser
     cfg["args"] = recommend_pool_workers(cfg["args"])   # --pool-workers: this engine path's CPU expert threads too
+    for line in pool_workers_note(cfg["args"]):
+        warn(line)
     cfg_path = home() / f"strata-{tag.lower()}.json"
     if vision != "none":
         old_cfg = cfg_path if cfg_path.is_file() else adopted
@@ -6203,7 +6270,7 @@ def install_own_path(a, family, gpu, hip, amd, cuda_tk, multi, port, ram, adopte
         cfg["vision"] = {"exe": str(eng / VEXE), "mmproj": str(mmproj), "model": str(gguf), "gpu": vision == "gpu",
                          "max_tokens": vt}          # the text model, opened vocab-only
         if vision == "cpu":
-            cfg["vision"]["threads"] = max(1, (os.cpu_count() or 8) // 2)
+            cfg["vision"]["threads"] = vision_threads()
     elif getattr(a, "vision_tokens", None) is not None:
         warn("--vision-tokens: images are off, so it is not used")
     # requests at once ("parallel", opt-in): written only when given; the recommendation is said once
@@ -7595,7 +7662,7 @@ def main() -> int:
                          "model": str(ready_paths(model, models_dir)["dense"] if ready else shards[0]),
                          "gpu": vision == "gpu", "max_tokens": vt}
         if vision == "cpu":
-            cfg["vision"]["threads"] = max(1, (os.cpu_count() or 8) // 2)
+            cfg["vision"]["threads"] = vision_threads()
     elif a.vision_tokens is not None:
         warn("--vision-tokens: images are off for this model, so it is not used")
     if a.images:                                       # only when given: else an earlier "images" is carried over
@@ -7611,6 +7678,8 @@ def main() -> int:
         ok("the settings tuned for this PC earlier are used" + (f" ({cal['date']})" if cal.get("date") else ""))
     else:                                              # #642: measured counts (a calibration) win over the rule
         cfg["args"] = recommend_pool_workers(cfg["args"])
+        for line in pool_workers_note(cfg["args"]):
+            warn(line)
         for line in two_socket_note(cpu_sockets()):    # bench tips: text only, the config is not touched
             say("  " + line)
     for line in bench_tips(cfg["args"], cfg.get("env"), ram, MODELS[model]["ram_gb"], gpu.get("vram_gb", 0.0), vision, WIN):

src/kernels/cpu/pool.cpp

@@ -15,7 +15,10 @@
 
 #include <cstdio>
 #include <cstdlib>
+#include <cstring>
 #include <stdexcept>
+#include <string>
+#include <thread>
 
 #if defined(_WIN32)
 #define WIN32_LEAN_AND_MEAN
@@ -79,6 +82,59 @@ HostCore host_core_setting() { return (HostCore) g_host_core.load(); }
 
 static CpuTopology detect_cpu_topology_impl(bool skip_first, PoolAffinity affinity);
 
+// The fields the default worker count needs (pool_plan.hpp): the logical processors the process may use, the CPU
+// package of the host core and of every worker core, whether the OS told physical cores from SMT siblings, the
+// container CPU quota.  On a multi-socket CPU (not hybrid) the host package comes first in `worker_cores`.
+static void fill_plan_fields(CpuTopology& topo, bool skip_first) {
+    topo.cpu_quota = container_cpu_quota();
+    topo.worker_package.assign(topo.worker_cores.size(), -1);
+#if defined(_WIN32)
+    (void) skip_first;
+    topo.logical = (int) GetActiveProcessorCount(ALL_PROCESSOR_GROUPS);
+    return;   // (the sockets of a Windows box are not told apart: one package)
+#elif defined(__APPLE__)
+    (void) skip_first;
+    topo.logical = (int) std::thread::hardware_concurrency();
+    return;   // no SMT on Apple silicon and no sysfs: the cores are the processors
+#else
+    std::vector<unsigned long> mask;
+    std::vector<int> allowed;
+    topo.logical = detail::get_thread_affinity(mask, &allowed) == 0 ? (int) allowed.size()
+                                                                     : (int) std::thread::hardware_concurrency();
+    auto read = [](int cpu, const char* what) -> long {
+        char path[96];
+        std::snprintf(path, sizeof path, "/sys/devices/system/cpu/cpu%d/topology/%s", cpu, what);
+        long v = -1;
+        if (std::FILE* f = std::fopen(path, "r")) {
+            if (std::fscanf(f, "%ld", &v) != 1) v = -1;
+            std::fclose(f);
+        }
+        return v;
+    };
+    bool known = !topo.worker_cores.empty() || topo.host_core >= 0;
+    auto pkg_of = [&](int cpu) {
+        const long pkg = read(cpu, "physical_package_id"), core = read(cpu, "core_id");
+        if (pkg < 0 || core < 0) known = false;
+        return (int) pkg;
+    };
+    if (topo.host_core >= 0) topo.host_package = pkg_of(topo.host_core);
+    for (size_t i = 0; i < topo.worker_cores.size(); ++i) topo.worker_package[i] = pkg_of(topo.worker_cores[i]);
+    topo.layout_known = known;
+    if (!known) return;
+    if (skip_first && !topo.is_hybrid && topo.host_core >= 0 && topo.host_package >= 0) {
+        // the host socket first (stable inside each socket): the default workers are the local ones
+        std::vector<size_t> order(topo.worker_cores.size());
+        for (size_t i = 0; i < order.size(); ++i) order[i] = i;
+        std::stable_partition(order.begin(), order.end(),
+                              [&](size_t i) { return topo.worker_package[i] == topo.host_package; });
+        std::vector<int> cores, pkgs;
+        for (size_t i : order) { cores.push_back(topo.worker_cores[i]); pkgs.push_back(topo.worker_package[i]); }
+        topo.worker_cores = std::move(cores);
+        topo.worker_package = std::move(pkgs);
+    }
+#endif
+}
+
 CpuTopology detect_cpu_topology(bool skip_first, PoolAffinity affinity) {
     CpuTopology topo = detect_cpu_topology_impl(skip_first, affinity);
     // --host-core last: the host takes the last core, the workers the others (the first included)
@@ -96,9 +152,61 @@ CpuTopology detect_cpu_topology(bool skip_first, PoolAffinity affinity) {
                                 topo.worker_cores.end());
         topo.host_core = topo.host_sibling;
     }
+    fill_plan_fields(topo, skip_first);
     return topo;
 }
 
+double container_cpu_quota() {
+#if defined(__linux__)
+    // cgroup v2: "max 100000" (no limit) or "<quota> <period>"; v1: cpu.cfs_quota_us (-1: none) / cpu.cfs_period_us
+    if (std::FILE* f = std::fopen("/sys/fs/cgroup/cpu.max", "r")) {
+        char a[32] = {0};
+        long period = 0;
+        double q = 0;
+        if (std::fscanf(f, "%31s %ld", a, &period) == 2 && std::strcmp(a, "max") != 0 && period > 0)
+            q = std::atof(a) / (double) period;
+        std::fclose(f);
+        return q > 0 ? q : 0.0;
+    }
+    long quota = -1, period = 0;
+    if (std::FILE* f = std::fopen("/sys/fs/cgroup/cpu/cpu.cfs_quota_us", "r")) {
+        if (std::fscanf(f, "%ld", &quota) != 1) quota = -1;
+        std::fclose(f);
+    }
+    if (std::FILE* f = std::fopen("/sys/fs/cgroup/cpu/cpu.cfs_period_us", "r")) {
+        if (std::fscanf(f, "%ld", &period) != 1) period = 0;
+        std::fclose(f);
+    }
+    return quota > 0 && period > 0 ? (double) quota / (double) period : 0.0;
+#else
+    return 0.0;
+#endif
+}
+
+PoolPlanInput pool_plan_input(const CpuTopology& topo) {
+    PoolPlanInput in;
+    in.logical = topo.logical > 0 ? topo.logical : (int) std::thread::hardware_concurrency();
+    in.cpu_quota = topo.cpu_quota;
+    if (topo.is_hybrid) { in.p_cores = topo.p_cores; in.e_cores = topo.e_cores; }
+    if (!topo.layout_known) return in;   // cores_per_package stays empty
+    // the cores per package, the host package first; the host core counts in its package
+    std::vector<std::pair<int, int>> counts;   // (package, cores)
+    auto add = [&](int pkg) {
+        for (auto& c : counts)
+            if (c.first == pkg) { ++c.second; return; }
+        counts.emplace_back(pkg, 1);
+    };
+    if (topo.host_core >= 0) add(topo.host_package);
+    for (size_t i = 0; i < topo.worker_cores.size(); ++i)
+        add(i < topo.worker_package.size() ? topo.worker_package[i] : -1);
+    if (topo.host_core >= 0) {   // (the host package first even when it has no worker core listed)
+        auto it = std::find_if(counts.begin(), counts.end(), [&](const auto& c) { return c.first == topo.host_package; });
+        if (it != counts.begin()) std::iter_swap(counts.begin(), it);
+    }
+    for (const auto& c : counts) in.cores_per_package.push_back(c.second);
+    return in;
+}
+
 int planned_host_core(PoolAffinity affinity) { return detect_cpu_topology(true, affinity).host_core; }
 
 static CpuTopology detect_cpu_topology_impl(bool skip_first, PoolAffinity affinity) {
@@ -501,13 +609,22 @@ ExpertPool::ExpertPool(int n_workers, bool pin, bool host_works, PoolAffinity af
     if (tasks < 0 || tasks > kMaxTasks) throw std::invalid_argument("expert pool tasks must be in 0..4096");
     if (const char* e = std::getenv("STRATA_POOL_SPIN_US"))   // a test knob; see kSpinBeforeSleep
         spin_before_sleep_ = std::chrono::microseconds((std::max)(0, std::atoi(e)));
+    const PoolPlanInput plan_in = pool_plan_input(topo_);
+    const PoolPlan plan = plan_pool_workers(plan_in);
     if (n_workers > 0) {
         n_ = n_workers;
+        // recommend, never force: a count above what this CPU uses well runs as configured, with a plain warning
+        static std::atomic<bool> warned{false};
+        const std::string why = pool_workers_warning(n_workers, plan_in, plan);
+        if (!why.empty() && !warned.exchange(true)) std::fprintf(stderr, "strata cpu pool: warning: %s\n", why.c_str());
     } else if (topo_.is_hybrid && affinity_ != PoolAffinity::All) {
         n_ = (std::max)(1, topo_.p_cores - 1);
     } else {
-        n_ = (int) topo_.worker_cores.size();
+        n_ = (std::min)(plan.workers, (std::max)(1, (int) topo_.worker_cores.size()));
+        plan_limit_ = plan.limit;
     }
+    plan_cores_ = plan.cores;
+    plan_logical_ = plan_in.logical;
     if (n_ < 1) n_ = 1;
     scratch_.resize((size_t) n_);
     wstate_.reset(new std::atomic<int32_t>[(size_t) n_]);

src/kernels/cpu/pool_plan_test.cpp

@@ -0,0 +1,136 @@
+// pool_plan_test: the default CPU expert worker count on mock CPUs (include/strata/kernels/cpu/pool_plan.hpp).
+// Pure arithmetic, no threads and no OS calls: it runs on any machine.
+#include "strata/kernels/cpu/pool_plan.hpp"
+
+#include <cstdio>
+#include <string>
+
+namespace cpu = strata::kernels::cpu;
+
+static int g_fail = 0;
+
+static void expect(bool ok, const char* what) {
+    if (!ok) {
+        ++g_fail;
+        std::fprintf(stderr, "FAIL: %s\n", what);
+    }
+}
+
+static cpu::PoolPlanInput cpu_of(int logical, std::vector<int> per_package, double quota = 0) {
+    cpu::PoolPlanInput in;
+    in.logical = logical;
+    in.cores_per_package = std::move(per_package);
+    in.cpu_quota = quota;
+    return in;
+}
+
+static void line(const char* name, const cpu::PoolPlanInput& in) {
+    const cpu::PoolPlan p = cpu::plan_pool_workers(in);
+    std::printf("  %-34s logical %3d  cores %3d (host socket %3d, %d socket%s)  -> %2d workers  [%s]\n", name, in.logical,
+                p.cores, p.cores_local, p.packages, p.packages == 1 ? "" : "s", p.workers, p.limit);
+}
+
+int main() {
+    std::printf("pool_plan_test: cap %d, reserve %d from %d workers\n", cpu::kPoolWorkersCap, cpu::kPoolReserve,
+                cpu::kPoolReserveFrom);
+
+    // 8 cores / 16 threads: every physical core but the host's, never the SMT siblings
+    const auto c8 = cpu_of(16, {8});
+    line("8c/16t", c8);
+    expect(cpu::plan_pool_workers(c8).workers == 7, "8c/16t -> 7 workers");
+
+    // 6 cores / 12 threads (the small test PCs) and a CPU without SMT
+    expect(cpu::plan_pool_workers(cpu_of(12, {6})).workers == 5, "6c/12t -> 5 workers");
+    expect(cpu::plan_pool_workers(cpu_of(4, {4})).workers == 3, "4c/4t (no SMT) -> 3 workers");
+    expect(cpu::plan_pool_workers(cpu_of(1, {1})).workers == 1, "1 core -> 1 worker (never 0)");
+    expect(cpu::plan_pool_workers(cpu_of(2, {2})).workers == 1, "2 cores -> 1 worker");
+
+    // 16 cores / 32 threads
+    const auto c16 = cpu_of(32, {16});
+    line("16c/32t", c16);
+    {
+        const int w = cpu::plan_pool_workers(c16).workers;
+        expect(w == 15 - (15 >= cpu::kPoolReserveFrom ? cpu::kPoolReserve : 0) || w == cpu::kPoolWorkersCap,
+               "16c/32t -> physical cores minus the host (and the reserve)");
+        expect(w <= 15, "16c/32t never above the physical cores minus the host");
+    }
+
+    // 64 cores / 128 threads: the cap, far below the logical count
+    const auto c64 = cpu_of(128, {64});
+    line("64c/128t", c64);
+    expect(cpu::plan_pool_workers(c64).workers == cpu::kPoolWorkersCap, "64c/128t -> the cap");
+    expect(cpu::plan_pool_workers(c64).workers < 64, "64c/128t is not near the logical count");
+
+    // 2 sockets x 64 cores / 256 threads: the user report - never 255, one socket's cores at most, under the cap
+    const auto c2s = cpu_of(256, {64, 64});
+    line("2-socket 128c/256t", c2s);
+    expect(cpu::plan_pool_workers(c2s).workers == cpu::kPoolWorkersCap, "2-socket 128c/256t -> the cap");
+    expect(cpu::plan_pool_workers(c2s).packages == 2, "2-socket: two packages");
+    expect(cpu::plan_pool_workers(c2s).cores == 128, "2-socket: 128 physical cores counted");
+
+    // 2 sockets x 20 cores (the Xeons of bench #674 #707: 17-18 workers beat 35): the host's socket only
+    const auto xeon = cpu_of(80, {20, 20});
+    line("2-socket 40c/80t (Xeon)", xeon);
+    expect(cpu::plan_pool_workers(xeon).workers == 18, "2x20c: 18 workers (bench #674 #707: 17-18 beat 35): the host socket only, minus the host and the reserve");
+    // ... and a smaller second socket does not matter; the host's socket comes first in the input
+    expect(cpu::plan_pool_workers(cpu_of(48, {6, 18})).workers == 5, "host socket of 6 cores: 5 workers");
+
+    // an unknown core layout (no sysfs, a container): half the logical processors for a many-thread CPU, never all
+    const auto unk = cpu_of(256, {});
+    line("256 logical, layout unknown", unk);
+    expect(!cpu::plan_pool_workers(unk).topology_known, "unknown layout is flagged");
+    expect(cpu::plan_pool_workers(unk).workers <= cpu::kPoolWorkersCap, "unknown 256 -> at most the cap");
+    expect(cpu::plan_pool_workers(unk).workers < 255, "unknown 256 -> never 255");
+    expect(cpu::plan_pool_workers(cpu_of(8, {})).workers == 7, "unknown, 8 logical -> 7 workers");
+
+    // hybrid CPUs: with more E-cores than P-cores, the P-cores but the host's plus half of the E-cores
+    {
+        auto h = cpu_of(20, {20});
+        h.p_cores = 8;
+        h.e_cores = 12;
+        line("hybrid 8P+12E (Core Ultra 7 265)", h);
+        expect(cpu::plan_pool_workers(h).workers == 13, "8P+12E -> 13 workers (measured: 12-16 beat the 19 of one per core)");
+        auto h2 = cpu_of(32, {24});
+        h2.p_cores = 8;
+        h2.e_cores = 16;
+        expect(cpu::plan_pool_workers(h2).workers == 15, "8P+16E -> 15 workers (#642)");
+        auto h3 = cpu_of(24, {16});
+        h3.p_cores = 8;
+        h3.e_cores = 8;
+        expect(cpu::plan_pool_workers(h3).workers == 15, "8P+8E: not more E-cores than P-cores, every core but the host's");
+    }
+
+    // a container quota
+    const auto q = cpu_of(128, {64}, 4.0);
+    line("64c, container quota 4 CPUs", q);
+    expect(cpu::plan_pool_workers(q).workers == 3, "quota 4 CPUs -> 3 workers");
+    expect(cpu::plan_pool_workers(cpu_of(128, {64}, 200.0)).workers == cpu::kPoolWorkersCap, "a big quota changes nothing");
+
+    // warnings: only above the plan, and each says what to do
+    const cpu::PoolPlan p2s = cpu::plan_pool_workers(c2s);
+    expect(cpu::pool_workers_warning(0, c2s, p2s).empty(), "0 (the default) never warns");
+    expect(cpu::pool_workers_warning(p2s.workers, c2s, p2s).empty(), "the plan itself never warns");
+    expect(cpu::pool_workers_warning(4, c2s, p2s).empty(), "fewer than the plan never warns");
+    {
+        const std::string w = cpu::pool_workers_warning(256, c2s, p2s);
+        expect(!w.empty(), "256 workers on 256 threads warns");
+        expect(w.find("--pool-workers " + std::to_string(p2s.workers)) != std::string::npos, "the warning names the recommendation");
+        expect(w.find("still runs") != std::string::npos, "the warning says it still runs");
+    }
+    expect(cpu::pool_workers_warning(200, c2s, p2s).find("physical cores") != std::string::npos,
+           "200 workers on 128 cores: more than the physical cores");
+    expect(cpu::pool_workers_warning(300, c2s, p2s).find("256 CPU threads") != std::string::npos,
+           "300 workers on 256 threads: more than the CPU threads");
+    expect(cpu::pool_workers_warning(30, xeon, cpu::plan_pool_workers(xeon)).find("socket") != std::string::npos,
+           "30 workers on 2x20c: crosses the socket");
+    expect(!cpu::pool_workers_warning(40, c64, cpu::plan_pool_workers(c64)).empty(), "40 workers on 64c is above the cap: warns");
+    expect(cpu::pool_workers_warning(8, q, cpu::plan_pool_workers(q)).find("container") != std::string::npos,
+           "8 workers under a 4-CPU quota: names the container limit");
+
+    if (g_fail) {
+        std::fprintf(stderr, "pool_plan_test: %d failure(s)\n", g_fail);
+        return 1;
+    }
+    std::printf("pool_plan_test: OK\n");
+    return 0;
+}

src/program/generate.cpp

@@ -5712,6 +5712,9 @@ int main(int argc, char** argv) {
                  o.pool_tasks ? o.pool_tasks : 3 * (pool.workers() + (pool.host_works() ? 1 : 0)),
                  o.pool_tasks ? " (capped by rows)" : " (automatic)",
                  pool.workers() + (pool.host_works() ? 1 : 0));
+    if (pool.plan_limit()[0] != '\0')   // the default count: what it was bounded by (a configured count is never changed)
+        std::fprintf(stderr, "strata generate: CPU pool: %d workers by default (%d physical cores, %d logical processors; bounded by %s)\n",
+                     pool.workers(), pool.plan_cores(), pool.plan_logical(), pool.plan_limit());
     if (pool.is_hybrid() && pool.affinity() != strata::kernels::cpu::PoolAffinity::All) {
         const char* aff_str = pool.affinity() == strata::kernels::cpu::PoolAffinity::PCores ? "p-cores" :
                               pool.affinity() == strata::kernels::cpu::PoolAffinity::All ? "all" : "auto";

tools/test_pool_plan.py

@@ -0,0 +1,111 @@
+"""The default count of CPU expert workers on mock CPUs, and what setup writes about thread counts.
+
+The engine's choice is include/strata/kernels/cpu/pool_plan.hpp (tested on its own by the pool_plan_test binary); setup mirrors
+it so it can warn about a configured count.  This test keeps the two in step and checks the mirror on the same mock
+topologies (8c/16t, 16c/32t, 64c/128t, a 2-socket 128c/256t box) plus the reports behind it: a 256-thread machine started
+~256 workers and slowed down.
+
+    python -m unittest tools.test_pool_plan
+"""
+from __future__ import annotations
+
+import re
+import sys
+import unittest
+from pathlib import Path
+from unittest import mock
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+sys.path.insert(0, str(ROOT / "tools"))
+import setup  # noqa: E402
+
+HEADER = ROOT / "include" / "strata" / "kernels" / "cpu" / "pool_plan.hpp"
+
+
+class InStepWithTheEngine(unittest.TestCase):
+    def header_constant(self, name):
+        m = re.search(rf"constexpr int {name}\s*=\s*(\d+);", HEADER.read_text(encoding="utf-8"))
+        self.assertIsNotNone(m, name)
+        return int(m.group(1))
+
+    def test_constants(self):
+        self.assertEqual(setup.POOL_WORKERS_CAP, self.header_constant("kPoolWorkersCap"))
+        self.assertEqual(setup.POOL_RESERVE_FROM, self.header_constant("kPoolReserveFrom"))
+        self.assertEqual(setup.POOL_RESERVE, self.header_constant("kPoolReserve"))
+
+
+class DefaultWorkers(unittest.TestCase):
+    def test_mock_topologies(self):
+        cap = setup.POOL_WORKERS_CAP
+        # physical cores, not threads, minus the host thread's core
+        self.assertEqual(setup.plan_pool_workers(16, [8]), 7)            # 8c/16t
+        self.assertEqual(setup.plan_pool_workers(12, [6]), 5)            # 6c/12t
+        self.assertEqual(setup.plan_pool_workers(32, [16]), 15)          # 16c/32t
+        self.assertEqual(setup.plan_pool_workers(128, [64]), cap)        # 64c/128t: the cap
+        self.assertEqual(setup.plan_pool_workers(256, [64, 64]), cap)    # 2-socket 128c/256t: the cap, never 255
+        self.assertLess(setup.plan_pool_workers(256, [64, 64]), 128)
+        # two sockets of 20 cores: the host's socket only (bench #674 #707: 17-18 workers beat 35)
+        self.assertEqual(setup.plan_pool_workers(80, [20, 20]), 18)
+        self.assertEqual(setup.plan_pool_workers(48, [6, 18]), 5)
+        self.assertEqual(setup.plan_pool_workers(1, [1]), 1)
+
+    def test_hybrid(self):
+        self.assertEqual(setup.plan_pool_workers(20, [20], hybrid=(8, 12)), 13)     # Core Ultra 7 265 (measured)
+        self.assertEqual(setup.plan_pool_workers(32, [24], hybrid=(8, 16)), 15)     # i9-14900KF (#642)
+        self.assertEqual(setup.plan_pool_workers(24, [16], hybrid=(8, 8)), 15)      # not more E-cores than P-cores
+        self.assertEqual(setup.hybrid_pool_workers((8, 12)), 13)                    # setup's own rule agrees
+
+    def test_unknown_layout_and_quota(self):
+        self.assertLessEqual(setup.plan_pool_workers(256, []), setup.POOL_WORKERS_CAP)
+        self.assertLess(setup.plan_pool_workers(256, None), 255)
+        self.assertEqual(setup.plan_pool_workers(8, None), 7)
+        self.assertEqual(setup.plan_pool_workers(128, [64], quota=4.0), 3)   # a container limit of 4 CPUs
+
+    def test_never_writes_a_big_count(self):
+        # the hybrid rule (P-cores + half the E-cores) is capped like every other count
+        with mock.patch.object(setup, "cpu_cores", return_value=(8, 120)):
+            args = setup.recommend_pool_workers(["--spec", "4"])
+        n = int(args[args.index("--pool-workers") + 1])
+        self.assertLessEqual(n, setup.POOL_WORKERS_CAP)
+        with mock.patch.object(setup, "cpu_cores", return_value=(8, 16)):   # an ordinary hybrid: the usual 15
+            args = setup.recommend_pool_workers([])
+        self.assertEqual(args, ["--pool-workers", "15"])
+        with mock.patch.object(setup, "cpu_cores", return_value=None):      # not hybrid: nothing written
+            self.assertEqual(setup.recommend_pool_workers(["--spec", "4"]), ["--spec", "4"])
+        # a count that is already there is never changed
+        with mock.patch.object(setup, "cpu_cores", return_value=(8, 120)):
+            self.assertEqual(setup.recommend_pool_workers(["--pool-workers", "200"]), ["--pool-workers", "200"])
+
+    def test_vision_threads(self):
+        self.assertEqual(setup.vision_threads(16), 8)
+        self.assertEqual(setup.vision_threads(256), setup.VISION_THREADS_CAP)
+        self.assertEqual(setup.vision_threads(1), 1)
+
+
+class ConfiguredCountWarns(unittest.TestCase):
+    big = (256, [64, 64])
+
+    def test_warns_above_the_plan_and_keeps_the_count(self):
+        args = ["--pool-workers", "255"]
+        note = setup.pool_workers_note(args, self.big)
+        self.assertEqual(len(note), 1)
+        self.assertIn("--pool-workers %d" % setup.plan_pool_workers(*self.big), note[0])
+        self.assertIn("stays as you set it", note[0])
+        self.assertEqual(args, ["--pool-workers", "255"])     # nothing rewritten
+
+    def test_quiet_when_fine(self):
+        self.assertEqual(setup.pool_workers_note([], self.big), [])
+        self.assertEqual(setup.pool_workers_note(["--pool-workers", "8"], self.big), [])
+        self.assertEqual(setup.pool_workers_note(["--pool-workers", str(setup.plan_pool_workers(*self.big))], self.big), [])
+        self.assertEqual(setup.pool_workers_note(["--pool-workers"], self.big), [])
+        self.assertEqual(setup.pool_workers_note(["--pool-workers", "x"], self.big), [])
+
+    def test_reasons(self):
+        self.assertIn("CPU threads", setup.pool_workers_note(["--pool-workers", "300"], self.big)[0])
+        self.assertIn("physical cores", setup.pool_workers_note(["--pool-workers", "200"], self.big)[0])
+        self.assertIn("one CPU socket", setup.pool_workers_note(["--pool-workers", "30"], (80, [20, 20]))[0])
+
+
+if __name__ == "__main__":
+    unittest.main()

tools/vision/strata_vision.cpp

@@ -131,7 +131,8 @@ int main(int argc, char** argv) {
     if (!gpu && !fa_given && ggml_cpu_has_avx512()) fa = LLAMA_FLASH_ATTN_TYPE_DISABLED;
     cp.flash_attn_type = fa;
     // on the CPU without --threads: one per core (mtmd's own default is 4 threads)
-    if (threads <= 0 && !gpu) threads = std::max(1u, std::thread::hardware_concurrency() / 2);
+    // (at most 16: a 256-thread server does not need 128 encoder threads; --threads overrides)
+    if (threads <= 0 && !gpu) threads = std::max(1u, std::min(16u, std::thread::hardware_concurrency() / 2));
     if (threads > 0) cp.n_threads = threads;
     if (max_tokens > 0) cp.image_max_tokens = max_tokens;
     if (min_tokens > 0) cp.image_min_tokens = min_tokens;