fix: implement child_process IPC

cli/args/mod.rs

@@ -881,6 +881,17 @@ impl CliOptions {
     .map(Some)
   }
 
+  pub fn node_ipc_fd(&self) -> Option<i32> {
+    let maybe_node_channel_fd = std::env::var("DENO_CHANNEL_FD").ok();
+    if let Some(node_channel_fd) = maybe_node_channel_fd {
+      // Remove so that child processes don't inherit this environment variable.
+      std::env::remove_var("DENO_CHANNEL_FD");
+      node_channel_fd.parse::<i32>().ok()
+    } else {
+      None
+    }
+  }
+
   pub fn resolve_main_module(&self) -> Result<ModuleSpecifier, AnyError> {
     match &self.flags.subcommand {
       DenoSubcommand::Bundle(bundle_flags) => {

cli/factory.rs

@@ -661,6 +661,7 @@ impl CliFactory {
       self.maybe_lockfile().clone(),
       self.feature_checker().clone(),
       self.create_cli_main_worker_options()?,
+      self.options.node_ipc_fd(),
     ))
   }
 

cli/standalone/mod.rs

@@ -477,6 +477,7 @@ pub async fn run(
       unstable: metadata.unstable,
       maybe_root_package_json_deps: package_json_deps_provider.deps().cloned(),
     },
+    None,
   );
 
   v8_set_flags(construct_v8_flags(&[], &metadata.v8_flags, vec![]));

cli/worker.rs

@@ -125,6 +125,7 @@ struct SharedWorkerState {
   maybe_inspector_server: Option<Arc<InspectorServer>>,
   maybe_lockfile: Option<Arc<Mutex<Lockfile>>>,
   feature_checker: Arc<FeatureChecker>,
+  node_ipc: Option<i32>,
 }
 
 impl SharedWorkerState {
@@ -416,6 +417,7 @@ impl CliMainWorkerFactory {
     maybe_lockfile: Option<Arc<Mutex<Lockfile>>>,
     feature_checker: Arc<FeatureChecker>,
     options: CliMainWorkerOptions,
+    node_ipc: Option<i32>,
   ) -> Self {
     Self {
       shared: Arc::new(SharedWorkerState {
@@ -436,6 +438,7 @@ impl CliMainWorkerFactory {
         maybe_inspector_server,
         maybe_lockfile,
         feature_checker,
+        node_ipc,
       }),
     }
   }
@@ -599,6 +602,7 @@ impl CliMainWorkerFactory {
           .options
           .maybe_binary_npm_command_name
           .clone(),
+        node_ipc_fd: shared.node_ipc,
       },
       extensions,
       startup_snapshot: crate::js::deno_isolate_init(),
@@ -793,6 +797,7 @@ fn create_web_worker_callback(
           .options
           .maybe_binary_npm_command_name
           .clone(),
+        node_ipc_fd: None,
       },
       extensions,
       startup_snapshot: crate::js::deno_isolate_init(),

ext/node/Cargo.toml

@@ -44,6 +44,7 @@ libc.workspace = true
 libz-sys.workspace = true
 md-5 = "0.10.5"
 md4 = "0.10.2"
+nix.workspace = true
 num-bigint.workspace = true
 num-bigint-dig = "0.8.2"
 num-integer = "0.1.45"

ext/node/lib.rs

@@ -297,6 +297,9 @@ deno_core::extension!(deno_node,
     ops::require::op_require_break_on_next_statement,
     ops::util::op_node_guess_handle_type,
     ops::crypto::op_node_create_private_key,
+    ops::ipc::op_node_ipc_pipe,
+    ops::ipc::op_node_ipc_write,
+    ops::ipc::op_node_ipc_read,
   ],
   esm_entry_point = "ext:deno_node/02_init.js",
   esm = [

ext/node/ops/ipc.rs

@@ -0,0 +1,179 @@
+// Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
+
+use std::cell::RefCell;
+use std::io;
+use std::os::fd::FromRawFd;
+use std::os::fd::RawFd;
+use std::rc::Rc;
+
+use deno_core::error::bad_resource_id;
+use deno_core::error::AnyError;
+use deno_core::op2;
+use deno_core::serde_json;
+use deno_core::CancelFuture;
+use deno_core::CancelHandle;
+use deno_core::OpState;
+use deno_core::RcRef;
+use deno_core::ResourceId;
+use tokio::net::UnixStream;
+
+struct IpcPipe {
+  // Better name?
+  inner: UnixStream,
+}
+
+impl IpcPipe {
+  fn new(fd: RawFd) -> Result<Self, std::io::Error> {
+    Ok(Self {
+      inner: UnixStream::from_std(unsafe {
+        std::os::unix::net::UnixStream::from_raw_fd(fd)
+      })?,
+    })
+  }
+
+  async fn write(self: Rc<Self>, data: &[u8]) -> Result<usize, std::io::Error> {
+    let mut offset = 0;
+    loop {
+      self.inner.writable().await?;
+      match self.inner.try_write(&data[offset..]) {
+        Ok(n) => {
+          offset += n;
+          if offset >= data.len() {
+            return Ok(offset);
+          }
+        }
+        Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+          continue;
+        }
+        Err(e) => return Err(e),
+      }
+    }
+  }
+
+  async fn read(&self, data: &mut [u8]) -> Result<usize, std::io::Error> {
+    loop {
+      self.inner.readable().await?;
+      match self.inner.try_read(&mut data[..]) {
+        Ok(n) => {
+          return Ok(n);
+        }
+        Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
+          continue;
+        }
+        Err(e) => return Err(e),
+      }
+    }
+  }
+}
+
+struct IpcJsonStream {
+  pipe: Rc<IpcPipe>,
+  buffer: RefCell<Vec<u8>>,
+  cancel: CancelHandle,
+}
+
+impl IpcJsonStream {
+  fn new(fd: RawFd) -> Result<Self, std::io::Error> {
+    Ok(Self {
+      pipe: Rc::new(IpcPipe::new(fd)?),
+      buffer: RefCell::new(Vec::new()),
+      cancel: CancelHandle::default(),
+    })
+  }
+
+  async fn read(self: Rc<Self>) -> Result<Vec<serde_json::Value>, AnyError> {
+    let mut buf = [0u8; 1024]; // TODO: Use a single growable buffer.
+    let mut msgs = Vec::new();
+    loop {
+      let n = self.pipe.read(&mut buf).await?;
+
+      let read = &buf[..n];
+      let mut chunk_boundary = 0;
+
+      for byte in read {
+        if *byte == b'\n' {
+          let chunk = &read[..chunk_boundary];
+          self.buffer.borrow_mut().extend_from_slice(chunk);
+
+          chunk_boundary = 0;
+          if chunk.is_empty() {
+            // Last chunk.
+            break;
+          }
+          msgs.push(serde_json::from_slice(&self.buffer.borrow())?);
+          self.buffer.borrow_mut().clear();
+        } else {
+          chunk_boundary += 1;
+        }
+      }
+
+      if chunk_boundary > 0 {
+        let buffer = &mut self.buffer.borrow_mut();
+        buffer.clear();
+        buffer.extend_from_slice(&read[..chunk_boundary]);
+      }
+
+      if !msgs.is_empty() {
+        return Ok(msgs);
+      }
+    }
+  }
+
+  async fn write(
+    self: Rc<Self>,
+    msg: serde_json::Value,
+  ) -> Result<(), AnyError> {
+    let mut buf = Vec::new();
+    serde_json::to_writer(&mut buf, &msg)?;
+    buf.push(b'\n');
+    self.pipe.clone().write(&buf).await?;
+    Ok(())
+  }
+}
+
+impl deno_core::Resource for IpcJsonStream {
+  fn close(self: Rc<Self>) {
+    self.cancel.cancel();
+  }
+}
+
+#[op2(fast)]
+#[smi]
+pub fn op_node_ipc_pipe(
+  state: &mut OpState,
+  #[smi] fd: i32,
+) -> Result<ResourceId, AnyError> {
+  Ok(state.resource_table.add(IpcJsonStream::new(fd)?))
+}
+
+#[op2(async)]
+pub async fn op_node_ipc_write(
+  state: Rc<RefCell<OpState>>,
+  #[smi] rid: ResourceId,
+  #[serde] value: serde_json::Value,
+) -> Result<(), AnyError> {
+  let stream = state
+    .borrow()
+    .resource_table
+    .get::<IpcJsonStream>(rid)
+    .map_err(|_| bad_resource_id())?;
+  stream.write(value).await?;
+  Ok(())
+}
+
+#[op2(async)]
+#[serde]
+pub async fn op_node_ipc_read(
+  state: Rc<RefCell<OpState>>,
+  #[smi] rid: ResourceId,
+) -> Result<Vec<serde_json::Value>, AnyError> {
+  let stream = state
+    .borrow()
+    .resource_table
+    .get::<IpcJsonStream>(rid)
+    .map_err(|_| bad_resource_id())?;
+
+  let cancel = RcRef::map(stream.clone(), |r| &r.cancel);
+  let msgs = stream.read().or_cancel(cancel).await??;
+  Ok(msgs)
+}

ext/node/ops/mod.rs

@@ -4,6 +4,7 @@ pub mod crypto;
 pub mod http;
 pub mod http2;
 pub mod idna;
+pub mod ipc;
 pub mod os;
 pub mod require;
 pub mod util;

ext/node/polyfills/02_init.js

@@ -7,15 +7,12 @@ const requireImpl = internals.requireImpl;
 import { nodeGlobals } from "ext:deno_node/00_globals.js";
 import "node:module";
 
-globalThis.nodeBootstrap = function (usesLocalNodeModulesDir, argv0) {
-  initialize(usesLocalNodeModulesDir, argv0);
-};
-
 let initialized = false;
 
 function initialize(
   usesLocalNodeModulesDir,
   argv0,
+  ipcFd,
 ) {
   if (initialized) {
     throw Error("Node runtime already initialized");
@@ -41,6 +38,7 @@ function initialize(
   // but it's the only way to get `args` and `version` and this point.
   internals.__bootstrapNodeProcess(argv0, Deno.args, Deno.version);
   internals.__initWorkerThreads();
+  internals.__setupChildProcessIpcChannel(ipcFd);
   // `Deno[Deno.internal].requireImpl` will be unreachable after this line.
   delete internals.requireImpl;
 }
@@ -52,6 +50,8 @@ function loadCjsModule(moduleName, isMain, inspectBrk) {
   requireImpl.Module._load(moduleName, null, { main: isMain });
 }
 
+globalThis.nodeBootstrap = initialize;
+
 internals.node = {
   initialize,
   loadCjsModule,

ext/node/polyfills/child_process.ts

@@ -10,6 +10,7 @@ import {
   ChildProcess,
   ChildProcessOptions,
   normalizeSpawnArguments,
+  setupChannel,
   type SpawnOptions,
   spawnSync as _spawnSync,
   type SpawnSyncOptions,
@@ -821,6 +822,14 @@ export function execFileSync(
   return ret.stdout as string | Buffer;
 }
 
+function setupChildProcessIpcChannel(fd: number) {
+  if (typeof fd != "number" || fd < 0) return;
+  setupChannel(process, fd);
+}
+
+globalThis.__bootstrap.internals.__setupChildProcessIpcChannel =
+  setupChildProcessIpcChannel;
+
 export default {
   fork,
   spawn,