Fix pthread wait issues with cancellation which caused semaphores not being waitable in pthread cancellation handlers.

Author: juj

src/library_pthread.js

@@ -51,25 +51,32 @@ var LibraryPThread = {
     // Called when we are performing a pthread_exit(), either explicitly called by programmer,
     // or implicitly when leaving the thread main function.
     threadExit: function(exitCode) {
-      PThread.runExitHandlers();
-      // No-op in the main thread. Note: Spec says we should join() all child threads, but since we don't have join,
-      // we might at least cancel all threads.
-      if (!ENVIRONMENT_IS_PTHREAD) return 0;
-
-      if (threadBlock) { // If we haven't yet exited?
-        Atomics.store(HEAPU32, (threadBlock + {{{ C_STRUCTS.pthread.threadExitCode }}} ) >> 2, exitCode);
+      var tb = _pthread_self();
+      if (tb) { // If we haven't yet exited?
+        Atomics.store(HEAPU32, (tb + {{{ C_STRUCTS.pthread.threadExitCode }}} ) >> 2, exitCode);
         // When we publish this, the main thread is free to deallocate the thread object and we are done.
         // Therefore set threadBlock = 0; above to 'release' the object in this worker thread.
-        Atomics.store(HEAPU32, (threadBlock + {{{ C_STRUCTS.pthread.threadStatus }}} ) >> 2, 1);
-        _emscripten_futex_wake(threadBlock + {{{ C_STRUCTS.pthread.threadStatus }}}, {{{ cDefine('INT_MAX') }}});
+        Atomics.store(HEAPU32, (tb + {{{ C_STRUCTS.pthread.threadStatus }}} ) >> 2, 1);
+
+        // Disable all cancellation so that executing the cleanup handlers won't trigger another JS
+        // canceled exception to be thrown.
+        Atomics.store(HEAPU32, (tb + {{{ C_STRUCTS.pthread.canceldisable }}} ) >> 2, 1);
+
+        PThread.runExitHandlers();
+
+        _emscripten_futex_wake(tb + {{{ C_STRUCTS.pthread.threadStatus }}}, {{{ cDefine('INT_MAX') }}});
         threadBlock = 0;
-        postMessage({ cmd: 'exit' });
+        if (ENVIRONMENT_IS_PTHREAD) {
+          postMessage({ cmd: 'exit' });
+        }
       }
     },
 
     threadCancel: function() {
       PThread.runExitHandlers();
       Atomics.store(HEAPU32, (threadBlock + {{{ C_STRUCTS.pthread.threadExitCode }}} ) >> 2, -1/*PTHREAD_CANCELED*/);
+      Atomics.store(HEAPU32, (threadBlock + {{{ C_STRUCTS.pthread.threadStatus }}} ) >> 2, 1); // Mark the thread as no longer running.
+      _emscripten_futex_wake(threadBlock + {{{ C_STRUCTS.pthread.threadStatus }}}, {{{ cDefine('INT_MAX') }}}); // wake all threads
       threadBlock = selfThreadId = 0; // Not hosting a pthread anymore in this worker, reset the info structures to null.
       postMessage({ cmd: 'cancelDone' });
     },
@@ -573,7 +580,9 @@ var LibraryPThread = {
   // Returns 0 on success, or one of the values -ETIMEDOUT, -EWOULDBLOCK or -EINVAL on error.
   emscripten_futex_wait: function(addr, val, timeout) {
     if (addr <= 0 || addr > HEAP8.length || addr&3 != 0) return -{{{ cDefine('EINVAL') }}};
+//    dump('futex_wait addr:' + addr + ' by thread: ' + _pthread_self() + (ENVIRONMENT_IS_PTHREAD?'(pthread)':'') + '\n');
     var ret = Atomics.futexWait(HEAP32, addr >> 2, val, timeout);
+//    dump('futex_wait done\n');
     if (ret == Atomics.TIMEDOUT) return -{{{ cDefine('ETIMEDOUT') }}};
     if (ret == Atomics.NOTEQUAL) return -{{{ cDefine('EWOULDBLOCK') }}};
     if (ret == 0) return 0;
@@ -584,6 +593,7 @@ var LibraryPThread = {
   // Pass count == INT_MAX to wake up all threads.
   emscripten_futex_wake: function(addr, count) {
     if (addr <= 0 || addr > HEAP8.length || addr&3 != 0 || count < 0) return -{{{ cDefine('EINVAL') }}};
+//    dump('futex_wake addr:' + addr + ' by thread: ' + _pthread_self() + (ENVIRONMENT_IS_PTHREAD?'(pthread)':'') + '\n');
     var ret = Atomics.futexWake(HEAP32, addr >> 2, count);
     if (ret >= 0) return ret;
     throw 'Atomics.futexWake returned an unexpected value ' + ret;

src/pthread-main.js

@@ -73,13 +73,13 @@ this.onmessage = function(e) {
       else
         result = asm.dynCall_i(e.data.start_routine); // as a hack, try signature 'i' as fallback.
     } catch(e) {
-      Atomics.store(HEAPU32, (threadBlock + 0 /*{{{ C_STRUCTS.pthread.threadStatus }}}*/ ) >> 2, 1); // Mark the thread as no longer running.
-      _emscripten_futex_wake(threadBlock + 0 /*{{{ C_STRUCTS.pthread.threadStatus }}}*/, 9999); // wake all threads
       if (e === 'Canceled!') {
         PThread.threadCancel();
         return;
       } else {
         Atomics.store(HEAPU32, (threadBlock + 4 /*{{{ C_STRUCTS.pthread.threadExitCode }}}*/ ) >> 2, -2 /*A custom entry specific to Emscripten denoting that the thread crashed.*/);
+        Atomics.store(HEAPU32, (threadBlock + 0 /*{{{ C_STRUCTS.pthread.threadStatus }}}*/ ) >> 2, 1); // Mark the thread as no longer running.
+        _emscripten_futex_wake(threadBlock + 0 /*{{{ C_STRUCTS.pthread.threadStatus }}}*/, 0x7FFFFFFF/*INT_MAX*/); // wake all threads
         throw e;
       }
     }

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

@@ -35,9 +35,14 @@ static int do_wait(volatile int *addr, int val,
 	}
 
 #ifdef __EMSCRIPTEN__
-	if (pthread_self()->cancelasync == PTHREAD_CANCEL_ASYNCHRONOUS) {
+	if (1 || pthread_self()->cancelasync == PTHREAD_CANCEL_ASYNCHRONOUS) {
 		do {
-			if (_pthread_isduecanceled(pthread_self())) return EINTR;
+			if (_pthread_isduecanceled(pthread_self())) {
+				// Emscripten-specific return value: The wait was canceled by user calling
+				// pthread_cancel() for this thread, and the caller needs to cooperatively
+				// cancel execution.
+				return ECANCELED;
+			}
 			// Must wait in slices in case this thread is cancelled in between.
 			double waitNsecs = at ? _pthread_nsecs_until(at) : INFINITY;
 			if (waitNsecs <= 0) {
@@ -73,9 +78,5 @@ int __timedwait(volatile int *addr, int val,
 	pthread_cleanup_pop(0);
 	if (!cleanup) pthread_setcancelstate(cs, 0);
 
-#ifdef __EMSCRIPTEN__
-	// XXX Emscripten: since we don't have signals, cooperatively test cancellation.
-	if (pthread_self()->cancelasync == PTHREAD_CANCEL_ASYNCHRONOUS) pthread_testcancel();
-#endif
 	return r;
 }

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

@@ -8,6 +8,13 @@ int pthread_cond_broadcast(pthread_cond_t *c)
 
 	a_inc(&c->_c_seq);
 
+#ifdef __EMSCRIPTEN__
+	// XXX Emscripten: TODO: This is suboptimal but works naively correctly for now. The Emscripten-specific code path below
+	// has a bug and does not work for some reason. Figure it out and remove this code block.
+	__wake(&c->_c_seq, -1, 0);
+	return 0;
+#endif
+
 	/* If cond var is process-shared, simply wake all waiters. */
 	if (c->_c_mutex == (void *)-1) {
 		__wake(&c->_c_seq, -1, 0);
@@ -28,6 +35,7 @@ int pthread_cond_broadcast(pthread_cond_t *c)
 #ifdef __EMSCRIPTEN__
 	int futexResult;
 	do {
+		// XXX Emscripten: Bug, this does not work correctly.
 		futexResult = emscripten_futex_wake_or_requeue(&c->_c_seq, !m->_m_type || (m->_m_lock&INT_MAX)!=pthread_self()->tid,
 			c->_c_seq, &m->_m_lock);
 	} while(futexResult == -EAGAIN);

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

@@ -64,13 +64,19 @@ int pthread_cond_timedwait(pthread_cond_t *restrict c, pthread_mutex_t *restrict
 
 	pthread_mutex_unlock(m);
 
-	do e = __timedwait(&c->_c_seq, seq, c->_c_clock, ts, cleanup, &cm, 0);
+	do {
+		e = __timedwait(&c->_c_seq, seq, c->_c_clock, ts, cleanup, &cm, 0);
+	}
 	while (c->_c_seq == seq && (!e || e==EINTR));
 	if (e == EINTR) e = 0;
 
 	unwait(c, m);
 
 	if ((r=pthread_mutex_lock(m))) return r;
 
+#ifdef __EMSCRIPTEN__
+	pthread_testcancel();
+#endif
+
 	return e;
 }

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

@@ -13,6 +13,9 @@ int sem_timedwait(sem_t *restrict sem, const struct timespec *restrict at)
 		a_inc(sem->__val+1);
 		a_cas(sem->__val, 0, -1);
 		r = __timedwait(sem->__val, -1, CLOCK_REALTIME, at, cleanup, sem->__val+1, 0);
+#ifdef __EMSCRIPTEN__
+		if (r == ECANCELED) r = EINTR;
+#endif
 		a_dec(sem->__val+1);
 		if (r) {
 			errno = r;