From 57119f54ab37dfb0d6093bb5b3d6b595c72a8f26 Mon Sep 17 00:00:00 2001 From: Andres Rios Tascon Date: Thu, 27 Aug 2026 14:32:55 -0400 Subject: [PATCH 1/3] fix: don't terminate the process when the thread pool can't be created `init_threads` called `exit(-1)` if `pthread_create` failed, which kills the host process rather than the library. On platforms without working threads (WebAssembly under Emscripten/Pyodide) every thread creation fails with EAGAIN, so `import numexpr` with `OMP_NUM_THREADS` or `NUMEXPR_MAX_THREADS` set, or any `set_num_threads(n)` with n > 1, tore down the whole interpreter. Fall back to serial evaluation instead: - `init_threads` returns -1 and leaves `gs.nthreads == 1` when the pool cannot be brought up, unwinding any threads that did start. - The pool shutdown sequence in `numexpr_set_nthreads` is factored out into `join_threads` so the partial-failure path can reuse it. - The signal-mask calls and `pthread_join` now warn instead of exiting, so no `exit()` calls remain in the extension. - `numexpr.nthreads` is read back from the VM after initialization so it reports the pool that actually exists rather than the one requested. Also fix `print_versions` on platforms with no dedicated `CPUInfo` subclass (Emscripten, FreeBSD, AIX): `cpu.info` resolved through `CPUInfoBase.__getattr__` to `lambda: None`, so `cpu.info[0]` raised `TypeError`, which the surrounding `except KeyError` did not catch. Assisted-by: claude-code:claude-opus-5[1m] --- numexpr/__init__.py | 8 +- numexpr/cpuinfo.py | 5 ++ numexpr/module.cpp | 147 ++++++++++++++++++++++------------ numexpr/tests/test_numexpr.py | 7 +- 4 files changed, 111 insertions(+), 56 deletions(-) diff --git a/numexpr/__init__.py b/numexpr/__init__.py index 63bb9e9..ec079c5 100644 --- a/numexpr/__init__.py +++ b/numexpr/__init__.py @@ -38,8 +38,12 @@ # Detect the number of cores ncores = detect_number_of_cores() -# Initialize the number of threads to be used -nthreads = _init_num_threads() +# Initialize the number of threads to be used. `_init_num_threads` returns the +# number of threads it requested; read back what the VM actually ended up with, +# since it falls back to serial evaluation on platforms that cannot create a +# thread pool (WebAssembly under Emscripten/Pyodide, for instance). +_init_num_threads() +nthreads = get_num_threads() # The default for VML is 1 thread (see #39) # set_vml_num_threads(1) diff --git a/numexpr/cpuinfo.py b/numexpr/cpuinfo.py index 897a4ca..76020c9 100755 --- a/numexpr/cpuinfo.py +++ b/numexpr/cpuinfo.py @@ -86,6 +86,11 @@ class CPUInfoBase(object): the availability of various CPU features. """ + # Platforms without a dedicated subclass (WebAssembly under Emscripten, + # for instance) fall back to this class, so `info` has to exist and be + # subscriptable rather than resolve through `__getattr__`. + info = [] + def _try_call(self, func): try: return func() diff --git a/numexpr/module.cpp b/numexpr/module.cpp index 67629bd..94c20bf 100644 --- a/numexpr/module.cpp +++ b/numexpr/module.cpp @@ -193,10 +193,55 @@ void *th_worker(void *tidptr) return(0); } -/* Initialize threads */ +/* + * Tear down the running thread pool. Assumes `gs.nthreads` matches the number + * of live workers and that they are (or will be) parked on the barrier. + */ +static void join_threads(void) +{ + int t, rc; + void *status; + + /* Tell all existing threads to finish */ + gs.end_threads = 1; + pthread_mutex_lock(&gs.count_threads_mutex); + if (gs.count_threads < gs.nthreads) { + gs.count_threads++; + do { + pthread_cond_wait(&gs.count_threads_cv, + &gs.count_threads_mutex); + } while (!gs.barrier_passed); + } + else { + gs.barrier_passed = 1; + pthread_cond_broadcast(&gs.count_threads_cv); + } + pthread_mutex_unlock(&gs.count_threads_mutex); + + /* Join exiting threads */ + for (t = 0; t < gs.nthreads; t++) { + rc = pthread_join(gs.threads[t], &status); + if (rc) { + fprintf(stderr, + "ERROR; return code from pthread_join() is %d\n", rc); + fprintf(stderr, "\tError detail: %s\n", strerror(rc)); + } + } + gs.init_threads_done = 0; + gs.end_threads = 0; +} + +/* + * Initialize the thread pool. Returns 0 on success and -1 if the pool could + * not be created, in which case numexpr falls back to running serially with + * `gs.nthreads == 1`. Platforms without working threads (WebAssembly under + * Emscripten/Pyodide, for instance) always take the fallback path, so this + * must never terminate the host process. + */ int init_threads(void) { - int tid, rc; + int tid, rc, created; + int masked = 0; if ( !(gs.nthreads > 1 && (!gs.init_threads_done || gs.pid != getpid())) ) { /* Thread pool must always be initialized once and once only. */ @@ -216,20 +261,25 @@ int init_threads(void) /* * Our worker threads should not deal with signals from the rest of the * application - mask everything temporarily in this thread, so our workers - * can inherit that mask + * can inherit that mask. Failing to mask is not fatal, it just means the + * workers may see signals meant for the main thread. */ sigset_t sigset_block_all, sigset_restore; rc = sigfillset(&sigset_block_all); if (rc != 0) { - fprintf(stderr, "ERROR; failed to block signals: sigfillset: %s", + fprintf(stderr, "WARNING; failed to block signals: sigfillset: %s\n", strerror(rc)); - exit(-1); } - rc = pthread_sigmask( SIG_BLOCK, &sigset_block_all, &sigset_restore); - if (rc != 0) { - fprintf(stderr, "ERROR; failed to block signals: pthread_sigmask: %s", - strerror(rc)); - exit(-1); + else { + rc = pthread_sigmask( SIG_BLOCK, &sigset_block_all, &sigset_restore); + if (rc != 0) { + fprintf(stderr, + "WARNING; failed to block signals: pthread_sigmask: %s\n", + strerror(rc)); + } + else { + masked = 1; + } } /* Now create the threads */ @@ -238,23 +288,47 @@ int init_threads(void) rc = pthread_create(&gs.threads[tid], NULL, th_worker, (void *)&gs.tids[tid]); if (rc) { - fprintf(stderr, - "ERROR; return code from pthread_create() is %d\n", rc); - fprintf(stderr, "\tError detail: %s\n", strerror(rc)); - exit(-1); + break; } } + created = tid; /* * Restore the signal mask so the main thread can process signals as * expected */ - rc = pthread_sigmask( SIG_SETMASK, &sigset_restore, NULL); - if (rc != 0) { + if (masked) { + int rc_mask = pthread_sigmask( SIG_SETMASK, &sigset_restore, NULL); + if (rc_mask != 0) { + fprintf(stderr, + "WARNING: failed to restore signal mask: pthread_sigmask: %s\n", + strerror(rc_mask)); + } + } + + if (created < gs.nthreads) { + /* + * The pool could not be brought up. Wind back whatever we did manage + * to start and fall back to serial execution rather than killing the + * host process. + */ fprintf(stderr, - "ERROR: failed to restore signal mask: pthread_sigmask: %s", - strerror(rc)); - exit(-1); + "WARNING; return code from pthread_create() is %d\n", rc); + fprintf(stderr, "\tError detail: %s\n", strerror(rc)); + fprintf(stderr, + "\tNumExpr could not start a thread pool of %d threads, " + "falling back to serial evaluation.\n", gs.nthreads); + if (created > 0) { + /* `join_threads` drives the barrier off `gs.nthreads`, so make it + match the number of workers that actually started. */ + gs.nthreads = created; + gs.init_threads_done = 1; + gs.pid = (int)getpid(); + join_threads(); + } + gs.nthreads = 1; + gs.init_threads_done = 0; + return(-1); } gs.init_threads_done = 1; /* Initialization done! */ @@ -267,8 +341,6 @@ int init_threads(void) int numexpr_set_nthreads(int nthreads_new) { int nthreads_old = gs.nthreads; - int t, rc; - void *status; // if (nthreads_new > MAX_THREADS) { // fprintf(stderr, @@ -291,38 +363,11 @@ int numexpr_set_nthreads(int nthreads_new) different from that in pid var (probably means that we are a subprocess, and thus threads are non-existent). */ if (gs.nthreads > 1 && gs.init_threads_done && gs.pid == getpid()) { - /* Tell all existing threads to finish */ - gs.end_threads = 1; - pthread_mutex_lock(&gs.count_threads_mutex); - if (gs.count_threads < gs.nthreads) { - gs.count_threads++; - do { - pthread_cond_wait(&gs.count_threads_cv, - &gs.count_threads_mutex); - } while (!gs.barrier_passed); - } - else { - gs.barrier_passed = 1; - pthread_cond_broadcast(&gs.count_threads_cv); - } - pthread_mutex_unlock(&gs.count_threads_mutex); - - /* Join exiting threads */ - for (t=0; t Date: Thu, 27 Aug 2026 14:35:40 -0400 Subject: [PATCH 2/3] test: skip tests needing processes or threads where the platform has neither Eight tests need to spawn an interpreter or start OS threads to test what they test, so they fail rather than skip on WebAssembly runtimes (Pyodide/Emscripten, WASI), which provide neither: - `test_locals_clears_globals` needs a top-level frame, so it has to run in a separate interpreter. - `test_threading_config.test_max_threads_{set,unset}` test import-time configuration, which needs a fresh process to re-run the extension's init. - `test_threading.*` needs real threads. - `test_subprocess.test_multiprocess` needs the `_multiprocessing` extension. Gate them on capability constants. `_thread` and `os.fork` are both present but non-functional under Emscripten, so threads are detected by actually starting one and processes by platform, matching what CPython's own test suite does. The skips use `unittest.skipUnless` rather than pytest markers so they work under `numexpr.test()` too, which runs the suite through `unittest`. Assisted-by: claude-code:claude-opus-5[1m] --- numexpr/tests/test_numexpr.py | 28 ++++++++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/numexpr/tests/test_numexpr.py b/numexpr/tests/test_numexpr.py index 7bda75a..ad5461a 100644 --- a/numexpr/tests/test_numexpr.py +++ b/numexpr/tests/test_numexpr.py @@ -10,6 +10,7 @@ #################################################################### import gc +import importlib.util import os import platform import subprocess @@ -53,6 +54,28 @@ MAX_THREADS = 16 +def _can_start_thread(): + """Whether the interpreter can actually start an OS thread.""" + import threading + thread = threading.Thread(target=lambda: None) + try: + thread.start() + except RuntimeError: + return False + thread.join() + return True + + +# WebAssembly runtimes (Pyodide/Emscripten, WASI) have no process support, and +# the standard builds are single-threaded. A handful of tests below need to +# spawn an interpreter or start threads to test what they test, so they cannot +# run there. `_thread` and `os.fork` are both present-but-non-functional under +# Emscripten, hence the platform check for processes and the probe for threads. +HAS_SUBPROCESS = sys.platform not in ('emscripten', 'wasi') +HAS_MULTIPROCESSING = importlib.util.find_spec('_multiprocessing') is not None +HAS_THREADS = _can_start_thread() + + if not pytest_available: def identity(f): return f @@ -349,6 +372,7 @@ def _test_refcount_disable_cache(self): assert sys.getrefcount(b) == 2 @pytest.mark.thread_unsafe + @unittest.skipUnless(HAS_SUBPROCESS, 'requires process support') def test_locals_clears_globals(self): # Check for issue #313, whereby clearing f_locals also clear f_globals # if in the top-frame. This cannot be done inside `unittest` as it is always @@ -1350,6 +1374,7 @@ def _environment(key, value): # Test cases for the threading configuration @pytest.mark.thread_unsafe class test_threading_config(TestCase): + @unittest.skipUnless(HAS_SUBPROCESS, 'requires process support') def test_max_threads_unset(self): # Has to be done in a subprocess as `importlib.reload` doesn't let us # re-initialize the threadpool @@ -1362,6 +1387,7 @@ def test_max_threads_unset(self): "exit(0)"]) subprocess.check_call([sys.executable, '-c', script]) + @unittest.skipUnless(HAS_SUBPROCESS, 'requires process support') def test_max_threads_set(self): # Has to be done in a subprocess as `importlib.reload` doesn't let us # re-initialize the threadpool @@ -1415,6 +1441,7 @@ def test_vml_threads_round_trip(self): # Case test for threads +@unittest.skipUnless(HAS_THREADS, 'requires working threads') class test_threading(TestCase): def test_thread(self): @@ -1531,6 +1558,7 @@ def _worker(qout=None): # Case test for subprocesses (via multiprocessing module) +@unittest.skipUnless(HAS_MULTIPROCESSING, 'requires the multiprocessing module') class test_subprocess(TestCase): @pytest.mark.thread_unsafe def test_multiprocess(self): From e83eab89d79ba0a2d24145d093f01120fcb29b59 Mon Sep 17 00:00:00 2001 From: Andres Rios Tascon Date: Thu, 27 Aug 2026 14:55:05 -0400 Subject: [PATCH 3/3] ci: build and test wheels for Pyodide Add a Pyodide entry to the wheel matrix so wasm32 wheels are built alongside the others. cibuildwheel provides the Emscripten SDK and the Pyodide runtime and runs `test-command` inside it, so these wheels are tested under WebAssembly rather than only built. No changes to the cibuildwheel configuration in `pyproject.toml` were needed; `CIBW_ARCHS_LINUX` is ignored for this platform. This requires cibuildwheel 4.x, which does not support Pyodide in the 3.1.3 previously used, so bump the action for every platform. Two consequences: - `CIBW_ENABLE: cpython-freethreading` is removed. That enable group no longer exists in 4.x because free-threaded builds are on by default, and passing it is a hard error. The free-threaded targets are unaffected: cp314t is still built and cp313t still skipped via `pyproject.toml`. - CPython 3.15 is now in cibuildwheel's default target set, so cp315 and cp315t wheels are produced for every platform. Assisted-by: claude-code:claude-opus-5[1m] --- .github/workflows/build.yml | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 7ae0f54..56382e5 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -14,7 +14,7 @@ jobs: env: CIBW_ARCHS_LINUX: ${{ matrix.arch }} CIBW_ARCHS_MACOS: "x86_64 arm64" - CIBW_ENABLE: cpython-freethreading + CIBW_PLATFORM: ${{ matrix.platform || 'auto' }} strategy: fail-fast: false @@ -57,6 +57,13 @@ jobs: artifact_name: "macos-universal2" python-version: "3.x" + # Pyodide / WebAssembly (build wheels). + - os: ubuntu-latest + arch: wasm32 + platform: pyodide + artifact_name: "pyodide-wasm32" + python-version: "3.x" + steps: - uses: actions/checkout@v3 @@ -79,7 +86,7 @@ jobs: # - Python version is "3.x" - name: Build wheels if: ${{ !matrix.numpy-version }} - uses: pypa/cibuildwheel@v3.1.3 + uses: pypa/cibuildwheel@v4.2.0 - name: Make sdist if: ${{ matrix.os == 'windows-latest' && !matrix.numpy-version }}