Run more low level thread primitive tests under pthreads. NFC

test/test_browser.py

@@ -5171,7 +5171,7 @@ def test_wasm_worker_cancel_all_wait_asyncs_at_address(self):
   # Tests emscripten_lock_init(), emscripten_lock_waitinf_acquire() and emscripten_lock_release()
   @also_with_minimal_runtime
   def test_wasm_worker_lock_waitinf(self):
-    self.btest('wasm_worker/lock_waitinf_acquire.c', expected='4000', cflags=['-sWASM_WORKERS'])
+    self.btest('wasm_worker/lock_waitinf_acquire.c', expected='0', cflags=['-sWASM_WORKERS'])
 
   # Tests emscripten_lock_wait_acquire() and emscripten_lock_try_acquire() in Worker.
   @also_with_minimal_runtime
@@ -5202,7 +5202,7 @@ def test_wasm_worker_lock_busyspin_wait(self):
   # Tests emscripten_lock_busyspin_waitinf_acquire() in Worker and main thread.
   @also_with_minimal_runtime
   def test_wasm_worker_lock_busyspin_waitinf(self):
-    self.btest('wasm_worker/lock_busyspin_waitinf_acquire.c', expected='1', cflags=['-sWASM_WORKERS'])
+    self.btest('wasm_worker/lock_busyspin_waitinf_acquire.c', expected='0', cflags=['-sWASM_WORKERS'])
 
   # Tests that proxied JS functions cannot be called from Wasm Workers
   @also_with_minimal_runtime
@@ -5222,7 +5222,7 @@ def test_wasm_worker_semaphore_waitinf_acquire(self):
   # Tests emscripten_semaphore_try_acquire() on the main thread
   @also_with_minimal_runtime
   def test_wasm_worker_semaphore_try_acquire(self):
-    self.btest('wasm_worker/semaphore_try_acquire.c', expected='0', cflags=['-sWASM_WORKERS'])
+    self.btest_exit('wasm_worker/semaphore_try_acquire.c', cflags=['-sWASM_WORKERS'])
 
   # Tests that calling any proxied function in a Wasm Worker will abort at runtime when ASSERTIONS are enabled.
   def test_wasm_worker_proxied_function(self):

test/test_core.py

@@ -283,6 +283,23 @@ def metafunc(self, asyncify, *args, **kwargs):
   return metafunc
 
 
+def also_with_wasm_workers(func):
+  assert callable(func)
+
+  @wraps(func)
+  def metafunc(self, ww, *args, **kwargs):
+    if ww:
+      if self.get_setting('WASM_ESM_INTEGRATION'):
+        self.skipTest('WASM_ESM_INTEGRATION is not compatible with WASM_WORKERS')
+      if is_sanitizing(self.cflags):
+        self.skipTest('sanitizers are not compatible with WASM_WORKERS')
+      self.cflags += ['-sWASM_WORKERS']
+    return func(self, *args, **kwargs)
+
+  parameterize(metafunc, {'': (False,), 'ww': (True,)})
+  return metafunc
+
+
 def no_optimize(note=''):
   assert not callable(note)
 
@@ -2749,12 +2766,33 @@ def test_pthread_run_on_main_thread(self):
   def test_pthread_is_lock_free(self):
     self.do_runf('pthread/is_lock_free.c', 'done\n', cflags=['-pthread'])
 
+  # These wasm_worker tests are also tested in pure wasm-worker mode in test_browser.py.
+  # We test them here in pthread-mode and in pthread+wasm-worker mode.
+
+  @requires_pthreads
+  @also_with_wasm_workers
+  def test_emscripten_lock_waitinf_acquire(self):
+    self.do_runf('wasm_worker/lock_waitinf_acquire.c', 'done\n', cflags=['-pthread', '-sPTHREAD_POOL_SIZE=4'])
+
   @requires_pthreads
+  @also_with_wasm_workers
   def test_emscripten_lock_wait_acquire(self):
     self.do_runf('wasm_worker/lock_wait_acquire.c', 'done\n', cflags=['-pthread'])
-    if not is_sanitizing(self.cflags) and not self.get_setting('WASM_ESM_INTEGRATION'):
-      # Also test the pthreads + WASM_WORKERS combination
-      self.do_runf('wasm_worker/lock_wait_acquire.c', 'done\n', cflags=['-pthread', '-sWASM_WORKERS'])
+
+  @requires_pthreads
+  @also_with_wasm_workers
+  def test_emscripten_lock_busyspin_waitinf(self):
+    self.do_runf('wasm_worker/lock_busyspin_waitinf_acquire.c', 'done\n', cflags=['-pthread'])
+
+  @requires_pthreads
+  @also_with_wasm_workers
+  def test_emscripten_semaphore_waitinf_acquire(self):
+    self.do_runf('wasm_worker/semaphore_waitinf_acquire.c', 'done\n', cflags=['-pthread', '-sPTHREAD_POOL_SIZE=7'])
+
+  @requires_pthreads
+  @also_with_wasm_workers
+  def test_emscripten_semaphore_try_acquire(self):
+    self.do_runf('wasm_worker/semaphore_try_acquire.c', 'done\n', cflags=['-pthread'])
 
   def test_tcgetattr(self):
     self.do_runf('termios/test_tcgetattr.c', 'success')

