diff --git a/ChangeLog.md b/ChangeLog.md index 584f498653de8..02b34b132f8ed 100644 --- a/ChangeLog.md +++ b/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 `-.`, matching the behavior diff --git a/src/jsifier.mjs b/src/jsifier.mjs index 6a05135ce6e5e..f8263bf858090 100644 --- a/src/jsifier.mjs +++ b/src/jsifier.mjs @@ -507,14 +507,20 @@ ${body}; }); } +// A sync-proxied body taking `sync` becomes `$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; @@ -522,6 +528,10 @@ function handleSyncArg(symbol, snippet, sig, proxyingMode) { 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,7 +688,7 @@ function(${args}) { } if (ASYNCIFY && isAsyncFunction == 'auto') { - snippet = handleAsyncFunction(snippet, sig, proxyingMode == 'sync'); + snippet = handleAsyncFunction(symbol, snippet, sig, proxyingMode == 'sync'); } if (proxyingMode) { @@ -684,11 +696,21 @@ function(${args}) { 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]' diff --git a/src/lib/libepoll.js b/src/lib/libepoll.js index 52cbcc97e74e2..85e2531491388 100644 --- a/src/lib/libepoll.js +++ b/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); diff --git a/src/lib/libpipefs.js b/src/lib/libpipefs.js index 0383adcc4f88f..766f5d5ad658e 100644 --- a/src/lib/libpipefs.js +++ b/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 }}}); diff --git a/src/lib/libpthread.js b/src/lib/libpthread.js index eabb60ac34f87..a407f12c9b78a 100644 --- a/src/lib/libpthread.js +++ b/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); diff --git a/src/lib/libsyscall.js b/src/lib/libsyscall.js index 1a0fab921c049..6256492f204e0 100644 --- a/src/lib/libsyscall.js +++ b/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 diff --git a/src/lib/libwasi.js b/src/lib/libwasi.js index 62ea2f7d7335c..976a9f6cd845d 100644 --- a/src/lib/libwasi.js +++ b/src/lib/libwasi.js @@ -237,6 +237,58 @@ var WasiLibrary = { } return FS.write(stream, view, 0, total, offset); }, +#if PTHREADS && !ASYNCIFY + // writev that waits on a blocking stream, giving a SyncDeferred if it must. + $doWritevBlocking: (stream, iov, iovcnt) => { + var buf, ptr, len; + if (iovcnt == 1) { + ptr = {{{ makeGetValue('iov', C_STRUCTS.iovec.iov_base, '*') }}}; + len = {{{ makeGetValue('iov', C_STRUCTS.iovec.iov_len, '*') }}}; + } else { + len = 0; + for (var i = 0, p = iov; i < iovcnt; i++, p += {{{ C_STRUCTS.iovec.__size__ }}}) { + len += {{{ makeGetValue('p', C_STRUCTS.iovec.iov_len, '*') }}}; + } + buf = new Uint8Array(len); + ptr = 0; + for (var i = 0; i < iovcnt; i++, iov += {{{ C_STRUCTS.iovec.__size__ }}}) { + var base = {{{ makeGetValue('iov', C_STRUCTS.iovec.iov_base, '*') }}}; + var n = {{{ makeGetValue('iov', C_STRUCTS.iovec.iov_len, '*') }}}; + buf.set(HEAPU8.subarray(base, base + n), ptr); + ptr += n; + } + ptr = 0; + } + var total = 0, deferred; + function step() { + while (len) { + var curr; + stream.canWait = true; + try { + curr = FS.write(stream, buf ?? HEAP8, ptr, len); + } catch (e) { + stream.canWait = false; + if (e.errno == {{{ cDefs.EAGAIN }}} && !(stream.flags & {{{ cDefs.O_NONBLOCK }}})) { + deferred ??= new SyncDeferred(); + waitStreamEvents(stream, {{{ cDefs.POLLOUT }}}, step); + return deferred; + } + if (total) break; + if (deferred) return deferred.reject(e); + throw e; + } + stream.canWait = false; + if (!curr) break; + total += curr; + ptr += curr; + len -= curr; + } + if (deferred) return deferred.resolve(total); + return total; + } + return step(); + }, +#endif #else // MEMFS filesystem disabled lite handling of stdout and stderr: $printCharBuffers: [null, [], []], // 1 => stdout, 2 => stderr @@ -257,6 +309,11 @@ var WasiLibrary = { #endif // SYSCALLS_REQUIRE_FILESYSTEM #if SYSCALLS_REQUIRE_FILESYSTEM +#if PTHREADS && !ASYNCIFY + // Not under ASYNCIFY, where 'auto' would make the import itself suspend, + // which a printf from a callback cannot afford. + fd_write__async: 'auto', +#endif #elif (!MINIMAL_RUNTIME || EXIT_RUNTIME) $flush_NO_FILESYSTEM: () => { // flush anything remaining in the buffers during shutdown @@ -270,10 +327,23 @@ var WasiLibrary = { fd_write__postset: () => addAtExit('flush_NO_FILESYSTEM()'), #else #endif - fd_write: (fd, iov, iovcnt, pnum) => { + fd_write: (fd, iov, iovcnt, pnum{{{ SYSCALLS_REQUIRE_FILESYSTEM && PTHREADS && !ASYNCIFY ? ', sync' : '' }}}) => { #if SYSCALLS_REQUIRE_FILESYSTEM var stream = SYSCALLS.getStreamFromFD(fd); +#if PTHREADS && !ASYNCIFY + var num = sync ? doWritev(stream, iov, iovcnt) : doWritevBlocking(stream, iov, iovcnt); + if (num instanceof SyncDeferred) { + return num.then((num) => { + {{{ makeSetValue('pnum', 0, 'num', SIZE_TYPE) }}}; + return 0; + }, (e) => { + if (e.name !== 'ErrnoError') throw e; + return e.errno; + }); + } +#else var num = doWritev(stream, iov, iovcnt); +#endif #else // hack to support printf in SYSCALLS_REQUIRE_FILESYSTEM=0 var num = 0; diff --git a/system/lib/pthread/proxying.c b/system/lib/pthread/proxying.c index 1342cb2cc924f..d655650c0dd4d 100644 --- a/system/lib/pthread/proxying.c +++ b/system/lib/pthread/proxying.c @@ -641,13 +641,11 @@ static void run_js_func(void* arg) { static void run_js_func_with_ctx(em_proxying_ctx* ctx, void* arg) { proxied_js_func_t* f = (proxied_js_func_t*)arg; + // arg is the caller's stack frame, which is gone once the JS call completes + // the proxy, so it is checked first. + assert(!f->owned); _emscripten_receive_on_main_thread_js( f->funcIndex, f->emAsmAddr, f->callingThread, f->bufSize, f->argBuffer, ctx, arg); - - // run_js_func_with_ctx is always synchronously proxied and therefore arg - // should never be owned on the main thread (i.e. the argument here always - // exists on the stack of the calling thread, it's never copied/malloced). - assert(!f->owned); } void _emscripten_run_js_on_main_thread_done(void* ctx, void* arg, double result) { diff --git a/test/codesize/test_codesize_minimal_pthreads.json b/test/codesize/test_codesize_minimal_pthreads.json index d7133be62a0cf..c3ef8621abbd4 100644 --- a/test/codesize/test_codesize_minimal_pthreads.json +++ b/test/codesize/test_codesize_minimal_pthreads.json @@ -1,10 +1,10 @@ { "a.out.js": 6889, - "a.out.js.gz": 3404, + "a.out.js.gz": 3403, "a.out.nodebug.wasm": 19139, "a.out.nodebug.wasm.gz": 8819, "total": 26028, - "total_gz": 12223, + "total_gz": 12222, "sent": [ "a (memory)", "b (exit)", diff --git a/test/codesize/test_codesize_minimal_pthreads_memgrowth.json b/test/codesize/test_codesize_minimal_pthreads_memgrowth.json index e209dad35c529..19dc1bc8a62f9 100644 --- a/test/codesize/test_codesize_minimal_pthreads_memgrowth.json +++ b/test/codesize/test_codesize_minimal_pthreads_memgrowth.json @@ -1,10 +1,10 @@ { "a.out.js": 7346, - "a.out.js.gz": 3620, + "a.out.js.gz": 3622, "a.out.nodebug.wasm": 19140, "a.out.nodebug.wasm.gz": 8820, "total": 26486, - "total_gz": 12440, + "total_gz": 12442, "sent": [ "a (memory)", "b (exit)", diff --git a/test/other/test_pipe_write_blocking.c b/test/other/test_pipe_write_blocking.c new file mode 100644 index 0000000000000..52dc81035e19c --- /dev/null +++ b/test/other/test_pipe_write_blocking.c @@ -0,0 +1,149 @@ +/* + * Copyright 2026 The Emscripten Authors. All rights reserved. + * Emscripten is available under two separate licenses, the MIT license and the + * University of Illinois/NCSA Open Source License. Both these licenses can be + * found in the LICENSE file. + */ + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#define CAPACITY 65536 +#define TOTAL (1 << 20) +#define BIG (1 << 18) +#define CHUNK 16384 + +int p[2]; +_Atomic size_t consumed; + +void consume(const unsigned char* buf, ssize_t n) { + for (ssize_t i = 0; i < n; i++) + assert(buf[i] == ((consumed + i) & 0xff)); + consumed += n; +} + +void fill(unsigned char* buf, size_t off, size_t n) { + for (size_t i = 0; i < n; i++) + buf[i] = (off + i) & 0xff; +} + +void write_all(void) { + unsigned char* buf = malloc(BIG); + fill(buf, 0, BIG); + assert(write(p[1], buf, BIG) == BIG); + for (size_t written = BIG; written < TOTAL; written += 4096) { + fill(buf, written, 4096); + assert(write(p[1], buf, 4096) == 4096); + } + free(buf); +} + +void* reader(void* arg) { + unsigned char buf[CHUNK]; + while (consumed < TOTAL) { + ssize_t n = read(p[0], buf, sizeof buf); + if (n < 0) { + assert(errno == EAGAIN); + struct pollfd pfd = {p[0], POLLIN, 0}; + assert(poll(&pfd, 1, -1) == 1); + continue; + } + consume(buf, n); + } + return NULL; +} + +void* closer(void* arg) { + usleep(50 * 1000); + assert(close(p[0]) == 0); + return NULL; +} + +// The main thread's own write, re-entered from a proxied call, is not held. +// Calls the import directly so an unstored count reads as 0. +EMSCRIPTEN_KEEPALIVE int main_writer(int n) { + unsigned char* buf = malloc(n); + fill(buf, 0, n); + __wasi_ciovec_t iov = {buf, n}; + __wasi_size_t num = 0; + assert(__wasi_fd_write(p[1], &iov, 1, &num) == 0); + free(buf); + return num; +} + +void read_back(size_t n) { + unsigned char buf[CHUNK]; + consumed = 0; + while (consumed < n) { + ssize_t got = read(p[0], buf, sizeof buf); + if (got < 0) { + assert(errno == EAGAIN); + usleep(1000); + continue; + } + consume(buf, got); + } +} + +void* big_writer(void* arg) { + unsigned char* buf = malloc(BIG); + fill(buf, 0, BIG); + assert(write(p[1], buf, BIG) == BIG); + // Held whichever thread reads. It completes inside the read that frees the + // last room, before that chunk is counted. + assert(consumed >= BIG - CAPACITY - CHUNK); + free(buf); + return NULL; +} + +void* run(void* arg) { + assert(pipe(p) == 0); + unsigned char* big = malloc(BIG); + memset(big, 'x', BIG); + pthread_t t; + assert(pthread_create(&t, NULL, closer, NULL) == 0); + assert(write(p[1], big, BIG) == CAPACITY); + assert(write(p[1], big, 1) == -1 && errno == EPIPE); + free(big); + assert(pthread_join(t, NULL) == 0); + assert(close(p[1]) == 0); + + assert(pipe(p) == 0); + assert(pthread_create(&t, NULL, reader, NULL) == 0); + write_all(); + assert(close(p[1]) == 0); + assert(pthread_join(t, NULL) == 0); + assert(consumed == TOTAL); + assert(close(p[0]) == 0); + + assert(pipe(p) == 0); + assert(MAIN_THREAD_EM_ASM_INT({ return _main_writer($0); }, BIG) == BIG); + read_back(BIG); + assert(close(p[0]) == 0); + assert(close(p[1]) == 0); + return NULL; +} + +// Without PROXY_TO_PTHREAD main waits in pthread_join, off its event loop. +int main() { + pthread_t t; + assert(pthread_create(&t, NULL, run, NULL) == 0); + assert(pthread_join(t, NULL) == 0); + + assert(pipe(p) == 0); + assert(pthread_create(&t, NULL, big_writer, NULL) == 0); + read_back(BIG); + assert(pthread_join(t, NULL) == 0); + assert(close(p[0]) == 0); + assert(close(p[1]) == 0); + printf("done\n"); + return 0; +} diff --git a/test/test_other.py b/test/test_other.py index 21e718438b0f8..8a97ce470ff21 100644 --- a/test/test_other.py +++ b/test/test_other.py @@ -13801,6 +13801,16 @@ def test_epoll_callback_level(self): # the callback every tick: documents the spin contract (use EPOLLET/unregister). self.do_runf('other/test_epoll_callback_level.c', 'done\n', cflags=['-sFORCE_FILESYSTEM', '-sEXIT_RUNTIME']) + @parameterized({ + '': (['-sPROXY_TO_PTHREAD'],), + 'main_joins': ([],), + # MAIN_MODULE=1 also sends library functions to wasm by name. + 'main_module': (['-sMAIN_MODULE', '-Wno-experimental'],), + }) + def test_pipe_write_blocking(self, args): + self.do_runf('other/test_pipe_write_blocking.c', 'done\n', + cflags=['-pthread', '-sEXIT_RUNTIME'] + args) + def test_epoll_callback_coalesce(self): # A burst of edges before the loop turns (many writes, a run of socket # chunks) schedules one delivery task, not one per edge: on hosts where