Surface IPC end-of-stream on win32

shared/win32/ipc.c

@@ -23,10 +23,19 @@ bare_ipc__on_alloc(uv_handle_t *handle, size_t suggested_size, uv_buf_t *buf) {
 
 void
 bare_ipc__on_read(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf) {
-  // End-of-stream when the worklet closes its IPC. Stop reading rather than
-  // aborting; surfacing EOF to the host is not yet implemented for win32.
+  bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) stream);
+
+  // End-of-stream when the worklet closes its IPC. Mark EOF and notify the
+  // reader so a subsequent bare_ipc_read() returns it as a zero-length read.
   if (nread == UV_EOF) {
     uv_read_stop(stream);
+
+    uv_mutex_lock(&ipc->reading);
+    ipc->eof = true;
+    uv_mutex_unlock(&ipc->reading);
+
+    if (ipc->poll && ipc->poll->cb) ipc->poll->cb(ipc->poll, bare_ipc_readable);
+
     return;
   }
 
@@ -41,8 +50,6 @@ bare_ipc__on_read(uv_stream_t *stream, ssize_t nread, const uv_buf_t *buf) {
   err = uv_read_stop(stream);
   assert(err == 0);
 
-  bare_ipc_t *ipc = (bare_ipc_t *) uv_handle_get_data((uv_handle_t *) stream);
-
   uv_mutex_lock(&ipc->reading);
 
   memcpy(ipc->read_buffer.base, buf->base, nread);
@@ -148,6 +155,7 @@ bare_ipc_init(bare_ipc_t *ipc, bare_worklet_t *worklet) {
   assert(err == 0);
 
   ipc->read_buffer.len = 0;
+  ipc->eof = false;
   ipc->poll = NULL;
 
   uv_barrier_wait(&ipc->ready);
@@ -185,6 +193,15 @@ bare_ipc_read(bare_ipc_t *ipc, void **data, size_t *len) {
     return 0;
   }
 
+  if (ipc->eof) {
+    *data = ipc->read_buffer.base;
+    *len = 0;
+
+    uv_mutex_unlock(&ipc->reading);
+
+    return 0;
+  }
+
   uv_mutex_unlock(&ipc->reading);
 
   return bare_ipc_would_block;

shared/win32/ipc.h

@@ -5,6 +5,7 @@
 extern "C" {
 #endif
 
+#include <stdbool.h>
 #include <uv.h>
 
 #include "../ipc.h"
@@ -13,6 +14,8 @@ struct bare_ipc_s {
   int incoming;
   int outgoing;
 
+  bool eof;
+
   struct {
     char base[BARE_IPC_READ_BUFFER_SIZE];
     size_t len;

shared/worklet.c

@@ -489,6 +489,12 @@ bare_worklet__on_thread(void *opaque) {
 
   state->finished = true;
 
+  // The 'exit' handler closes Bare.IPC. Flush the loop so the close completes
+  // and the host sees EOF before we park waiting for teardown. This is a no-op
+  // where uv_close is synchronous (posix) but is required on platforms where
+  // the close only completes via the loop (win32).
+  uv_run(&loop, UV_RUN_NOWAIT);
+
   uv_sem_wait(&finished);
 
   uv_sem_destroy(&finished);

test/shared/CMakeLists.txt

@@ -1,21 +1,14 @@
 list(APPEND tests
   worklet-assets
   worklet-destroy-before-start
+  worklet-exit-eof
   worklet-ipc
   worklet-ipc-large-write
+  worklet-jsend-eof
   worklet-suspend-resume
   worklet-terminate-after-suspend
 )
 
-# The win32 IPC layer does not yet surface end-of-stream, so the EOF tests only
-# run on platforms with the posix read path.
-if(NOT WIN32)
-  list(APPEND tests
-    worklet-exit-eof
-    worklet-jsend-eof
-  )
-endif()
-
 execute_process(
   COMMAND node fixtures/generate.js
   WORKING_DIRECTORY ${CMAKE_CURRENT_LIST_DIR}

test/shared/worklet-exit-eof.c

@@ -4,44 +4,73 @@
 #include "../../shared/worklet.h"
 
 // When the worklet exits on its own (Bare.exit()), Bare.IPC is closed as part
-// of process exit, so the host read reaches EOF (returns 0) without the host
-// having to call terminate/destroy first.
+// of process exit, so the host's read reaches EOF (a zero-length read) without
+// the host having to call terminate/destroy first.
+
+static uv_async_t finished;
+static bool eof = false;
+
+static void
+on_poll(bare_ipc_poll_t *poll, int events) {
+  if ((events & bare_ipc_readable) == 0) return;
+
+  void *data;
+  size_t len;
+  int err = bare_ipc_read(bare_ipc_poll_get_ipc(poll), &data, &len);
+  assert(err == 0 || err == bare_ipc_would_block);
+
+  if (err == 0 && len == 0) {
+    eof = true;
+
+    bare_ipc_poll_stop(poll);
+
+    uv_async_send(&finished);
+  }
+}
+
+static void
+on_finish(uv_async_t *handle) {
+  uv_close((uv_handle_t *) handle, NULL);
+}
 
 int
 main() {
-  int e;
+  int err;
+
+  uv_loop_t *loop = uv_default_loop();
+  err = uv_async_init(loop, &finished, on_finish);
+  assert(err == 0);
 
   bare_worklet_t worklet;
-  e = bare_worklet_init(&worklet, NULL);
-  assert(e == 0);
+  err = bare_worklet_init(&worklet, NULL);
+  assert(err == 0);
 
   char *code = "setTimeout(() => Bare.exit(0), 50)";
-
   uv_buf_t source = uv_buf_init(code, strlen(code));
-
-  e = bare_worklet_start(&worklet, "app.js", &source, 0, NULL);
-  assert(e == 0);
+  err = bare_worklet_start(&worklet, "app.js", &source, 0, NULL);
+  assert(err == 0);
 
   bare_ipc_t ipc;
-  e = bare_ipc_init(&ipc, &worklet);
-  assert(e == 0);
+  err = bare_ipc_init(&ipc, &worklet);
+  assert(err == 0);
 
-  void *data;
-  size_t len;
+  bare_ipc_poll_t poll;
+  err = bare_ipc_poll_init(&poll, &ipc);
+  assert(err == 0);
 
-  bool eof = false;
-  for (int i = 0; i < 200; i++) {
-    e = bare_ipc_read(&ipc, &data, &len);
-    assert(e == 0 || e == bare_ipc_would_block);
-    if (e == 0 && len == 0) {
-      eof = true;
-      break;
-    }
-    uv_sleep(10);
-  }
+  err = bare_ipc_poll_start(&poll, bare_ipc_readable, on_poll);
+  assert(err == 0);
+
+  err = uv_run(loop, UV_RUN_DEFAULT);
+  assert(err == 0);
 
   assert(eof); // The self-exit closed Bare.IPC and the host saw EOF
 
+  err = uv_loop_close(loop);
+  assert(err == 0);
+
+  bare_ipc_poll_destroy(&poll);
+
   bare_ipc_destroy(&ipc);
 
   bare_worklet_destroy(&worklet);

test/shared/worklet-jsend-eof.c

@@ -3,46 +3,74 @@
 #include "../../shared/ipc.h"
 #include "../../shared/worklet.h"
 
-// Corroborates worklet-exit-no-eof: the host EOF comes from the JS side. When
-// the worklet ends its Bare.IPC (closing the b1 write end), the host read
-// returns 0 (EOF) - the signal the bindings turn into (nil, nil).
+// When the worklet ends its Bare.IPC (closing the write end), the host's read
+// reaches EOF (a zero-length read) - the signal the bindings turn into a null
+// read. This covers the explicit end() path, independent of process exit.
+
+static uv_async_t finished;
+static bool eof = false;
+
+static void
+on_poll(bare_ipc_poll_t *poll, int events) {
+  if ((events & bare_ipc_readable) == 0) return;
+
+  void *data;
+  size_t len;
+  int err = bare_ipc_read(bare_ipc_poll_get_ipc(poll), &data, &len);
+  assert(err == 0 || err == bare_ipc_would_block);
+
+  if (err == 0 && len == 0) {
+    eof = true;
+
+    bare_ipc_poll_stop(poll);
+
+    uv_async_send(&finished);
+  }
+}
+
+static void
+on_finish(uv_async_t *handle) {
+  uv_close((uv_handle_t *) handle, NULL);
+}
 
 int
 main() {
-  int e;
+  int err;
+
+  uv_loop_t *loop = uv_default_loop();
+  err = uv_async_init(loop, &finished, on_finish);
+  assert(err == 0);
 
   bare_worklet_t worklet;
-  e = bare_worklet_init(&worklet, NULL);
-  assert(e == 0);
+  err = bare_worklet_init(&worklet, NULL);
+  assert(err == 0);
 
   char *code = "setTimeout(() => Bare.IPC.end(), 50)";
-
   uv_buf_t source = uv_buf_init(code, strlen(code));
-
-  e = bare_worklet_start(&worklet, "app.js", &source, 0, NULL);
-  assert(e == 0);
+  err = bare_worklet_start(&worklet, "app.js", &source, 0, NULL);
+  assert(err == 0);
 
   bare_ipc_t ipc;
-  e = bare_ipc_init(&ipc, &worklet);
-  assert(e == 0);
+  err = bare_ipc_init(&ipc, &worklet);
+  assert(err == 0);
 
-  void *data;
-  size_t len;
+  bare_ipc_poll_t poll;
+  err = bare_ipc_poll_init(&poll, &ipc);
+  assert(err == 0);
 
-  // Poll for up to ~2s for EOF. Expect read to return 0 once b1 closes.
-  bool eof = false;
-  for (int i = 0; i < 200; i++) {
-    e = bare_ipc_read(&ipc, &data, &len);
-    assert(e == 0 || e == bare_ipc_would_block);
-    if (e == 0 && len == 0) {
-      eof = true;
-      break;
-    }
-    uv_sleep(10);
-  }
+  err = bare_ipc_poll_start(&poll, bare_ipc_readable, on_poll);
+  assert(err == 0);
+
+  err = uv_run(loop, UV_RUN_DEFAULT);
+  assert(err == 0);
 
   assert(eof); // The JS-side end reached the host as EOF
 
+  err = uv_loop_close(loop);
+  assert(err == 0);
+
+  bare_ipc_poll_destroy(&poll);
+
   bare_ipc_destroy(&ipc);
 
   bare_worklet_destroy(&worklet);