test/wasm_worker/lock_busyspin_waitinf_acquire.c

@@ -1,8 +1,15 @@
+#include <assert.h>
 #include <emscripten.h>
-#include <emscripten/wasm_worker.h>
 #include <emscripten/threading.h>
 #include <stdlib.h>
-#include <assert.h>
+
+// This test can be run under pthreads *or* Wasm Workers
+#ifdef __EMSCRIPTEN_PTHREADS__
+#include <pthread.h>
+_Atomic bool done = false;
+#else
+#include <emscripten/wasm_worker.h>
+#endif
 
 // Tests emscripten_lock_busyspin_waitinf_acquire().
 
@@ -11,29 +18,63 @@ emscripten_lock_t lock = EMSCRIPTEN_LOCK_T_STATIC_INITIALIZER;
 volatile int sharedState = 0;
 
 void worker_main() {
+  emscripten_out("worker_main");
   emscripten_lock_busyspin_waitinf_acquire(&lock);
   emscripten_atomic_add_u32((void*)&sharedState, 1);
-#ifdef REPORT_RESULT
-  REPORT_RESULT(sharedState);
+  assert(sharedState == 1);
+#ifdef __EMSCRIPTEN_PTHREADS__
+  emscripten_out("done");
+  done = true;
+  exit(0);
+#else
+  REPORT_RESULT(0);
 #endif
 }
 
+#ifdef __EMSCRIPTEN_PTHREADS__
+pthread_t t;
+
+void* pthread_main(void* arg) {
+  worker_main();
+  return NULL;
+}
+
+void nothing(void* userData) {
+  if (!done) {
+    emscripten_set_timeout(nothing, 100, 0);
+  }
+}
+#else
 char stack[1024];
+#endif
 
 void releaseLock(void *userData) {
+  emscripten_out("releaseLock");
   emscripten_atomic_sub_u32((void*)&sharedState, 1);
   emscripten_lock_release(&lock);
 }
 
 int main() {
+  emscripten_out("in main");
   // Acquire the lock at startup.
   emscripten_lock_busyspin_waitinf_acquire(&lock);
   emscripten_atomic_add_u32((void*)&sharedState, 1);
 
+#ifdef __EMSCRIPTEN_PTHREADS__
+  // Spawn a Pthread to try to take the lock. It will succeed only after
+  // releaseLock() gets called.
+  pthread_create(&t, NULL, pthread_main, NULL);
+  // Add an infinite timeout to make sure the node runtime stays alive
+  // after main returns.
+  // See https://github.com/emscripten-core/emscripten/issues/23092
+  emscripten_set_timeout(nothing, 100, 0);
+#else
   // Spawn a Worker to try to take the lock. It will succeed only after releaseLock()
   // gets called.
   emscripten_wasm_worker_t worker = emscripten_create_wasm_worker(stack, sizeof(stack));
   emscripten_wasm_worker_post_function_v(worker, worker_main);
+#endif
 
   emscripten_set_timeout(releaseLock, 1000, 0);
+  return 0;
 }

test/wasm_worker/lock_waitinf_acquire.c

@@ -1,9 +1,26 @@
+#include <assert.h>
 #include <emscripten.h>
-#include <emscripten/wasm_worker.h>
 #include <emscripten/em_math.h>
 #include <emscripten/threading.h>
 #include <stdlib.h>
-#include <assert.h>
+
+// This test can be run under pthreads *or* Wasm Workers
+#ifdef __EMSCRIPTEN_PTHREADS__
+#include <pthread.h>
+#endif
+
+#ifdef __EMSCRIPTEN_WASM_WORKERS__
+#include <emscripten/wasm_worker.h>
+#else
+// When WASM_WORKERS is not defined we create dummy/fake version of
+// emscripten_wasm_worker_self_id and emscripten_wasm_worker_sleep.
+#define _GNU_SOURCE
+#include <unistd.h>
+int emscripten_wasm_worker_self_id() { return gettid(); }
+void emscripten_wasm_worker_sleep(int64_t nsecs) {
+  emscripten_thread_sleep(nsecs / 1000000);
+}
+#endif
 
 // Tests emscripten_lock_init(), emscripten_lock_waitinf_acquire() and emscripten_lock_release()
 
@@ -13,13 +30,19 @@ emscripten_lock_t lock = (emscripten_lock_t)12345315; // initialize with garbage
 volatile int sharedState0 = 0;
 volatile int sharedState1 = 1;
 
