Expose FFI runtime worker pool control

vortex-ffi/README.md

@@ -1,5 +1,22 @@
 # Vortex C bindings
 
+## Runtime threading
+
+The FFI uses a shared, caller-driven runtime. By default Vortex creates no runtime worker threads:
+the host threads currently executing FFI calls drive the runtime. Multiple host threads making
+concurrent FFI calls can drive runtime work in parallel while keeping thread ownership entirely in
+the host application.
+
+Applications that want a single FFI operation to make progress on additional threads may opt into
+Vortex-owned background workers with `vx_runtime_set_worker_threads`. This is a process-global
+setting shared by every FFI session. Calling it with a non-zero count changes the threading model
+from host-thread-only execution to a combination of host threads and Vortex-owned workers. Calling
+it with zero signals the background workers to stop and restores the host-thread-only
+configuration.
+
+Applications that already supply concurrency through their own host threads should leave the
+worker count at its default of zero to avoid oversubscription.
+
 ## Updating Headers
 
 If you're developing FFI and want to rebuild `cinclude/vortex.h`, run:

vortex-ffi/cinclude/vortex.h

@@ -681,6 +681,27 @@ typedef struct {
 extern "C" {
 #endif // __cplusplus
 
+/**
+ * Set the number of background worker threads driving the shared FFI runtime.
+ *
+ * Calling this with a non-zero count opts the process into a Vortex-owned thread pool. These
+ * background threads drive the same executor as host threads currently inside FFI calls. If this
+ * function is never called, Vortex creates no runtime worker threads and execution remains
+ * entirely host-thread-driven.
+ *
+ * This setting is process-global and affects all FFI sessions. Passing zero restores the
+ * host-thread-only configuration by signalling all background workers to stop. Increasing the
+ * count starts workers immediately; decreasing it signals excess workers to stop.
+ */
+void vx_runtime_set_worker_threads(size_t worker_threads);
+
+/**
+ * Return the configured number of Vortex-owned background worker threads.
+ *
+ * Zero means the runtime is entirely driven by host threads entering FFI calls.
+ */
+size_t vx_runtime_worker_count(void);
+
 /**
  * Increase reference count on vx_array
  */

vortex-ffi/src/lib.rs

@@ -45,15 +45,48 @@ use vortex::dtype::FieldName;
 use vortex::error::VortexResult;
 use vortex::error::vortex_ensure;
 use vortex::io::runtime::current::CurrentThreadRuntime;
+use vortex::io::runtime::current::CurrentThreadWorkerPool;
 
 #[cfg(all(feature = "mimalloc", not(miri)))]
 #[global_allocator]
 static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
 
-/// A shared runtime for all FFI operations.
-// TODO(ngates): also create a CurrentThreadPool to manage background worker threads.
+/// A shared, caller-driven runtime for all FFI operations.
+///
+/// By default the runtime owns no threads: host threads drive its executor while they are inside
+/// an FFI call. Concurrent calls from multiple host threads can therefore execute runtime tasks in
+/// parallel without any Vortex-owned worker threads.
 static RUNTIME: LazyLock<CurrentThreadRuntime> = LazyLock::new(CurrentThreadRuntime::new);
 
+/// Optional Vortex-owned background workers that drive the shared FFI runtime.
+///
+/// Creating the pool does not create threads. It starts empty so callers retain the default
+/// host-thread-owned execution model until they opt in with [`vx_runtime_set_worker_threads`].
+static POOL: LazyLock<CurrentThreadWorkerPool> = LazyLock::new(|| RUNTIME.new_pool());
+
+/// Set the number of background worker threads driving the shared FFI runtime.
+///
+/// Calling this with a non-zero count opts the process into a Vortex-owned thread pool. These
+/// background threads drive the same executor as host threads currently inside FFI calls. If this
+/// function is never called, Vortex creates no runtime worker threads and execution remains
+/// entirely host-thread-driven.
+///
+/// This setting is process-global and affects all FFI sessions. Passing zero restores the
+/// host-thread-only configuration by signalling all background workers to stop. Increasing the
+/// count starts workers immediately; decreasing it signals excess workers to stop.
+#[unsafe(no_mangle)]
+pub extern "C" fn vx_runtime_set_worker_threads(worker_threads: usize) {
+    POOL.set_workers(worker_threads);
+}
+
+/// Return the configured number of Vortex-owned background worker threads.
+///
+/// Zero means the runtime is entirely driven by host threads entering FFI calls.
+#[unsafe(no_mangle)]
+pub extern "C" fn vx_runtime_worker_count() -> usize {
+    POOL.worker_count()
+}
+
 /// Return the shared FFI runtime for layered FFI crates that drive Vortex streams produced through
 /// `vortex-ffi`.
 ///
@@ -106,6 +139,18 @@ mod tests {
     use crate::sink::vx_array_sink_open_file;
     use crate::sink::vx_array_sink_push;
     use crate::string::vx_view;
+    use crate::vx_runtime_set_worker_threads;
+    use crate::vx_runtime_worker_count;
+
+    #[test]
+    #[cfg_attr(miri, ignore)]
+    fn runtime_worker_pool_configuration() {
+        assert_eq!(vx_runtime_worker_count(), 0);
+        vx_runtime_set_worker_threads(2);
+        assert_eq!(vx_runtime_worker_count(), 2);
+        vx_runtime_set_worker_threads(0);
+        assert_eq!(vx_runtime_worker_count(), 0);
+    }
 
     /// Panic if error is NULL. Free the error if it's not
     pub(crate) fn assert_error(error: *mut vx_error) {

vortex-ffi/tests/default_runtime.rs

@@ -0,0 +1,22 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright the Vortex contributors
+
+use std::thread;
+
+use vortex::io::runtime::BlockingRuntime;
+use vortex_ffi::ffi_runtime;
+use vortex_ffi::vx_runtime_worker_count;
+
+#[test]
+#[cfg_attr(miri, ignore)]
+fn runtime_is_host_thread_driven_by_default() {
+    assert_eq!(vx_runtime_worker_count(), 0);
+
+    let host_thread = thread::current().id();
+    let task = ffi_runtime()
+        .handle()
+        .spawn(async move { thread::current().id() });
+    let executor_thread = ffi_runtime().block_on(task);
+
+    assert_eq!(executor_thread, host_thread);
+}