Experiment with atomic builtins

Seems to work!

Author: kleisauke

system/lib/libc/musl/src/internal/pthread_impl.h

@@ -183,7 +183,10 @@ hidden int __timedwait_cp(volatile int *, int, clockid_t, const struct timespec
 hidden void __wait(volatile int *, volatile int *, int, int);
 static inline void __wake(volatile void *addr, int cnt, int priv)
 {
-#ifdef __EMSCRIPTEN__
+#ifdef __EMSCRIPTEN_PTHREADS__
+	(void)priv;
+	__builtin_wasm_memory_atomic_notify((int *)addr, cnt < 0 ? -1 : cnt);
+#elif defined(__EMSCRIPTEN__)
 	(void)priv;
 	emscripten_futex_wake(addr, cnt < 0 ? INT_MAX : cnt);
 #else
@@ -196,21 +199,24 @@ static inline void __wake(volatile void *addr, int cnt, int priv)
 
 static inline void __futexwait(volatile void *addr, int val, int priv)
 {
-#ifdef __EMSCRIPTEN__
+#ifdef __EMSCRIPTEN_PTHREADS__
 	(void)priv;
-	const int is_runtime_thread = emscripten_is_main_runtime_thread();
+	int is_runtime_thread = emscripten_is_main_runtime_thread();
 	if (is_runtime_thread) {
 		int e;
 		do {
 			// Main runtime thread may need to run proxied calls, so sleep in very small slices to be responsive.
-			e = emscripten_futex_wait(addr, val, 1);
+			e = __builtin_wasm_memory_atomic_wait32((int *)addr, val, 1000000);
 			// Assist other threads by executing proxied operations that are effectively singlethreaded.
 			emscripten_main_thread_process_queued_calls();
-		} while(e == -ETIMEDOUT);
+		} while(e == 2/* ETIMEDOUT*/);
 	} else {
 		// Can wait in one go.
-		emscripten_futex_wait(addr, val, INFINITY);
+		__builtin_wasm_memory_atomic_wait32((int *)addr, val, -1);
 	}
+#elif defined(__EMSCRIPTEN__)
+	(void)priv;
+	emscripten_futex_wait(addr, val, INFINITY);
 #else
 	if (priv) priv = FUTEX_PRIVATE;
 	__syscall(SYS_futex, addr, FUTEX_WAIT|priv, val, 0) != -ENOSYS ||

system/lib/libc/musl/src/thread/__timedwait.c

@@ -56,8 +56,8 @@ int __timedwait_cp(volatile int *addr, int val,
 	}
 #ifdef __EMSCRIPTEN__
 	pthread_t self = __pthread_self();
-	double msecsToSleep = top ? (top->tv_sec * 1000 + top->tv_nsec / 1000000.0) : INFINITY;
-	const int is_runtime_thread = emscripten_is_main_runtime_thread();
+	double msecsToSleep = top ? (top->tv_sec * 1000.0 + top->tv_nsec / 1000000.0) : INFINITY;
+	int is_runtime_thread = emscripten_is_main_runtime_thread();
 
 	// Main runtime thread may need to run proxied calls, so sleep in very small slices to be responsive.
 	const double maxMsecsToSleep = is_runtime_thread ? 1 : 100;
@@ -75,14 +75,20 @@ int __timedwait_cp(volatile int *addr, int val,
 				return ECANCELED;
 			}
 			// Must wait in slices in case this thread is cancelled in between.
-			r = -emscripten_futex_wait(addr, val, msecsToSleep > maxMsecsToSleep ? maxMsecsToSleep : msecsToSleep);
+			r = __builtin_wasm_memory_atomic_wait32((int *)addr, val,
+				(msecsToSleep > maxMsecsToSleep ? maxMsecsToSleep : msecsToSleep) * /*NSEC_PER_MSEC*/1000000);
+			if (r == 2) r = ETIMEDOUT;
+			if (r == 1) r = EWOULDBLOCK;
 			// Assist other threads by executing proxied operations that are effectively singlethreaded.
 			if (is_runtime_thread) emscripten_main_thread_process_queued_calls();
+
 			msecsToSleep = sleepUntilTime - emscripten_get_now();
-		} while(r == ETIMEDOUT && msecsToSleep > 0);
+		} while (r == ETIMEDOUT && msecsToSleep > 0);
 	} else {
 		// Can wait in one go.
-		r = -emscripten_futex_wait(addr, val, msecsToSleep);
+		r = __builtin_wasm_memory_atomic_wait32((int *)addr, val, top ? (msecsToSleep * /*NSEC_PER_MSEC*/1000000) : -1);
+		if (r == 2) r = ETIMEDOUT;
+		if (r == 1) r = EWOULDBLOCK;
 	}
 #else
 	r = -__futex4_cp(addr, FUTEX_WAIT|priv, val, top);

system/lib/libc/musl/src/thread/__wait.c

