fix(thread-pool): preserve binding defaults and honor CPU limits

Author: luo

THREAD_POOL_BEHAVIOR.md

@@ -57,7 +57,9 @@ This is a C++ source compatibility change. It intentionally turns ambiguous
 calls into compile errors instead of silently choosing the wrong behavior.
 
 CPU binding is implemented on Linux excluding Android. On unsupported
-platforms, the binding flag does not change thread affinity.
+platforms, the binding flag does not change thread affinity. Linux binding
+enumerates the calling thread's allowed CPU mask, so non-zero-based and
+restricted cpusets are honored. Affinity API failures are logged.
 
 ## DB global query and optimize pools
 
@@ -66,8 +68,8 @@ initialized independently from `CgroupUtil::getCpuLimit()`:
 
 | Configuration | Default count | Default binding |
 |---|---:|---|
-| Query pool | Cgroup CPU limit | Disabled |
-| Optimize pool | Cgroup CPU limit | Disabled |
+| Query pool | Cgroup CPU limit | Enabled |
+| Optimize pool | Cgroup CPU limit | Enabled |
 
 Therefore:
 
@@ -76,6 +78,8 @@ Therefore:
   CPU capacity.
 - In a CPU-limited container, the DB defaults may differ from
   `std::thread::hardware_concurrency()`.
+- Cgroup v2 `cpu.max` values are parsed as `quota period`; `max period` is
+  treated as unlimited and falls back to the online CPU count.
 - Query and optimize counts are independent. Setting `query_threads` does not
   implicitly change `optimize_threads`.
 
@@ -85,7 +89,9 @@ runtime behavior was hardware-concurrency threads with binding enabled. This
 meant configured values such as `optimize_threads=1` were ignored.
 
 After this fix, the configured count and binding are both forwarded
-explicitly, so the effective behavior matches the configuration.
+explicitly, so the effective behavior matches the configuration. Binding
+remains enabled by default to preserve the previous effective behavior; users
+can explicitly disable it when scheduler-managed placement is preferable.
 
 ## Python API
 
@@ -106,21 +112,21 @@ Behavior:
 |---|---|
 | Count is `None` | Use the corresponding cgroup-derived DB default |
 | Count is a positive integer | Use exactly that many threads |
-| Binding is `None` | Use the DB default (`False`) |
+| Binding is `None` | Use the DB default (`True`) |
 | Binding is `False` | Do not bind worker threads |
 | Binding is `True` | Bind worker threads where supported |
 
 Examples:
 
 ```python
-# Both pools use their cgroup-derived counts without binding.
+# Both pools use their cgroup-derived counts with binding.
 zvec.init()
 
-# Optimize uses exactly one unbound worker.
+# Optimize uses exactly one bound worker.
 zvec.init(optimize_threads=1)
 
-# Query uses four bound workers; optimize keeps its independent defaults.
-zvec.init(query_threads=4, query_thread_binding=True)
+# Explicitly let the OS schedule both pools without affinity pinning.
+zvec.init(query_thread_binding=False, optimize_thread_binding=False)
 ```
 
 ## C API
@@ -134,7 +140,7 @@ zvec_config_data_set_optimize_thread_binding(config, binding);
 zvec_config_data_get_optimize_thread_binding(config);
 ```
 
-If these setters are not called, binding remains disabled.
+If these setters are not called, binding remains enabled.
 
 ## Internal call-site migration
 
@@ -146,7 +152,8 @@ behavior:
 - Recall tools with no requested count still use hardware concurrency with
   binding.
 - Explicit benchmark and recall thread counts are now honored.
-- `Index::Merge` uses `write_concurrency` without binding.
+- `Index::Merge` uses `write_concurrency` with binding, preserving its previous
+  effective affinity behavior.
 - DB query and optimize pools use their configured count and binding.
 - Existing call sites already passing both count and binding are unchanged.
 
@@ -165,7 +172,10 @@ ThreadPool(threads, true);
 ## Performance validation
 
 The fix was validated with an end-to-end Python DB API benchmark using the
-SIFT dataset:
+SIFT dataset. The measured post-fix revision used disabled binding defaults.
+The current compatibility-preserving defaults keep binding enabled, so the
+post-fix placement measured below is reproduced by explicitly setting both
+binding options to `False`.
 
 | Parameter | Value |
 |---|---|
@@ -192,7 +202,18 @@ The test machine exposes 64 logical CPUs. The benchmark called:
 zvec.init(optimize_threads=1)
 ```
 
-It did not override the query thread count or either binding option.
+It did not override the query thread count or either binding option. In that
+measured revision the effective binding defaults were `False`; with the
+current defaults, use the following equivalent configuration for the post-fix
+run:
+
+```python
+zvec.init(
+    optimize_threads=1,
+    query_thread_binding=False,
+    optimize_thread_binding=False,
+)
+```
 
 | Pool / behavior | Before the fix | After the fix |
 |---|---|---|
