Blocking pipe writes from pthreads wait for the reader

ChangeLog.md

@@ -21,7 +21,10 @@ See docs/process.md for more on how version tagging works.
 6.0.14 (in development)
 -----------------------
 - Pipes now have a 64 KiB capacity: non-blocking writes to a full pipe fail
-  with `EAGAIN`. For now, blocking writes remain unbounded. (#27831)
+  with `EAGAIN`, and (JS FS) blocking writes from a pthread wait for the reader,
+  as on Linux. A pthread (including `main` under `PROXY_TO_PTHREAD`) that writes
+  more than 64 KiB before anything reads the pipe now blocks where it previously
+  completed. On the main thread blocking writes remain unbounded. (#27831)
 - When compiling and linking in a single command, auxiliary compiler output
   files (such as `.dwo` files from `-gsplit-dwarf` or `.json` files from
   `-ftime-trace`) are now named `<output>-<input>.<ext>`, matching the behavior

src/jsifier.mjs

@@ -507,21 +507,31 @@ ${body};
   });
 }
 
+// A sync-proxied body taking `sync` becomes `<name>$sync`. The symbol keeps
+// the import's arity and passes true, and only the proxied table entry passes
+// false, so nothing the call re-enters on the main thread sees the pthread's.
+const callerBoundSync = new Set();
+const syncBodies = {};
+
 // An `__async: 'auto'` body returns a value or a Promise, as it can. A body
 // whose last parameter is named `sync` receives there whether it must complete
 // synchronously on this call, as a returned Promise could not be waited for.
 // That is never the case under ASYNCIFY/JSPI, nor on behalf of a sync-proxied
 // pthread caller, which awaits it; otherwise it is.
 function handleSyncArg(symbol, snippet, sig, proxyingMode) {
-  const sync = ASYNCIFY ? 'false' :
-    PTHREADS && proxyingMode == 'sync' ? '!PThread.currentProxiedOperationCallerThread' : 'true';
+  const fromCaller = !ASYNCIFY && PTHREADS && proxyingMode == 'sync';
+  const sync = ASYNCIFY ? 'false' : 'true';
   return modifyJSFunction(snippet, (args, body, async_, oneliner) => {
     const params = args.split(',').map((a) => a.trim()).filter((a) => a);
     if (params.at(-1) != 'sync') return snippet;
     if (sig && params.length != sig.length) {
       error(`JS library error: '${symbol}' must take 'sync' as parameter ${sig.length}, after the ${sig.length - 1} of its '__sig'`);
       return snippet;
     }
+    if (fromCaller) {
+      callerBoundSync.add(symbol);
+      return snippet;
+    }
     if (!oneliner) body = `{\n${body}\n}`;
     const outer = params.slice(0, -1).join(', ');
     return `\
@@ -531,25 +541,27 @@ function(${outer}) {
   });
 }
 
-function handleAsyncFunction(snippet, sig, proxied) {
+function handleAsyncFunction(symbol, snippet, sig, proxied) {
   const return64 = sig && (MEMORY64 && sig.startsWith('p') || sig.startsWith('j'))
   let handleAsync = 'Asyncify.handleAsync(innerFunc)'
   if (return64 && ASYNCIFY == 1) {
     handleAsync = makeReturn64(handleAsync);
   }
-  // When dispatching on behalf of a proxied caller (PROXY_SYNC_ASYNC), the
-  // caller awaits the returned promise, so return it directly rather than
-  // suspending the main thread.
+  // A proxied caller awaits the result itself, and a Promise wrapper would
+  // defer a synchronous completion to a microtask, so return it directly.
   let proxiedDispatch = '';
-  if (ASYNCIFY == 1 && PTHREADS && proxied) {
-    proxiedDispatch = 'if (PThread.currentProxiedOperationCallerThread) return innerFunc();\n  ';
+  let syncArg = '';
+  if (PTHREADS && proxied) {
+    callerBoundSync.add(symbol);
+    proxiedDispatch = 'if (!sync) return innerFunc();\n  ';
+    syncArg = 'sync';
   }
   return modifyJSFunction(snippet, (args, body, async_, oneliner) => {
     if (!oneliner) {
       body = `{\n${body}\n}`;
     }
     return `\
-function(${args}) {
+function(${args}${args && syncArg ? ', ' : ''}${syncArg}) {
   let innerFunc = ${async_} () => ${body};
   ${proxiedDispatch}return ${handleAsync};
 }\n`;
@@ -676,19 +688,29 @@ function(${args}) {
     }
 
     if (ASYNCIFY && isAsyncFunction == 'auto') {
-      snippet = handleAsyncFunction(snippet, sig, proxyingMode == 'sync');
+      snippet = handleAsyncFunction(symbol, snippet, sig, proxyingMode == 'sync');
     }
 
     if (proxyingMode) {
       if (!['sync', 'async', 'none'].includes(proxyingMode)) {
         error(`JS library error: invalid proxying mode '${symbol}__proxy: ${proxyingMode}' specified`);
       }
       if (SHARED_MEMORY && proxyingMode != 'none') {
+        let tableEntry = mangled;
         if (PTHREADS) {
           snippet = modifyJSFunction(snippet, (args, body, async_, oneliner) => {
             if (oneliner) {
               body = `return ${body}`;
             }
+            let callArgs = args;
+            if (callerBoundSync.has(symbol)) {
+              const params = args.split(',').map((a) => a.trim()).filter((a) => a);
+              callArgs = params.slice(0, -1).join(', ');
+              const sep = callArgs ? ', ' : '';
+              tableEntry = `(${callArgs}) => ${mangled}$sync(${callArgs}${sep}false)`;
+              syncBodies[symbol] = `${async_}function ${mangled}$sync(${args}) {\n${body}\n}\n`;
+              body = `return ${mangled}$sync(${callArgs}${sep}true);`;
+            }
             let proxyMode = PROXY_ASYNC;
             if (proxyingMode === 'sync') {
               const isAsyncFunction = LibraryManager.library[symbol + '__async'];
@@ -702,9 +724,9 @@ function(${args}) {
             const proxyFunc =
               MEMORY64 && rtnType == 'p' ? 'proxyToMainThreadPtr' : 'proxyToMainThread';
             return `
-${async_}function(${args}) {
+${async_}function(${callArgs}) {
 if (ENVIRONMENT_IS_PTHREAD)
-  return ${proxyFunc}(${proxiedFunctionTable.length}, 0, ${proxyMode}${args ? ', ' : ''}${args});
+  return ${proxyFunc}(${proxiedFunctionTable.length}, 0, ${proxyMode}${callArgs ? ', ' : ''}${callArgs});
 ${body}
 }\n`;
           });
@@ -720,7 +742,7 @@ function(${args}) {
 }\n`,
           );
         }
-        proxiedFunctionTable.push(mangled);
+        proxiedFunctionTable.push(tableEntry);
       }
     }
 
@@ -951,6 +973,7 @@ function(${args}) {
         snippet = processLibraryFunction(snippet, symbol, mangled, deps, isStub);
         if (!isStub) {
           addImplicitDeps(snippet);
+          if (symbol in syncBodies) addImplicitDeps(syncBodies[symbol]);
         }
       }
 
@@ -1002,6 +1025,9 @@ function(${args}) {
           // Handle regular (non-arrow) functions
           contentText = contentText.replace(/function(?:\s+([^(]+))?\s*\(/, `function ${mangled}(`);
         }
+        if (symbol in syncBodies) {
+          contentText = syncBodies[symbol] + contentText;
+        }
       } else if (typeof snippet === 'string' && snippet.startsWith(';')) {
         // In JS libraries
         //   foo: ';[code here verbatim]'

src/lib/libepoll.js

@@ -302,33 +302,30 @@ var EpollLibrary = {
   $epollPwait__internal: true,
   $epollPwait: (ep, ev, maxevents, timeout, sync) => {
 #if PTHREADS || ASYNCIFY
-    // Always resolve through a Promise here: when proxied from a worker the
-    // result is delivered by promise resolution, so a bare value would break
-    // the proxy (it has no `.then`). Block on the epoll's own readiness - each
-    // registration's persistent listener wakes ep.node on a leaf edge - and
-    // re-derive on wake, resolving the count or 0 after `timeout`.
+    // Re-derives readiness on each wake of ep.node and settles a SyncDeferred
+    // with the count, or with 0 once `timeout` passes.
     if (!sync) {
-      return new Promise((resolve) => {
-        var count = doEpollWait(ep, ev, maxevents);
-        if (count || !timeout) {
-          resolve(count);
-          return;
-        }
-        var done = false;
-        var reg = ep.node.addListener(() => {
-          if (done) return;
-          var c = doEpollWait(ep, ev, maxevents);
-          if (c) finish(c);
-        });
-        var timer = timeout > 0 ? setTimeout(() => finish(0), timeout) : undefined;
-        function finish(c) {
-          if (done) return;
-          done = true;
-          reg.listeners.delete(reg.entry);
-          if (timer) clearTimeout(timer);
-          resolve(c);
-        }
+      var deferred = new SyncDeferred();
+      var n = doEpollWait(ep, ev, maxevents);
+      if (n || !timeout) {
+        deferred.resolve(n);
+        return deferred;
+      }
+      var done = false;
+      var finish = (c) => {
+        if (done) return;
+        done = true;
+        reg.listeners.delete(reg.entry);
+        if (timer) clearTimeout(timer);
+        deferred.resolve(c);
+      };
+      var reg = ep.node.addListener(() => {
+        if (done) return;
+        var c = doEpollWait(ep, ev, maxevents);
+        if (c) finish(c);
       });
+      var timer = timeout > 0 ? setTimeout(() => finish(0), timeout) : undefined;
+      return deferred;
     }
 #endif
     var count = doEpollWait(ep, ev, maxevents);

src/lib/libpipefs.js

@@ -229,9 +229,9 @@ addToLibrary({
           throw new FS.ErrnoError({{{ cDefs.EPIPE }}});
         }
 
-        // PIPE_BUF rule: a small write is all-or-nothing, a larger one takes
-        // what fits.
-        if (stream.flags & {{{ cDefs.O_NONBLOCK }}}) {
+        // A blocking writer is capped only when fd_write marks it able to
+        // wait. A write up to PIPE_BUF is all-or-nothing, a larger one partial.
+        if ({{{ PTHREADS && !ASYNCIFY ? 'stream.canWait || ' : '' }}}(stream.flags & {{{ cDefs.O_NONBLOCK }}})) {
           var room = PIPEFS.CAPACITY - PIPEFS.queuedBytes(pipe);
           if (room <= 0 || (dataLen <= {{{ cDefs.PIPE_BUF }}} && dataLen > room)) {
             throw new FS.ErrnoError({{{ cDefs.EAGAIN }}});

src/lib/libpthread.js

@@ -1079,8 +1079,13 @@ var LibraryPThread = {
     var coerceRtn = (rtn) => rtn;
 #endif
     if (ctx) {
-      // A PROXY_SYNC_ASYNC function may complete synchronously with a plain value.
-      Promise.resolve(rtn).then((rtn) => __emscripten_run_js_on_main_thread_done(ctx, ctxArgs, coerceRtn(rtn)));
+      // The main thread may itself be blocked on the caller and never reach a
+      // microtask, so finish a value at once and a thenable when it settles.
+      if (rtn?.then) {
+        rtn.then((rtn) => __emscripten_run_js_on_main_thread_done(ctx, ctxArgs, coerceRtn(rtn)));
+      } else {
+        __emscripten_run_js_on_main_thread_done(ctx, ctxArgs, coerceRtn(rtn));
+      }
       return;
     }
     rtn = coerceRtn(rtn);

src/lib/libsyscall.js

@@ -628,9 +628,10 @@ var SyscallsLibrary = {
   // ready and `timeout` is non-zero, register one waiter per fd on its node
   // wait-queue and re-derive the whole set on any wake (the wake flags are just
   // the trigger), resolving then or, for a positive `timeout`, once it elapses.
-  // A negative `timeout` waits forever. Returns a Promise of the ready count.
+  // A negative `timeout` waits forever. Settles a SyncDeferred with the count.
   $doPollAsync__internal: true,
-  $doPollAsync: (fds, nfds, timeout) => new Promise((resolve) => {
+  $doPollAsync: (fds, nfds, timeout) => {
+    var deferred = new SyncDeferred();
     var regs = [];
     var timer;
     var done = false;
@@ -650,7 +651,7 @@ var SyscallsLibrary = {
       done = true;
       for (var r of regs) r.listeners.delete(r.entry);
       if (timer) clearTimeout(timer);
-      resolve(count);
+      deferred.resolve(count);
     }
     var count = derive();
     if (count || !timeout) {
@@ -667,7 +668,45 @@ var SyscallsLibrary = {
       }
       if (timeout > 0) timer = setTimeout(() => finish(0), timeout);
     }
-  }),
+    return deferred;
+  },
+  // Calls `cb` synchronously from the notify that makes `events` ready.
+  $waitStreamEvents__internal: true,
+  $waitStreamEvents: (stream, events, cb) => {
+    var reg = stream.node.addListener(() => {
+      if (!pollOne(stream.fd, events)) return;
+      reg.listeners.delete(reg.entry);
+      cb();
+    });
+  },
+
+  // A Promise-shaped result that settles synchronously. The main thread may be
+  // blocked servicing proxied calls without turning its event loop, where a
+  // Promise's microtask would never reach the waiting pthread.
+  $SyncDeferred__internal: true,
+  $SyncDeferred: class SyncDeferred {
+    constructor() {
+      this.settle = null;
+      this.settled = null;
+    }
+    then(onFulfilled, onRejected) {
+      var next = new SyncDeferred();
+      var settle = (fulfilled, v) => {
+        try {
+          if (fulfilled) next.resolve(onFulfilled ? onFulfilled(v) : v);
+          else if (onRejected) next.resolve(onRejected(v));
+          else next.reject(v);
+        } catch (e) {
+          next.reject(e);
+        }
+      };
+      if (this.settled) settle(...this.settled);
+      else this.settle = settle;
+      return next;
+    }
+    resolve(v) { this.settle ? this.settle(true, v) : this.settled = [true, v]; }
+    reject(e) { this.settle ? this.settle(false, e) : this.settled = [false, e]; }
+  },
 #endif
   // libc routes zero-timeout poll() calls here: the same synchronous
   // readiness derivation as __syscall_poll, but as a plain import that never