@@ -13,10 +13,10 @@ void __wait(volatile int *addr, volatile int *waiters, int val, int priv)
 	if (waiters) a_inc(waiters);
 #ifdef __EMSCRIPTEN__
 	pthread_t self = __pthread_self();
-	const int is_runtime_thread = emscripten_is_main_runtime_thread();
+	int is_runtime_thread = emscripten_is_main_runtime_thread();
 
 	// Main runtime thread may need to run proxied calls, so sleep in very small slices to be responsive.
-	const double maxMsecsToSleep = is_runtime_thread ? 1 : 100;
+	const int64_t maxNsecsToSleep = is_runtime_thread ? 1000000 : 100000000;
 
 	while (*addr==val) {
 		if (self->cancelasync) {
@@ -27,13 +27,13 @@ void __wait(volatile int *addr, volatile int *waiters, int val, int priv)
 					return;
 				}
 				// Must wait in slices in case this thread is cancelled in between.
-				e = emscripten_futex_wait(addr, val, maxMsecsToSleep);
+				e = __builtin_wasm_memory_atomic_wait32((int *)addr, val, maxNsecsToSleep);
 				// Assist other threads by executing proxied operations that are effectively singlethreaded.
 				if (is_runtime_thread) emscripten_main_thread_process_queued_calls();
-			} while(e == -ETIMEDOUT);
+			} while(e == /*ETIMEDOUT*/2);
 		} else {
 			// Can wait in one go.
-			emscripten_futex_wait(addr, val, INFINITY);
+			__builtin_wasm_memory_atomic_wait32((int *)addr, val, -1);
 		}
 	}
 #else

system/lib/libc/musl/src/thread/pthread_cond_timedwait.c

@@ -55,7 +55,7 @@ static inline void unlock_requeue(volatile int *l, volatile int *r, int w)
 	// primitive is strictly not needed, since it is more like an optimization to avoid spuriously waking
 	// all waiters, just to make them wait on another location immediately afterwards. Here we do exactly
 	// that: wake every waiter.
-	emscripten_futex_wake(l, 0x7FFFFFFF);
+	__builtin_wasm_memory_atomic_notify((int *)l, -1);
 #else
 	if (w) __wake(l, 1, 1);
 	else __syscall(SYS_futex, l, FUTEX_REQUEUE|FUTEX_PRIVATE, 0, 1, r) != -ENOSYS

system/lib/libc/musl/src/time/nanosleep.c

@@ -12,7 +12,7 @@ int nanosleep(const struct timespec *req, struct timespec *rem)
 		errno = EINVAL;
 		return -1;
 	}
-	emscripten_thread_sleep(req->tv_sec * 1000.0 + req->tv_nsec / 1e6);
+	emscripten_thread_sleep(req->tv_sec * 1000.0 + req->tv_nsec / 1000000.0);
 	return 0;
 #else
 	return __syscall_ret(-__clock_nanosleep(CLOCK_REALTIME, 0, req, rem));

system/lib/libc/musl/src/unistd/usleep.c

@@ -1,14 +1,14 @@
 #define _GNU_SOURCE
 #include <unistd.h>
 #include <time.h>