@@ -301,13 +322,17 @@ Behavior is unchanged for:
 - `ThreadPool()`;
 - `ThreadPool(n, false)`;
 - `ThreadPool(n, true)`;
+- `Index::Merge` CPU binding;
+- DB query and optimize pool binding defaults;
 - internal call sites whose old behavior was already explicit and correct.
 
 Behavior intentionally changes for:
 
 - one-argument thread counts that were previously interpreted as booleans;
 - DB thread-count configuration that was previously ignored;
-- `Index::Merge` and recall resize paths affected by ambiguous or reversed
-  arguments;
-- DB binding defaults, which now explicitly default to disabled instead of
-  being accidentally enabled by overload resolution.
+- recall resize paths affected by reversed arguments.
+
+DB binding remains enabled by default, matching the previous effective
+runtime behavior. Adding binding fields to the public C++ `ConfigData` struct
+changes its layout, so consumers linking across a C++ ABI boundary must rebuild
+against the updated headers.

python/tests/detail/test_db_config.py

@@ -157,6 +157,10 @@ def test_query_threads(self):
     def test_query_thread_binding(self):
         zvec.init(query_thread_binding=True)
 
+    @run_in_subprocess
+    def test_query_thread_binding_disabled(self):
+        zvec.init(query_thread_binding=False)
+
     @run_in_subprocess
     def test_query_thread_binding_invalid(self):
         with pytest.raises(TypeError):
@@ -186,6 +190,10 @@ def test_optimize_threads(self):
     def test_optimize_thread_binding(self):
         zvec.init(optimize_thread_binding=True)
 
+    @run_in_subprocess
+    def test_optimize_thread_binding_disabled(self):
+        zvec.init(optimize_thread_binding=False)
+
     @run_in_subprocess
     def test_optimize_thread_binding_invalid(self):
         with pytest.raises(TypeError):

python/zvec/zvec.py

