diff --git a/src/lib/libsigs.js b/src/lib/libsigs.js index cb70908f30512..0aa47c9459242 100644 --- a/src/lib/libsigs.js +++ b/src/lib/libsigs.js @@ -464,6 +464,8 @@ sigs = { _wasmfs_opfs_set_size_access__sig: 'vpijp', _wasmfs_opfs_set_size_file__sig: 'vpijp', _wasmfs_opfs_write_access__sig: 'iipij', + _wasmfs_poll_notify__sig: 'vp', + _wasmfs_poll_wait__sig: 'ipi', _wasmfs_stdin_get_char__sig: 'i', _wasmfs_thread_utils_heartbeat__sig: 'vp', alBuffer3f__sig: 'viifff', diff --git a/src/lib/libwasmfs.js b/src/lib/libwasmfs.js index aaff5b4558dc3..13d7d3be3fe9f 100644 --- a/src/lib/libwasmfs.js +++ b/src/lib/libwasmfs.js @@ -24,6 +24,7 @@ addToLibrary({ $wasmFSPreloadingFlushed: false, $wasmFSDevices: {}, $wasmFSDeviceStreams: {}, + $wasmFSPollWaiters: {}, $FS__deps: [ '$MEMFS', @@ -563,5 +564,33 @@ addToLibrary({ return c; } return -1; + }, + + _wasmfs_poll_wait__async: 'auto', + _wasmfs_poll_wait: (key, ms, sync) => { + if (sync) { + return 0; + } else { + return new Promise((resolve) => { + var timer; + if (ms >= 0) { + timer = setTimeout(() => { + delete wasmFSPollWaiters[key]; + resolve(0); + }, ms); + } + wasmFSPollWaiters[key] = {resolve, timer}; + }); + } + }, + + _wasmfs_poll_notify: (key) => { + var entry = wasmFSPollWaiters[key]; + if (!entry) + return; + + delete wasmFSPollWaiters[key]; + clearTimeout(entry.timer); + entry.resolve(1); } }); diff --git a/system/lib/standalone/standalone.c b/system/lib/standalone/standalone.c index 58e3c208325b7..40d569e1221ef 100644 --- a/system/lib/standalone/standalone.c +++ b/system/lib/standalone/standalone.c @@ -319,6 +319,17 @@ int _wasmfs_stdin_get_char(void) { return c; } +int _wasmfs_poll_wait(void* key, int ms) { + (void)key; + (void)ms; + return 0; +} + +void _wasmfs_poll_notify(void* key) { + (void)key; + return; +} + // In the non-standalone build we define this helper function in JS to avoid // signature mismatch issues. // See: https://github.com/emscripten-core/posixtestsuite/issues/6 @@ -333,6 +344,8 @@ int _setitimer_js(int which, double timeout) { return -1; } +weak int emscripten_is_main_browser_thread() { return true; } + weak uintptr_t emscripten_stack_snapshot(void) { return 0; } diff --git a/system/lib/wasmfs/file.h b/system/lib/wasmfs/file.h index fab493801ddd4..0da38fd8e580d 100644 --- a/system/lib/wasmfs/file.h +++ b/system/lib/wasmfs/file.h @@ -24,6 +24,7 @@ #include namespace wasmfs { +class WaitQueue; // Note: The general locking strategy for all Files is to only hold 1 lock at a // time to prevent deadlock. This methodology can be seen in getDirs(). @@ -56,6 +57,9 @@ class File : public std::enable_shared_from_this { SymlinkKind = 3 }; + // For notifying when `poll()` may change + virtual WaitQueue* getWaitQueue() { return nullptr; } + const FileKind kind; template bool is() const { diff --git a/system/lib/wasmfs/pipe_backend.h b/system/lib/wasmfs/pipe_backend.h index 256bf9554b420..4e93c32ab4068 100644 --- a/system/lib/wasmfs/pipe_backend.h +++ b/system/lib/wasmfs/pipe_backend.h @@ -11,6 +11,7 @@ #include "file.h" #include "support.h" +#include "wait_queue.h" namespace wasmfs { @@ -24,6 +25,8 @@ struct PipeData { std::queue bytes; bool readClosed = false; bool writeClosed = false; + WaitQueue readerQueue; + WaitQueue writerQueue; }; // A PipeFile is a simple file that has a reference to a PipeData that it @@ -35,10 +38,20 @@ class PipeFile : public DataFile { std::shared_ptr data; bool writeEnd; + WaitQueue* getWaitQueue() override { + return writeEnd ? &data->writerQueue : &data->readerQueue; + } + int open(oflags_t) override { return 0; } int close() override { std::lock_guard lock(data->mutex); - (writeEnd ? data->writeClosed : data->readClosed) = true; + if (writeEnd) { + data->writeClosed = true; + data->readerQueue.notifyAll(); + } else { + data->readClosed = true; + data->writerQueue.notifyAll(); + } return 0; } @@ -50,6 +63,8 @@ class PipeFile : public DataFile { for (size_t i = 0; i < len; i++) { data->bytes.push(buf[i]); } + + data->readerQueue.notifyAll(); return len; } @@ -72,6 +87,7 @@ class PipeFile : public DataFile { for (size_t i = 0; i < len; i++) { if (data->bytes.empty()) { if (i) { + data->writerQueue.notifyAll(); return i; } return data->writeClosed ? 0 : -EAGAIN; @@ -79,6 +95,7 @@ class PipeFile : public DataFile { buf[i] = data->bytes.front(); data->bytes.pop(); } + data->writerQueue.notifyAll(); return len; } diff --git a/system/lib/wasmfs/syscalls.cpp b/system/lib/wasmfs/syscalls.cpp index 59df616def9e4..7f91de70478d2 100644 --- a/system/lib/wasmfs/syscalls.cpp +++ b/system/lib/wasmfs/syscalls.cpp @@ -12,6 +12,7 @@ #include #include #include +#include #include #include #include @@ -1436,7 +1437,13 @@ int __syscall_pipe2(int fd[2], int flags) { return 0; } -int __syscall_poll(struct pollfd* fds, nfds_t nfds, int timeout) { +// Compute revents for every fd and return how many are ready. If `waiter` is +// non-null, also register it in the wait queue of each pollable file. +static int doPoll(struct pollfd* fds, + nfds_t nfds, + PollWaiter* waiter, + std::vector>& files, + std::vector& regs) { auto fileTable = wasmFS.getFileTable().locked(); // Process the list of FDs and compute their revents masks. Count the number @@ -1456,6 +1463,13 @@ int __syscall_poll(struct pollfd* fds, nfds_t nfds, int timeout) { if (openFile) { auto flags = openFile->locked().getFlags(); auto file = openFile->locked().getFile(); + if (waiter) { + if (auto* queue = file->getWaitQueue()) { + // Keep the file, and so its queue, alive while registered. + files.push_back(openFile); + regs.emplace_back(*queue, *waiter); + } + } // POLLERR and POLLHUP are reported whether or not they were requested. mask = file->locked().poll(flags) & (pollfd->events | POLLERR | POLLHUP); } @@ -1464,14 +1478,38 @@ int __syscall_poll(struct pollfd* fds, nfds_t nfds, int timeout) { } pollfd->revents = mask; } - // TODO: This should block based on the timeout. The old FS did not do so due - // to web limitations, which we should perhaps revisit (especially with - // pthreads and asyncify). return nonzero; } -// libc routes zero-timeout poll() calls here (see musl's poll.c). WasmFS's -// __syscall_poll never blocks, so the zero-timeout probe is the same call. +int __syscall_poll(struct pollfd* fds, nfds_t nfds, int timeout) { + PollWaiter waiter; + std::vector> files; + std::vector regs; + + if (timeout == 0) { + return doPoll(fds, nfds, nullptr, files, regs); + } + + files.reserve(nfds); + regs.reserve(nfds); + + PollWaiter::Deadline deadline; + if (timeout > 0) { + deadline = + std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout); + } + + // The first scan registers the waiter; later ones only recompute. + int nonzero = doPoll(fds, nfds, &waiter, files, regs); + bool timedOut = false; + while (!nonzero && !timedOut) { + timedOut = !waiter.wait(deadline); + nonzero = doPoll(fds, nfds, nullptr, files, regs); + } + return nonzero; +} + +// libc routes zero-timeout poll() calls here (see musl's poll.c). int __syscall_poll_nonblocking(struct pollfd* fds, nfds_t nfds) { return __syscall_poll(fds, nfds, 0); } diff --git a/system/lib/wasmfs/wait_queue.h b/system/lib/wasmfs/wait_queue.h new file mode 100644 index 0000000000000..e6fc92046b571 --- /dev/null +++ b/system/lib/wasmfs/wait_queue.h @@ -0,0 +1,195 @@ +// 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. + +// Wait queues let a thread block in poll() until the readiness of one of the +// files it watches may have changed. +// +// - A PollWaiter is owned by one blocking poll() call (on its stack). +// - A WaitQueue is owned by a pollable object (e.g. one side of a pipe) and +// holds the waiters currently interested in it. +// - A WaitRegistration puts a waiter in a queue for as long as it lives. +// +// The object that owns a queue calls notifyAll() whenever its poll() result may +// have changed. That marks every registered waiter as triggered and wakes it. + +#pragma once + +#include "wasmfs_internal.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace wasmfs { + +class WaitQueue; +class WaitRegistration; + +class PollWaiter { +public: + using Deadline = std::optional; + + PollWaiter() = default; + PollWaiter(const PollWaiter&) = delete; + PollWaiter& operator=(const PollWaiter&) = delete; + + // Sleep until a queue this waiter is registered in is notified, or until + // `deadline` passes (std::nullopt waits forever). Returns immediately if a + // notification arrived since the previous wait(). Clears the notification + // before returning. Returns true if notified, false on timeout. + bool wait(Deadline deadline) { + if (emscripten_is_main_browser_thread()) { + return waitAsync(deadline); + } + + std::unique_lock lock(mutex); + auto isTriggered = [&]() { return triggered; }; + bool notified = true; + if (deadline) { + notified = cond.wait_until(lock, *deadline, isTriggered); + } else { + cond.wait(lock, isTriggered); + } + triggered = false; + return notified; + } + +private: + std::mutex mutex; + std::condition_variable cond; + bool triggered = false; + bool async = false; + + bool waitAsync(Deadline deadline) { + { + std::lock_guard lock(mutex); + if (triggered) { + triggered = false; + return true; + } + async = true; + } + + int ms = -1; + if (deadline) { + auto remaining = std::chrono::ceil( + *deadline - std::chrono::steady_clock::now()); + ms = static_cast(std::max(remaining.count(), 0)); + } + int res = _wasmfs_poll_wait(this, ms); + + std::lock_guard lock(mutex); + bool notified = res || triggered; + triggered = false; + return notified; + } + + friend class WaitQueue; +}; + +class WaitQueue { +public: + WaitQueue() = default; + WaitQueue(const WaitQueue&) = delete; + WaitQueue& operator=(const WaitQueue&) = delete; + + ~WaitQueue() { + assert(waiters.empty() && "WaitQueue destroyed while waiters registered"); + } + + // Wake every registered waiter. Call after any change that may affect the + // poll() result of the object owning this queue. + void notifyAll() { + // Keep the queue locked while touching waiters: a waiter cannot be + // unregistered (and so cannot be destroyed) until we are done with it. + std::lock_guard lock(mutex); + for (auto* waiter : waiters) { + bool isAsync = false; + { + std::lock_guard waiterLock(waiter->mutex); + waiter->triggered = true; + isAsync = waiter->async; + } + + if (isAsync) { + if (emscripten_is_main_browser_thread()) { + _wasmfs_poll_notify(waiter); + } else { + // poll() coming from main thread + emscripten_proxy_async(emscripten_proxy_get_system_queue(), + emscripten_main_runtime_thread_id(), + _wasmfs_poll_notify, + waiter); + } + } else { + waiter->cond.notify_one(); + } + } + } + +private: + std::mutex mutex; + // A waiter may appear more than once (e.g. the same fd passed twice to + // poll()); each registration adds and removes exactly one entry. + std::vector waiters; + + void add(PollWaiter* waiter) { + std::lock_guard lock(mutex); + waiters.push_back(waiter); + } + + void remove(PollWaiter* waiter) { + std::lock_guard lock(mutex); + auto it = std::find(waiters.begin(), waiters.end(), waiter); + assert(it != waiters.end()); + waiters.erase(it); + } + + friend class WaitRegistration; +}; + +class WaitRegistration { +public: + WaitRegistration(WaitQueue& queue, PollWaiter& waiter) + : queue(&queue), waiter(&waiter) { + queue.add(&waiter); + } + + ~WaitRegistration() { reset(); } + + WaitRegistration(const WaitRegistration&) = delete; + WaitRegistration& operator=(const WaitRegistration&) = delete; + + WaitRegistration(WaitRegistration&& other) noexcept + : queue(std::exchange(other.queue, nullptr)), waiter(other.waiter) {} + + WaitRegistration& operator=(WaitRegistration&& other) noexcept { + if (this != &other) { + reset(); + queue = std::exchange(other.queue, nullptr); + waiter = other.waiter; + } + return *this; + } + +private: + WaitQueue* queue; + PollWaiter* waiter; + + void reset() { + if (queue) { + queue->remove(waiter); + queue = nullptr; + } + } +}; + +} // namespace wasmfs diff --git a/system/lib/wasmfs/wasmfs_internal.h b/system/lib/wasmfs/wasmfs_internal.h index 784b1cdd33c6d..32e32638665cb 100644 --- a/system/lib/wasmfs/wasmfs_internal.h +++ b/system/lib/wasmfs/wasmfs_internal.h @@ -21,6 +21,10 @@ void _wasmfs_copy_preloaded_file_data(uint32_t index, uint8_t* data); // Returns the next character from stdin, or -1 on EOF. int _wasmfs_stdin_get_char(void); +// if notified returns 1, 0 for timeout +int _wasmfs_poll_wait(void* key, int ms); +void _wasmfs_poll_notify(void* key); + #ifdef __cplusplus } #endif diff --git a/test/core/test_poll_blocking_asyncify_thread_writer.c b/test/core/test_poll_blocking_asyncify_thread_writer.c new file mode 100644 index 0000000000000..72d76b88c4d70 --- /dev/null +++ b/test/core/test_poll_blocking_asyncify_thread_writer.c @@ -0,0 +1,82 @@ +/* + * 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. + */ + +// poll() suspended on the main thread (via ASYNCIFY/JSPI) must be woken when +// another thread writes to the polled pipe. Unlike test_poll_blocking.c this +// must not be built with PROXY_TO_PTHREAD: the poll has to run on the real main +// thread, where it cannot block and suspends instead. + +#include +#include +#include +#include +#include +#include +#include +#include + +#define WRITE_DELAY_MS 300 +// Long enough that waking only at the timeout is clearly distinguishable from +// being woken by the write. +#define POLL_TIMEOUT_MS 3000 +// It is possible for the node timers (such as setTimeout or Atomics.wait) to +// wake up slightly earlier than requested. Because we measure times accurately +// using clock_gettime, we give tests a 20 milliseconds error margin to avoid +// flaky timeouts. +#define TIMEOUT_MARGIN_MS 20 + +int64_t timespec_delta_ms(struct timespec* begin, struct timespec* end) { + int64_t delta_sec = end->tv_sec - begin->tv_sec; + int64_t delta_nsec = end->tv_nsec - begin->tv_nsec; + + assert(delta_sec >= 0); + assert(delta_nsec > -1000000000 && delta_nsec < 1000000000); + + int64_t delta_ms = (delta_sec * 1000000000LL + delta_nsec) / 1000000; + assert(delta_ms >= 0); + return delta_ms; +} + +int pipe_fds[2]; + +void* write_after_sleep(void* arg) { + usleep(WRITE_DELAY_MS * 1000); + printf("write_after_sleep\n"); + const char* t = "test\n"; + assert(write(pipe_fds[1], t, strlen(t)) == (ssize_t)strlen(t)); + return NULL; +} + +int main() { + struct timespec begin = {0}; + struct timespec end = {0}; + pthread_t tid; + + assert(pipe(pipe_fds) == 0); + struct pollfd fds = {pipe_fds[0], POLLIN, 0}; + + clock_gettime(CLOCK_MONOTONIC, &begin); + assert(pthread_create(&tid, NULL, write_after_sleep, NULL) == 0); + // A finite timeout, so a missed wakeup fails the duration check below + // instead of hanging the test. + assert(poll(&fds, 1, POLL_TIMEOUT_MS) == 1); + clock_gettime(CLOCK_MONOTONIC, &end); + assert(fds.revents & POLLIN); + + int64_t duration = timespec_delta_ms(&begin, &end); + printf(" -> duration: %lld ms\n", duration); + assert(duration >= WRITE_DELAY_MS - TIMEOUT_MARGIN_MS); + // Woken by the write, not by the timeout. + assert(duration < POLL_TIMEOUT_MS / 2); + + // The writer is done by now, so joining from the main thread is brief. + pthread_join(tid, NULL); + close(pipe_fds[0]); + close(pipe_fds[1]); + printf("done\n"); + return 0; +} diff --git a/test/test_core.py b/test/test_core.py index 94d51b5273b9e..3be319aca163d 100644 --- a/test/test_core.py +++ b/test/test_core.py @@ -9741,27 +9741,22 @@ def test_syscall_intercept(self): self.do_core_test('test_syscall_intercept.c') @requires_pthreads - @no_wasmfs('https://github.com/emscripten-core/emscripten/issues/26736') def test_select_blocking(self): self.do_runf('core/test_select_blocking.c', cflags=['-pthread', '-sPROXY_TO_PTHREAD=1', '-sEXIT_RUNTIME=1']) @requires_pthreads - @no_wasmfs('https://github.com/emscripten-core/emscripten/issues/26736') def test_pselect_blocking(self): self.do_runf('core/test_pselect_blocking.c', cflags=['-pthread', '-sPROXY_TO_PTHREAD=1', '-sEXIT_RUNTIME=1']) @requires_pthreads - @no_wasmfs('https://github.com/emscripten-core/emscripten/issues/26736') def test_poll_blocking(self): self.do_runf('core/test_poll_blocking.c', cflags=['-pthread', '-sPROXY_TO_PTHREAD=1', '-sEXIT_RUNTIME=1']) @requires_pthreads - @no_wasmfs('https://github.com/emscripten-core/emscripten/issues/26736') def test_ppoll_blocking(self): self.do_runf('core/test_ppoll_blocking.c', cflags=['-pthread', '-sPROXY_TO_PTHREAD=1', '-sEXIT_RUNTIME=1']) @with_asyncify_and_jspi - @no_wasmfs('https://github.com/emscripten-core/emscripten/issues/26736') def test_poll_blocking_asyncify(self): if self.get_setting('JSPI') and engine_is_v8(self.get_current_js_engine()): self.skipTest('test requires setTimeout which is not supported under v8') @@ -9775,12 +9770,23 @@ def test_poll_blocking_asyncify(self): @requires_node_25 @with_asyncify_and_jspi @requires_pthreads - @no_wasmfs('https://github.com/emscripten-core/emscripten/issues/26736') def test_poll_blocking_asyncify_pthread(self): if self.get_setting('JSPI') and engine_is_v8(self.get_current_js_engine()): self.skipTest('test requires setTimeout which is not supported under v8') self.do_runf('core/test_poll_blocking.c', cflags=['-pthread', '-sPROXY_TO_PTHREAD=1', '-sEXIT_RUNTIME=1']) + # A poll() suspended on the main thread must wake up when another thread + # writes to the pipe. + # See test_poll_blocking_asyncify_pthread for @requires_node_25. + @requires_node_25 + @with_asyncify_and_jspi + @requires_pthreads + def test_poll_blocking_asyncify_thread_writer(self): + if self.get_setting('JSPI') and engine_is_v8(self.get_current_js_engine()): + self.skipTest('test requires setTimeout which is not supported under v8') + self.do_runf('core/test_poll_blocking_asyncify_thread_writer.c', 'done\n', + cflags=['-pthread', '-sPROXY_TO_PTHREAD=0', '-sEXIT_RUNTIME=1', '-sPTHREAD_POOL_SIZE=1']) + @with_asyncify_and_jspi @needs_epoll def test_epoll_blocking_asyncify(self):