-volatile int numWorkersAlive = 0;
+#define NUM_THREADS 4
+
+volatile int numWorkersAlive = NUM_THREADS;
 
 void test_ended() {
   emscripten_outf("Worker %d last thread to finish. Reporting test end with sharedState0=%d, sharedState1=%d", emscripten_wasm_worker_self_id(), sharedState0, sharedState1);
   assert(sharedState0 == sharedState1 + 1 || sharedState1 == sharedState0 + 1);
-#ifdef REPORT_RESULT
-  REPORT_RESULT(sharedState0);
+  assert(sharedState0 == 4000);
+  emscripten_out("done");
+#if __EMSCRIPTEN_PTHREADS__
+  exit(0);
+#else
+  REPORT_RESULT(0);
 #endif
 }
 
@@ -53,13 +76,32 @@ void worker_main() {
   }
 }
 
+#ifdef __EMSCRIPTEN_PTHREADS__
+pthread_t threads[NUM_THREADS];
+
+void* pthread_main(void* arg) {
+  worker_main();
+  return NULL;
+}
+#endif
+
 int main() {
   emscripten_lock_init(&lock);
 
-#define NUM_THREADS 4
-  numWorkersAlive = NUM_THREADS;
   for (int i = 0; i < NUM_THREADS; ++i) {
+#ifdef __EMSCRIPTEN_PTHREADS__
+    pthread_create(&threads[i], NULL, pthread_main, NULL);
+#else
     emscripten_wasm_worker_t worker = emscripten_malloc_wasm_worker(1024);
     emscripten_wasm_worker_post_function_v(worker, worker_main);
+#endif
   }
+
+#ifdef __EMSCRIPTEN_PTHREADS__
+  for (int i = 0; i < NUM_THREADS; ++i) {
+    pthread_join(threads[i], NULL);
+  }
+#endif
+
+  return 0;
 }

test/wasm_worker/semaphore_try_acquire.c

@@ -30,7 +30,6 @@ int main() {
   idx = emscripten_semaphore_try_acquire(&available, 9);
   assert(idx == 1);
 
-#ifdef REPORT_RESULT
-  REPORT_RESULT(0);
-#endif
+  emscripten_out("done");
+  return 0;
 }

test/wasm_worker/semaphore_waitinf_acquire.c

@@ -1,11 +1,18 @@
+#include <assert.h>
 #include <emscripten/console.h>
-#include <emscripten/wasm_worker.h>
 #include <emscripten/threading.h>
 #include <stdlib.h>
-#include <assert.h>
+
+#ifdef __EMSCRIPTEN_PTHREADS__
+#include <pthread.h>
+#else
+#include <emscripten/wasm_worker.h>
+#endif
 
 // Tests emscripten_semaphore_init(), emscripten_semaphore_waitinf_acquire() and emscripten_semaphore_release()
 
+#define NUM_THREADS 6
+
 emscripten_semaphore_t threadsWaiting = (emscripten_semaphore_t)12345315; // initialize with garbage
 emscripten_semaphore_t threadsRunning = EMSCRIPTEN_SEMAPHORE_T_STATIC_INITIALIZER(0); // initialize with static initializer
 emscripten_semaphore_t threadsCompleted = EMSCRIPTEN_SEMAPHORE_T_STATIC_INITIALIZER(0);
@@ -62,20 +69,49 @@ void control_thread() {
   assert(threadCounter == 6);
 
   emscripten_out("control_thread: test finished");
-#ifdef REPORT_RESULT
+#if defined(REPORT_RESULT) && !defined(__EMSCRIPTEN_PTHREADS__)
   REPORT_RESULT(0);
 #endif
 }
 
+#ifdef __EMSCRIPTEN_PTHREADS__
+void* control_pthread(void* arg) {
+  control_thread();
+  return NULL;
+}
+void* worker_pthread(void* arg) {
+  worker_main();
+  return NULL;
+}
+#endif
+
 int main() {
+  emscripten_out("in main");
   emscripten_semaphore_init(&threadsWaiting, 0);
 
+#ifdef __EMSCRIPTEN_PTHREADS__
+  pthread_t p;
+  pthread_t workers[NUM_THREADS];
+  int rtn = pthread_create(&p, NULL, control_pthread, NULL);
+  assert(rtn == 0);
+
+  for (int i = 0; i < NUM_THREADS; ++i) {
+    rtn = pthread_create(&workers[i], NULL, worker_pthread, NULL);
+    assert(rtn == 0);
+  }
+
+  pthread_join(p, NULL);
+  for (int i = 0; i < NUM_THREADS; ++i) {
+    pthread_join(workers[i], NULL);
+  }
+  emscripten_out("done");
+#else
   emscripten_wasm_worker_t worker = emscripten_malloc_wasm_worker(1024);
   emscripten_wasm_worker_post_function_v(worker, control_thread);
 
-#define NUM_THREADS 6
   for (int i = 0; i < NUM_THREADS; ++i) {
     emscripten_wasm_worker_t worker = emscripten_malloc_wasm_worker(1024);
     emscripten_wasm_worker_post_function_v(worker, worker_main);
   }
+#endif
 }