From fad0229d2ee5615fab1cabe0bce1be2300284418 Mon Sep 17 00:00:00 2001 From: oxura Date: Fri, 23 Oct 2026 12:49:29 +0600 Subject: [PATCH 1/2] fix(process): use one subprocess worker on Bun and Node --- bend2/base.bend | 2 + bend2/effs/process_run.js | 83 ++++++++++++++++++++++----------------- tests/io/process_run.bend | 15 +++++++ 3 files changed, 64 insertions(+), 36 deletions(-) diff --git a/bend2/base.bend b/bend2/base.bend index 88f0d781a9..e7e9f18eab 100644 --- a/bend2/base.bend +++ b/bend2/base.bend @@ -201,6 +201,8 @@ def IO.args_bytes() -> IO(List>): import "./effs/args.c" import "./effs/args.js" +# Runs an argv subprocess on C, Bun and Node, without a shell. +# max_output bounds combined stdout/stderr bytes; timeout_ms includes pipe draining. def Process.run(program: String, args: List, input: String, max_output: U32, timeout_ms: U32) -> IO(Result<&1, &1, U32 & String, U32 & (String & String)>): diff --git a/bend2/effs/process_run.js b/bend2/effs/process_run.js index 50f9ef20e9..14e5bd3f73 100644 --- a/bend2/effs/process_run.js +++ b/bend2/effs/process_run.js @@ -3,10 +3,12 @@ const PROCESS_WORKER = ` const { errno, signals } = require("node:os").constants; +const { parentPort } = require("node:worker_threads"); +const { spawn } = require("node:child_process"); let port; let flag; -self.onmessage = async ({ data }) => { +parentPort.on("message", async (data) => { if (data.port !== undefined) { ({ port, flag } = data); return; @@ -21,40 +23,49 @@ self.onmessage = async ({ data }) => { port.postMessage(got); Atomics.store(flag, 0, 1); Atomics.notify(flag, 0); -}; +}); -async function run({ argv, input, env, max, ms }) { - const proc = Bun.spawn({ cmd: argv, stdin: input, env, stdout: "pipe", - stderr: "pipe" }); - const readers = [proc.stdout.getReader(), proc.stderr.getReader()]; - let size = 0; - const read = async (reader) => { - const chunks = []; - for (let r; !(r = await reader.read()).done;) { - chunks.push(r.value); - size += r.value.length; - if (size > max) { - throw "EFBIG"; - } - } - return Buffer.concat(chunks).toString("utf8"); - }; - let timer; - try { - const [out, err] = await Promise.race([ - Promise.all([...readers.map(read), proc.exited]), - new Promise((_, no) => timer = setTimeout(no, Math.min(ms, 2 ** 31 - 1), - "ETIMEDOUT"))]); - const sig = proc.signalCode === null ? 0 : signals[proc.signalCode] ?? 0; - return { code: proc.exitCode ?? 128 + sig, out, err }; - } catch (e) { - proc.kill("SIGKILL"); - readers.forEach((r) => r.cancel().catch(() => {})); - await proc.exited; - throw e; - } finally { - clearTimeout(timer); - } +function run({ argv, input, env, max, ms }) { + return new Promise((yes, no) => { + const proc = spawn(argv[0], argv.slice(1), { env, + stdio: ["pipe", "pipe", "pipe"] }); + const chunks = [[], []]; + let size = 0; + let error; + const fail = (e) => { + if (error !== undefined) return; + error = e; + proc.kill("SIGKILL"); + proc.stdin.destroy(); + proc.stdout.destroy(); + proc.stderr.destroy(); + }; + proc.once("error", fail); + proc.stdin.on("error", (e) => { + if (e.code !== "EPIPE") fail(e); + }); + [proc.stdout, proc.stderr].forEach((stream, i) => { + stream.on("error", fail); + stream.on("data", (chunk) => { + if (error !== undefined) return; + size += chunk.length; + if (size > max) fail("EFBIG"); + else chunks[i].push(chunk); + }); + }); + // An exited child need not drain stdin held open by its descendants. + proc.once("exit", () => proc.stdin.destroy()); + const timer = setTimeout(() => fail("ETIMEDOUT"), + Math.min(ms, 2 ** 31 - 1)); + proc.once("close", (code, signal) => { + clearTimeout(timer); + if (error !== undefined) no(error); + else yes({ code: code ?? 128 + (signals[signal] ?? 0), + out: Buffer.concat(chunks[0]).toString("utf8"), + err: Buffer.concat(chunks[1]).toString("utf8") }); + }); + proc.stdin.end(input); + }); } `; @@ -70,10 +81,10 @@ function process_run(program, args, input, maxOutput, timeoutMs) { return io_fail(22); } if (process_worker === null) { - const { MessageChannel, receiveMessageOnPort } = require("node:worker_threads"); + const { Worker, MessageChannel, receiveMessageOnPort } = require("node:worker_threads"); const { port1, port2 } = new MessageChannel(); const flag = new Int32Array(new SharedArrayBuffer(4)); - const worker = new Worker(URL.createObjectURL(new Blob([PROCESS_WORKER]))); + const worker = new Worker(PROCESS_WORKER, { eval: true }); worker.unref(); worker.postMessage({ port: port2, flag }, [port2]); process_worker = { worker, port: port1, flag, receive: receiveMessageOnPort }; diff --git a/tests/io/process_run.bend b/tests/io/process_run.bend index 76113c4520..f649c1fb4e 100644 --- a/tests/io/process_run.bend +++ b/tests/io/process_run.bend @@ -91,6 +91,18 @@ def main() -> IO(Unit): r1 : Result<&1, &1, U32 & String, U32 & (String & String)> <- Process.run("/bin/echo", ["hi"], "", 128, 4294967295) IO.print("largest timeout: " ++ answer(r1, 0, "hi\n", "")) + partial : Result<&1, &1, U32 & String, U32 & (String & String)> <- + Process.run("/usr/bin/head", ["-c", "3"], + String.repeat(String.repeat("z", 128n), 1000n), 128, 3000) + IO.print("partial stdin: " ++ answer(partial, 0, "zzz", "")) + unread : Result<&1, &1, U32 & String, U32 & (String & String)> <- + Process.run("/usr/bin/true", [], + String.repeat(String.repeat("z", 128n), 1000n), 128, 3000) + IO.print("unread stdin: " ++ answer(unread, 0, "", "")) + held : Result<&1, &1, U32 & String, U32 & (String & String)> <- + Process.run("/bin/sh", ["-c", "sleep 2 <&0 >/dev/null 2>&1 & exit 4"], + String.repeat(String.repeat("z", 256n), 1200n), 128, 500) + IO.print("descendant stdin: " ++ answer(held, 4, "", "")) #|stdin: True #|literal argv: True @@ -111,3 +123,6 @@ def main() -> IO(Unit): #|runaway descendant: True #|closed stdio: True #|largest timeout: True +#|partial stdin: True +#|unread stdin: True +#|descendant stdin: True From ff29e9d94d6086642635f2e06944b0dfe45a6369 Mon Sep 17 00:00:00 2001 From: Lorenzobattistela Date: Fri, 23 Oct 2026 07:13:45 -0300 Subject: [PATCH 2/2] fix(process): keep feeding stdin to descendants on Node Node's ChildProcess destroys proc.stdin when the child exits, so a descendant that inherited stdin got at most one pipe buffer of input. Hold stdin outside the ChildProcess and close it once the child has exited and stdout and stderr have closed, as the C lane does. Co-Authored-By: Claude Opus 5.5 (1M context) --- bend2/effs/process_run.js | 17 ++++++++++++----- tests/io/process_run.bend | 6 ++++++ 2 files changed, 18 insertions(+), 5 deletions(-) diff --git a/bend2/effs/process_run.js b/bend2/effs/process_run.js index 14e5bd3f73..6b7d34d72f 100644 --- a/bend2/effs/process_run.js +++ b/bend2/effs/process_run.js @@ -29,6 +29,9 @@ function run({ argv, input, env, max, ms }) { return new Promise((yes, no) => { const proc = spawn(argv[0], argv.slice(1), { env, stdio: ["pipe", "pipe", "pipe"] }); + // Node destroys proc.stdin at exit; a descendant may still read it. + const stdin = proc.stdin; + proc.stdin = null; const chunks = [[], []]; let size = 0; let error; @@ -36,12 +39,12 @@ function run({ argv, input, env, max, ms }) { if (error !== undefined) return; error = e; proc.kill("SIGKILL"); - proc.stdin.destroy(); + stdin.destroy(); proc.stdout.destroy(); proc.stderr.destroy(); }; proc.once("error", fail); - proc.stdin.on("error", (e) => { + stdin.on("error", (e) => { if (e.code !== "EPIPE") fail(e); }); [proc.stdout, proc.stderr].forEach((stream, i) => { @@ -53,8 +56,12 @@ function run({ argv, input, env, max, ms }) { else chunks[i].push(chunk); }); }); - // An exited child need not drain stdin held open by its descendants. - proc.once("exit", () => proc.stdin.destroy()); + // Feed stdin until the child exits and its stdout and stderr close. + let live = 3; + const done = () => --live || stdin.destroy(); + proc.once("exit", done); + proc.stdout.once("close", done); + proc.stderr.once("close", done); const timer = setTimeout(() => fail("ETIMEDOUT"), Math.min(ms, 2 ** 31 - 1)); proc.once("close", (code, signal) => { @@ -64,7 +71,7 @@ function run({ argv, input, env, max, ms }) { out: Buffer.concat(chunks[0]).toString("utf8"), err: Buffer.concat(chunks[1]).toString("utf8") }); }); - proc.stdin.end(input); + stdin.end(input); }); } `; diff --git a/tests/io/process_run.bend b/tests/io/process_run.bend index f649c1fb4e..c06e6a7fe8 100644 --- a/tests/io/process_run.bend +++ b/tests/io/process_run.bend @@ -103,6 +103,11 @@ def main() -> IO(Unit): Process.run("/bin/sh", ["-c", "sleep 2 <&0 >/dev/null 2>&1 & exit 4"], String.repeat(String.repeat("z", 256n), 1200n), 128, 500) IO.print("descendant stdin: " ++ answer(held, 4, "", "")) + late : Result<&1, &1, U32 & String, U32 & (String & String)> <- + Process.run("/bin/sh", ["-c", "(sleep 0.2; cat) <&0 & exit 0"], + String.repeat(String.repeat("z", 128n), 1000n), 200000, 3000) + IO.print("late stdin reader: " ++ answer(late, 0, + String.repeat(String.repeat("z", 128n), 1000n), "")) #|stdin: True #|literal argv: True @@ -126,3 +131,4 @@ def main() -> IO(Unit): #|partial stdin: True #|unread stdin: True #|descendant stdin: True +#|late stdin reader: True