Align/unify wait in slices logic. NFC.

Author: kleisauke

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

@@ -222,12 +222,11 @@ static inline void __futexwait(volatile void *addr, int val, int priv)
 	if (is_runtime_thread) {
 		int e;
 		do {
-			// Main runtime thread waits in _very_ small slices so that it stays responsive to assist proxied
-			// pthread calls.
+			// 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);
 			// Assist other threads by executing proxied operations that are effectively singlethreaded.
 			emscripten_main_thread_process_queued_calls();
-		} while(e == -ETIMEDOUT);
+		} while (e == -ETIMEDOUT);
 	} else {
 		// Can wait in one go.
 		emscripten_futex_wait(addr, val, INFINITY);

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

@@ -58,11 +58,11 @@ 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;
-	int is_runtime_thread = emscripten_is_main_runtime_thread();
+	double msecsToSleep = top ? (top->tv_sec * 1000.0 + top->tv_nsec / 1000000.0) : INFINITY;
+	const 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 maxMsecsSliceToSleep = is_runtime_thread ? 1 : 100;
+	const double maxMsecsToSleep = is_runtime_thread ? 1 : 100;
 
 	// cp suffix in the function name means "cancellation point", so this wait can be cancelled
 	// by the users unless current threads cancellability is set to PTHREAD_CANCEL_DISABLE
@@ -76,18 +76,13 @@ int __timedwait_cp(volatile int *addr, int val,
 				// cancel execution.
 				return ECANCELED;
 			}
+			// Must wait in slices in case this thread is cancelled in between.
+			r = -emscripten_futex_wait((void*)addr, val, msecsToSleep > maxMsecsToSleep ? maxMsecsToSleep : msecsToSleep);
 			// Assist other threads by executing proxied operations that are effectively singlethreaded.
 			if (is_runtime_thread) emscripten_main_thread_process_queued_calls();
-			// Must wait in slices in case this thread is cancelled in between.
-			double waitMsecs = sleepUntilTime - emscripten_get_now();
-			if (waitMsecs <= 0) {
-				r = ETIMEDOUT;
-				break;
-			}
-			if (waitMsecs > maxMsecsSliceToSleep)
-				waitMsecs = maxMsecsSliceToSleep;
-			r = -emscripten_futex_wait((void*)addr, val, waitMsecs);
-		} while(r == ETIMEDOUT);
+
+			msecsToSleep = sleepUntilTime - emscripten_get_now();
+		} while (r == ETIMEDOUT && msecsToSleep > 0);
 	} else {
 		// Can wait in one go.
 		r = -emscripten_futex_wait((void*)addr, val, msecsToSleep);

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

@@ -13,24 +13,24 @@ void __wait(volatile int *addr, volatile int *waiters, int val, int priv)
 	if (waiters) a_inc(waiters);
 #ifdef __EMSCRIPTEN__
 	pthread_t self = __pthread_self();
-	int is_runtime_thread = emscripten_is_main_runtime_thread();
+	const 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 maxMsecsSliceToSleep = is_runtime_thread ? 1 : 100;
+	const double maxMsecsToSleep = is_runtime_thread ? 1 : 100;
 
 	while (*addr==val) {
 		if (is_runtime_thread || self->cancelasync) {
-			// Must wait in slices in case this thread is cancelled in between.
 			int e;
 			do {
 				if (self->cancel) {
 					if (waiters) a_dec(waiters);
 					return;
 				}
+				// Must wait in slices in case this thread is cancelled in between.
+				e = emscripten_futex_wait((void*)addr, val, maxMsecsToSleep);
 				// Assist other threads by executing proxied operations that are effectively singlethreaded.
 				if (is_runtime_thread) emscripten_main_thread_process_queued_calls();
-				e = emscripten_futex_wait((void*)addr, val, maxMsecsSliceToSleep);
-			} while(e == -ETIMEDOUT);
+			} while (e == -ETIMEDOUT);
 		} else {
 			// Can wait in one go.
 			emscripten_futex_wait((void*)addr, val, INFINITY);

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);
+	emscripten_futex_wake(l, INT_MAX);
 #else
 	if (w) __wake(l, 1, 1);
 	else __syscall(SYS_futex, l, FUTEX_REQUEUE|FUTEX_PRIVATE, 0, 1, r) != -ENOSYS

system/lib/pthread/library_pthread.c

@@ -69,33 +69,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;
 
-  // runtime thread may need to run proxied calls, so sleep in very small slices to be responsive.
+  // Runtime thread may need to run proxied calls, so sleep in very small slices to be responsive.
   const double maxMsecsSliceToSleep = emscripten_is_main_runtime_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;
+
+    emscripten_futex_wait(&dummyZeroAddress, 0,
+      msecsToSleep > maxMsecsSliceToSleep ? maxMsecsSliceToSleep : msecsToSleep);
+  } while (msecsToSleep > 0);
 
   emscripten_conditional_set_current_thread_status(
     EM_THREAD_STATUS_SLEEPING, EM_THREAD_STATUS_RUNNING);
@@ -369,25 +365,24 @@ static CallQueue* GetOrAllocateQueue(void* target) {
   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 r;
-
   int done = atomic_load(&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 = atomic_load(&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;
+  if (done) return EMSCRIPTEN_RESULT_SUCCESS;
+
+  emscripten_set_current_thread_status(EM_THREAD_STATUS_WAITPROXY);
+
+  double timeoutUntilTime = emscripten_get_now() + timeoutMSecs;
+  do {
+    emscripten_futex_wait(&call->operationDone, 0, timeoutMSecs);
+    done = atomic_load(&call->operationDone);
+
+    timeoutMSecs = timeoutUntilTime - emscripten_get_now();
+  } while (!done && timeoutMSecs > 0);
+
+  emscripten_set_current_thread_status(EM_THREAD_STATUS_RUNNING);
+
+  return done ? EMSCRIPTEN_RESULT_SUCCESS : EMSCRIPTEN_RESULT_TIMED_OUT;
 }
 
 EMSCRIPTEN_RESULT emscripten_wait_for_call_i(
@@ -444,7 +439,7 @@ int _emscripten_do_dispatch_to_thread(pthread_t target_thread, em_queued_call* c
     // 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);
+      emscripten_futex_wait((void *)&q->call_queue_head, head, INFINITY);
       pthread_mutex_lock(&call_queue_lock);
       head = q->call_queue_head;
       tail = q->call_queue_tail;
@@ -624,7 +619,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);
+  emscripten_futex_wake((void *)&q->call_queue_head, INT_MAX);
 
   thread_is_processing_queued_calls = false;
 }