Make numexpr work under Pyodide/WebAssembly

.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 }}

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)
 

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()

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<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));
-                exit(-1);
-            }
-        }
-        gs.init_threads_done = 0;
-        gs.end_threads = 0;
+        join_threads();
     }
 
-    /* Launch a new pool of threads (if necessary) */
+    /* Launch a new pool of threads (if necessary).  If the pool cannot be
+       created, `init_threads` leaves us in serial mode. */
     gs.nthreads = nthreads_new;
     init_threads();
 

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):
@@ -1566,13 +1594,14 @@ def print_versions():
     (sysname, nodename, release, os_version, machine, processor) = platform.uname()
     print('Platform:          %s-%s-%s' % (sys.platform, machine, os_version))
     try:
-        # cpuinfo doesn't work on OSX well it seems, so protect these outputs
-        # with a try block
+        # cpuinfo doesn't work on OSX well it seems, and platforms with no
+        # dedicated CPUInfo subclass report nothing at all, so protect these
+        # outputs with a try block
         cpu_info = cpu.info[0]
         print('CPU vendor:        %s' % cpu_info.get('VendorIdentifier', ''))
         print('CPU model:         %s' % cpu_info.get('ProcessorNameString', ''))
         print('CPU clock speed:   %s MHz' % cpu_info.get('~MHz',''))
-    except KeyError:
+    except (KeyError, IndexError, TypeError):
         pass
     print('VML available?     %s' % use_vml)
     if use_vml: