Swap worklet IPC from pipes to the in-process queue

android/src/main/java/to/holepunch/bare/kit/IPC.java

@@ -1,6 +1,7 @@
 package to.holepunch.bare.kit;
 
 import java.io.Closeable;
+import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.charset.Charset;
 import java.util.concurrent.atomic.AtomicReference;
@@ -38,6 +39,8 @@ public IPC(Worklet worklet) {
     this.worklet = worklet;
 
     handle = init(worklet.handle);
+
+    worklet.ipc = this;
   }
 
   private native ByteBuffer
@@ -60,6 +63,8 @@ public IPC(Worklet worklet) {
 
   private void
   update() {
+    if (handle == null) return;
+
     update(handle, readable != null, writable != null);
   }
 
@@ -90,10 +95,13 @@ public IPC(Worklet worklet) {
   /**
    * Reads whatever is available without blocking. Returns null when nothing is
    * ready yet, and an empty buffer at end-of-stream, once the worklet has
-   * closed its end. Wait for readable() before retrying a null.
+   * closed its end or this IPC is closed. Wait for readable() before retrying
+   * a null.
    */
   public ByteBuffer
   read() {
+    if (handle == null) return ByteBuffer.allocateDirect(0);
+
     ByteBuffer buffer = read(handle);
 
     if (buffer == null) {
@@ -134,6 +142,8 @@ public IPC(Worklet worklet) {
 
   public int
   write(ByteBuffer data) {
+    if (handle == null) return -1;
+
     ByteBuffer buffer;
 
     if (data.isDirect()) {
@@ -149,6 +159,12 @@ public IPC(Worklet worklet) {
 
   public void
   write(ByteBuffer data, WriteCallback callback) {
+    if (handle == null) {
+      callback.apply(new IOException("IPC closed"));
+
+      return;
+    }
+
     ByteBuffer buffer;
 
     if (data.isDirect()) {
@@ -186,8 +202,18 @@ public IPC(Worklet worklet) {
     }
   }
 
+  /**
+   * Retires this IPC, which must happen before its worklet is destroyed.
+   * terminate() does it for you, and closing twice is a no-op.
+   */
   public void
   close() {
+    if (handle == null) return;
+
     destroy(handle);
+
+    handle = null;
+
+    worklet.ipc = null;
   }
 }

android/src/main/java/to/holepunch/bare/kit/Worklet.java

@@ -51,6 +51,9 @@ public static class Options {
 
   ByteBuffer handle;
 
+  // Unowned: the IPC holds this worklet and clears this when it closes.
+  IPC ipc;
+
   public Worklet(Options options) {
     if (options == null) options = new Options();
 
@@ -158,6 +161,8 @@ public Worklet(Options options) {
 
   public void
   terminate() {
+    if (ipc != null) ipc.close();
+
     terminate(handle);
 
     handle = null;

apple/BareKit/BareKit.h

@@ -85,7 +85,7 @@
 /**
  * Reads whatever is available without blocking. Returns `nil` when nothing is
  * ready yet, and empty data at end-of-stream, once the worklet has closed its
- * end. Wait for `readable` before retrying a `nil`.
+ * end or this ipc is closed. Wait for `readable` before retrying a `nil`.
  */
 - (NSData *_Nullable)read;
 
@@ -95,11 +95,19 @@
  */
 - (void)read:(void (^_Nonnull)(NSData *_Nullable data, NSError *_Nullable error))completion;
 
+/**
+ * Writes whatever fits without blocking, returning the number of bytes taken,
+ * or -1 once this ipc is closed.
+ */
 - (NSInteger)write:(NSData *_Nonnull)data;
 
 - (void)write:(NSData *_Nonnull)data
    completion:(void (^_Nonnull)(NSError *_Nullable error))completion;
 
+/**
+ * Retires this ipc, which must happen before its worklet is destroyed.
+ * `terminate` does it for you, and closing twice is a no-op.
+ */
 - (void)close;
 
 @end

apple/BareKit/BareKit.m

@@ -103,6 +103,9 @@ - (_Nullable instancetype)init {
 @implementation BareWorklet {
 @public
   bare_worklet_t _worklet;
+
+  // Unowned: the ipc retains this worklet and clears this when it closes.
+  BareIPC *_ipc;
 }
 
 + (void)optimizeForMemory:(BOOL)enabled {
@@ -234,6 +237,8 @@ - (void)resume {
 }
 
 - (void)terminate {
+  [_ipc close];
+
   int err;
   err = bare_worklet_terminate(&_worklet);
   assert(err == 0);
@@ -303,6 +308,8 @@ @implementation BareIPC {
 
   bare_ipc_t _ipc;
   bare_ipc_poll_t _poll;
+
+  bool _closed;
 }
 
 - (_Nullable instancetype)initWithWorklet:(BareWorklet *_Nonnull)worklet {
@@ -320,6 +327,8 @@ - (_Nullable instancetype)initWithWorklet:(BareWorklet *_Nonnull)worklet {
     assert(err == 0);
 
     _poll.data = (__bridge void *) self;
+
+    worklet->_ipc = self;
   }
 
   return self;
@@ -347,6 +356,8 @@ - (void)setWritable:(void (^)(BareIPC *_Nonnull))writable {
 }
 
 - (void)update {
+  if (_closed) return;
+
   int events = 0;
 
   if (_readable) events |= bare_ipc_readable;
@@ -369,6 +380,8 @@ - (void)poll:(int)events {
 }
 
 - (NSData *_Nullable)read {
+  if (_closed) return [NSData data];
+
   int err;
 
   void *data;
@@ -384,6 +397,8 @@ - (NSData *_Nullable)read {
 }
 
 - (void)read:(void (^_Nonnull)(NSData *_Nullable data, NSError *_Nullable error))completion {
+  if (_closed) return completion(nil, nil);
+
   int err;
 
   void *data;
@@ -411,6 +426,8 @@ - (void)read:(void (^_Nonnull)(NSData *_Nullable data, NSError *_Nullable error)
 }
 
 - (NSInteger)write:(NSData *_Nonnull)data {
+  if (_closed) return -1;
+
   int err;
   err = bare_ipc_write(&_ipc, data.bytes, data.length);
   assert(err >= 0 || err == bare_ipc_would_block);
@@ -424,6 +441,13 @@ - (NSInteger)write:(NSData *_Nonnull)data {
 
 - (void)write:(NSData *_Nonnull)data
    completion:(void (^_Nonnull)(NSError *_Nullable error))completion {
+  if (_closed) {
+    return completion([NSError
+      errorWithDomain:@"to.holepunch.bare.kit"
+                 code:-1
+             userInfo:@{NSLocalizedDescriptionKey : @"IPC closed"}]);
+  }
+
   NSInteger written = [self write:data];
 
   if (written == data.length) {
@@ -457,11 +481,19 @@ - (void)write:(NSData *_Nonnull)data
 }
 
 - (void)close {
+  if (_closed) return;
+
+  _closed = true;
+
+  _worklet->_ipc = nil;
+
   bare_ipc_poll_destroy(&_poll);
   bare_ipc_destroy(&_ipc);
 }
 
 - (void)dealloc {
+  [self close];
+
   [_worklet release];
 
   [super dealloc];

package.json

@@ -22,11 +22,11 @@
     "bare-crypto": "^1.2.8",
     "bare-events": "^2.4.2",
     "bare-fs": "^4.0.1",
-    "bare-ipc": "^1.0.1",
     "bare-logger": "^2.0.0",
     "bare-module": "^7.0.1",
     "bare-module-resolve": "^1.6.2",
     "bare-path": "^3.0.0",
+    "bare-stream": "^2.13.4",
     "bare-system-logger": "^1.0.1",
     "bare-unpack": "^1.0.2",
     "bare-url": "^2.0.1"

shared/CMakeLists.txt

@@ -57,8 +57,6 @@ if(APPLE)
   target_sources(
     bare_worklet
     PRIVATE
-      posix/ipc.h
-      posix/ipc.c
       apple/ipc.h
       apple/ipc.m
       apple/suspension.h
@@ -90,8 +88,6 @@ if(ANDROID)
   target_sources(
     bare_worklet
     PRIVATE
-      posix/ipc.h
-      posix/ipc.c
       android/ipc.c
       android/ipc.h
       android/suspension.c
@@ -115,10 +111,8 @@ if(LINUX)
   target_sources(
     bare_worklet
     PRIVATE
-      posix/ipc.h
-      posix/ipc.c
-      linux/ipc.c
-      linux/ipc.h
+      uv/ipc.c
+      uv/ipc.h
       linux/suspension.c
       linux/suspension.h
   )
@@ -134,8 +128,8 @@ if(WIN32)
   target_sources(
     bare_worklet
     PRIVATE
-      win32/ipc.c
-      win32/ipc.h
+      uv/ipc.c
+      uv/ipc.h
       win32/suspension.c
       win32/suspension.h
   )