WASMFS Support blocking poll/select

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',

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);
   }
 });

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;
 }

system/lib/wasmfs/file.h

@@ -24,6 +24,7 @@
 #include <wasi/api.h>
 
 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<File> {
     SymlinkKind = 3
   };
 
+  // For notifying when `poll()` may change
+  virtual WaitQueue* getWaitQueue() { return nullptr; }
+
   const FileKind kind;
 
   template<class T> bool is() const {

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<uint8_t> 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<PipeData> 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,13 +87,15 @@ 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;
       }
       buf[i] = data->bytes.front();
       data->bytes.pop();
     }
+    data->writerQueue.notifyAll();
     return len;
   }
 

system/lib/wasmfs/syscalls.cpp

@@ -12,6 +12,7 @@
 #include <emscripten/heap.h>
 #include <emscripten/html5.h>
 #include <emscripten/syscalls.h>
+#include <emscripten/threading.h>
 #include <errno.h>
 #include <math.h>
 #include <mutex>
@@ -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<std::shared_ptr<OpenFileState>>& files,
+                  std::vector<WaitRegistration>& 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<std::shared_ptr<OpenFileState>> files;
+  std::vector<WaitRegistration> 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);
 }