@@ -83,14 +83,14 @@ def init(
             Must be ≥ 1 if provided.
         query_thread_binding (Optional[bool], optional):
             Whether to bind query worker threads to CPU cores. Defaults to
-            Zvec's internal setting, currently ``False``.
+            Zvec's internal setting, currently ``True``.
         optimize_threads (Optional[int], optional):
             Threads for background tasks (e.g., compaction, indexing).
             If ``None`` (default), inferred independently from available CPU
             cores (via cgroup).
         optimize_thread_binding (Optional[bool], optional):
             Whether to bind optimize worker threads to CPU cores. Defaults to
-            Zvec's internal setting, currently ``False``.
+            Zvec's internal setting, currently ``True``.
         invert_to_forward_scan_ratio (Optional[float], optional):
             Threshold to switch from inverted index to full forward scan.
             Range: [0.0, 1.0]. Higher → more aggressive index skipping.

src/ailego/parallel/thread_pool.cc

@@ -12,33 +12,64 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <zvec/ailego/logger/logger.h>
 #include <zvec/ailego/parallel/thread_pool.h>
 
-#if (defined(__linux) || defined(__linux__)) && !defined(__ANDROID__)
+#if defined(__linux__) && !defined(__ANDROID__)
 #include <pthread.h>
 
+static inline bool GetAllowedCpuMask(cpu_set_t *mask) {
+  CPU_ZERO(mask);
+  const int error = pthread_getaffinity_np(pthread_self(), sizeof(*mask), mask);
+  if (error != 0) {
+    LOG_WARN("Failed to get thread affinity mask, error[%d]", error);
+    return false;
+  }
+  return true;
+}
+
 static inline void BindThreads(std::vector<std::thread> &pool) {
-  uint32_t hc = std::thread::hardware_concurrency();
-  if (hc > 1) {
-    cpu_set_t mask;
-
-    for (size_t i = 0u; i < pool.size(); ++i) {
-      CPU_ZERO(&mask);
-      CPU_SET(i % hc, &mask);
-      pthread_setaffinity_np(pool[i].native_handle(), sizeof(mask), &mask);
+  cpu_set_t allowed_mask;
+  if (!GetAllowedCpuMask(&allowed_mask)) {
+    return;
+  }
+
+  std::vector<int> allowed_cpus;
+  for (int cpu = 0; cpu < CPU_SETSIZE; ++cpu) {
+    if (CPU_ISSET(cpu, &allowed_mask)) {
+      allowed_cpus.push_back(cpu);
+    }
+  }
+  if (allowed_cpus.empty()) {
+    LOG_WARN("Cannot bind thread pool: affinity mask has no allowed CPUs");
+    return;
+  }
+
+  cpu_set_t target_mask;
+  for (size_t i = 0u; i < pool.size(); ++i) {
+    CPU_ZERO(&target_mask);
+    CPU_SET(allowed_cpus[i % allowed_cpus.size()], &target_mask);
+    const int error = pthread_setaffinity_np(pool[i].native_handle(),
+                                             sizeof(target_mask), &target_mask);
+    if (error != 0) {
+      LOG_WARN("Failed to bind thread pool worker[%zu], error[%d]", i, error);
     }
   }
 }
 
 static inline void UnbindThreads(std::vector<std::thread> &pool) {
   cpu_set_t mask;
   CPU_ZERO(&mask);
-
-  for (size_t i = 0u; i < CPU_SETSIZE; ++i) {
-    CPU_SET(i, &mask);
+  for (int cpu = 0; cpu < CPU_SETSIZE; ++cpu) {
+    CPU_SET(cpu, &mask);
   }
+
   for (size_t i = 0u; i < pool.size(); ++i) {
-    pthread_setaffinity_np(pool[i].native_handle(), sizeof(mask), &mask);
+    const int error =
+        pthread_setaffinity_np(pool[i].native_handle(), sizeof(mask), &mask);
+    if (error != 0) {
+      LOG_WARN("Failed to unbind thread pool worker[%zu], error[%d]", i, error);
+    }
   }
 }
 #else
@@ -130,4 +161,4 @@ bool ThreadPool::picking(ThreadPool::Task *task) {
 }
 
 }  // namespace ailego
-}  // namespace zvec
+}  // namespace zvec

src/core/interface/index.cc

@@ -986,7 +986,7 @@ int Index::Merge(const std::vector<Index::Pointer> &indexes,
     reducer->set_thread_pool(options.pool);
   } else {
     local_thread_pool =
-        std::make_unique<ailego::ThreadPool>(options.write_concurrency, false);
+        std::make_unique<ailego::ThreadPool>(options.write_concurrency, true);
     reducer->set_thread_pool(local_thread_pool.get());
   }
 

src/db/common/cgroup_util.cc

@@ -13,6 +13,8 @@
 // limitations under the License.
 
 #include "db/common/cgroup_util.h"
+#include <charconv>
+#include <system_error>
 
 namespace zvec {
 
@@ -114,19 +116,15 @@ bool CgroupUtil::readCpuCgroup() {
   // cgroup v2
   std::ifstream file("/sys/fs/cgroup/cpu.max");
   if (file.is_open()) {
-    uint64_t quota, period;
-    char slash;
-    file >> quota >> slash >> period;
-    file.close();
+    std::string cpu_max;
+    std::getline(file, cpu_max);
 
-    if (quota != std::numeric_limits<uint64_t>::max() && quota != 0 &&
-        period > 0) {
-      cpu_cores_ =
-          static_cast<int>(std::ceil(static_cast<double>(quota) / period));
+    int cpu_cores = 0;
+    if (parseCpuMax(cpu_max, &cpu_cores)) {
+      cpu_cores_ = cpu_cores;
       return true;
-    } else {
-      return false;
     }
+    return false;
   }
 
   // cgroup v1
@@ -150,6 +148,43 @@ bool CgroupUtil::readCpuCgroup() {
   return false;
 }
 
+bool CgroupUtil::parseCpuMax(const std::string &cpu_max, int *cpu_cores) {
+  if (cpu_cores == nullptr) {
+    return false;
+  }
+
+  std::istringstream stream(cpu_max);
+  std::string quota_token;
+  uint64_t period = 0;
+  if (!(stream >> quota_token >> period) || period == 0) {
+    return false;
+  }
+
+  std::string trailing_token;
+  if (stream >> trailing_token) {
+    return false;
+  }
+  if (quota_token == "max") {
+    return false;
+  }
+
+  uint64_t quota = 0;
+  const auto result = std::from_chars(
+      quota_token.data(), quota_token.data() + quota_token.size(), quota);
+  if (result.ec != std::errc{} ||
+      result.ptr != quota_token.data() + quota_token.size() || quota == 0) {
+    return false;
+  }
+
+  const uint64_t cores = quota / period + (quota % period != 0 ? 1 : 0);
+  if (cores > static_cast<uint64_t>(std::numeric_limits<int>::max())) {
+    return false;
+  }
+
+  *cpu_cores = static_cast<int>(cores);
+  return true;
+}
+
 void CgroupUtil::updateMemoryLimit() {
   if (readMemoryCgroup()) {
     return;
@@ -491,4 +526,4 @@ double CgroupUtil::calculateMacOSCpuUsage() {
 }
 #endif
 
-}  // namespace zvec
+}  // namespace zvec

src/db/common/cgroup_util.h

@@ -45,6 +45,10 @@ class CgroupUtil {
   static int getCpuLimit();
   static uint64_t getMemoryLimit();
 
+  // Parse a cgroup v2 cpu.max value. Returns false for an unlimited or
+  // malformed value so callers can fall back to the host CPU count.
+  static bool parseCpuMax(const std::string &cpu_max, int *cpu_cores);
+
   // Static methods to get other resources
   static double getCpuUsage();
   static uint64_t getMemoryUsage();
@@ -101,4 +105,4 @@ class CgroupUtil {
 #endif
 };
 
-}  // namespace zvec
+}  // namespace zvec