-#ifdef __EMSCRTIPEN__
+#ifdef __EMSCRIPTEN__
 #include <emscripten/threading.h>
 #endif
 
 int usleep(unsigned useconds)
 {
-#ifdef __EMSCRTIPEN__
-	emscripten_thread_sleep(usec / 1e3);
+#ifdef __EMSCRIPTEN__
+	emscripten_thread_sleep(useconds / 1000.0);
 	return 0;
 #else
 	struct timespec tv = {

system/lib/pthread/library_pthread.c

@@ -112,33 +112,29 @@ void emscripten_thread_sleep(double msecs) {
   double now = emscripten_get_now();
   double target = now + msecs;
 
-  __pthread_testcancel(); // pthreads spec: sleep is a cancellation point, so must test if this
-                          // thread is cancelled during the sleep.
-  emscripten_current_thread_process_queued_calls();
-
   // If we have less than this many msecs left to wait, busy spin that instead.
-  const double minimumTimeSliceToSleep = 0.1;
+  const double minTimeSliceToSleep = 0.1;
 
-  // main thread may need to run proxied calls, so sleep in very small slices to be responsive.
+  // Main browser thread may need to run proxied calls, so sleep in very small slices to be responsive.
   const double maxMsecsSliceToSleep = emscripten_is_main_browser_thread() ? 1 : 100;
 
   emscripten_conditional_set_current_thread_status(
     EM_THREAD_STATUS_RUNNING, EM_THREAD_STATUS_SLEEPING);
-  now = emscripten_get_now();
-  while (now < target) {
+
+  double msecsToSleep;
+  do {
     // Keep processing the main loop of the calling thread.
     __pthread_testcancel(); // pthreads spec: sleep is a cancellation point, so must test if this
                             // thread is cancelled during the sleep.
     emscripten_current_thread_process_queued_calls();
 
-    now = emscripten_get_now();
-    double msecsToSleep = target - now;
-    if (msecsToSleep > maxMsecsSliceToSleep)
-      msecsToSleep = maxMsecsSliceToSleep;
-    if (msecsToSleep >= minimumTimeSliceToSleep)
-      emscripten_futex_wait(&dummyZeroAddress, 0, msecsToSleep);
-    now = emscripten_get_now();
-  };
+    msecsToSleep = target - emscripten_get_now();
+    if (msecsToSleep < minTimeSliceToSleep)
+      continue;
+
+    __builtin_wasm_memory_atomic_wait32((int *)&dummyZeroAddress, 0,
+      (msecsToSleep > maxMsecsSliceToSleep ? maxMsecsSliceToSleep : msecsToSleep) * /*NSEC_PER_MSEC*/1000000);
+  } while (msecsToSleep > 0);
 
   emscripten_conditional_set_current_thread_status(
     EM_THREAD_STATUS_SLEEPING, EM_THREAD_STATUS_RUNNING);
@@ -336,7 +332,7 @@ static void _do_call(em_queued_call* q) {
   } else {
     // The caller owns this call object, it is listening to it and will free it up.
     q->operationDone = 1;
-    emscripten_futex_wake(&q->operationDone, INT_MAX);
+    __builtin_wasm_memory_atomic_notify((int *)&q->operationDone, -1);
   }
 }
 
@@ -390,25 +386,27 @@ static CallQueue* GetOrAllocateQueue(
   return q;
 }
 
+// TODO(kleisauke): All paths call this with timeoutMSecs == INFINITY, perhaps drop this param?
 EMSCRIPTEN_RESULT emscripten_wait_for_call_v(em_queued_call* call, double timeoutMSecs) {
+  int done = emscripten_atomic_load_u32(&call->operationDone);
+  if (done) return EMSCRIPTEN_RESULT_SUCCESS;
+
+  int waitIndefinitely = isinf(timeoutMSecs);
+
+  emscripten_set_current_thread_status(EM_THREAD_STATUS_WAITPROXY);
+
+  double timeoutUntilTime = emscripten_get_now() + timeoutMSecs;
   int r;
+  do {
+    r = __builtin_wasm_memory_atomic_wait32((int *)&call->operationDone, 0,
+      waitIndefinitely ? -1 : (timeoutMSecs * /*NSEC_PER_MSEC*/1000000));
 
-  int done = emscripten_atomic_load_u32(&call->operationDone);
-  if (!done) {
-    double now = emscripten_get_now();
-    double waitEndTime = now + timeoutMSecs;
-    emscripten_set_current_thread_status(EM_THREAD_STATUS_WAITPROXY);
-    while (!done && now < waitEndTime) {
-      r = emscripten_futex_wait(&call->operationDone, 0, waitEndTime - now);
-      done = emscripten_atomic_load_u32(&call->operationDone);
-      now = emscripten_get_now();
-    }
-    emscripten_set_current_thread_status(EM_THREAD_STATUS_RUNNING);
-  }
-  if (done)
-    return EMSCRIPTEN_RESULT_SUCCESS;
-  else
-    return EMSCRIPTEN_RESULT_TIMED_OUT;
+    timeoutMSecs = timeoutUntilTime - emscripten_get_now();
+  } while (r == /*ETIMEDOUT*/2 && timeoutMSecs > 0);
+
+  emscripten_set_current_thread_status(EM_THREAD_STATUS_RUNNING);
+
+  return r == 0 ? EMSCRIPTEN_RESULT_SUCCESS : EMSCRIPTEN_RESULT_TIMED_OUT;
 }
 
 EMSCRIPTEN_RESULT emscripten_wait_for_call_i(
@@ -470,7 +468,7 @@ int _emscripten_do_dispatch_to_thread(
     // If queue of the main browser thread is full, then we wait. (never drop messages for the main
     // browser thread)
     if (target_thread == emscripten_main_browser_thread_id()) {
-      emscripten_futex_wait((void*)&q->call_queue_head, head, INFINITY);
+      __builtin_wasm_memory_atomic_wait32((int *)&q->call_queue_head, head, -1);
       pthread_mutex_lock(&call_queue_lock);
       head = emscripten_atomic_load_u32((void*)&q->call_queue_head);
       tail = emscripten_atomic_load_u32((void*)&q->call_queue_tail);
@@ -680,7 +678,7 @@ void emscripten_current_thread_process_queued_calls() {
   pthread_mutex_unlock(&call_queue_lock);
 
   // If the queue was full and we had waiters pending to get to put data to queue, wake them up.
-  emscripten_futex_wake((void*)&q->call_queue_head, 0x7FFFFFFF);
+  __builtin_wasm_memory_atomic_notify((int *)&q->call_queue_head, -1);
 
   if (emscripten_is_main_browser_thread())
     bool_main_thread_inside_nested_process_queued_calls = 0;