Skip to content
Closed
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 39 additions & 28 deletions src/runtime/ipc.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1539,34 +1539,39 @@ impl SendQueue {
let done = self.queue.with_mut(|queue| {
let first = &mut queue[0];
let to_send_len = first.data.list.len() - first.data.cursor;
if n as usize == to_send_len {
if first.handle.is_some() {
// the message was fully written, but it had a handle.
// we must wait for ACK or NACK before sending any more messages.
let item = queue.remove(0);
self.waiting_for_ack.with_mut(|w| {
if w.is_some() {
log!("[error] already waiting for ack. this should never happen.");
}
// shift the item off the queue and move it to waiting_for_ack
*w = Some(item);
});
Done::AwaitAck
} else {
// the message was fully sent, but there may be more items in the queue.
// shift the queue and try to send the next item immediately.
Done::Completed(queue.remove(0))
// `n` is the write count from the socket (at most i32::MAX per write).
// The coalesced item can be larger than that, so compare as usize.
Comment thread
robobun marked this conversation as resolved.
Resolved
match usize::try_from(n) {
Ok(written) if written == to_send_len => {
if first.handle.is_some() {
// the message was fully written, but it had a handle.
// we must wait for ACK or NACK before sending any more messages.
Comment thread
robobun marked this conversation as resolved.
Resolved
let item = queue.remove(0);
self.waiting_for_ack.with_mut(|w| {
if w.is_some() {
log!("[error] already waiting for ack. this should never happen.");
}
// shift the item off the queue and move it to waiting_for_ack
*w = Some(item);
});
Done::AwaitAck
} else {
// the message was fully sent, but there may be more items in the queue.
// shift the queue and try to send the next item immediately.
Comment thread
robobun marked this conversation as resolved.
Resolved
Done::Completed(queue.remove(0))
}
}
} else if n > 0 && n < i32::try_from(first.data.list.len()).expect("int cast") {
// the item was partially sent; update the cursor and wait for writable to send the rest
// (if we tried to send a handle, a partial write means the handle wasn't sent yet.)
first.data.cursor += usize::try_from(n).expect("int cast");
Done::Partial
} else if n == 0 {
// no bytes written; wait for writable
Done::NoProgress
} else {
Done::Error
Ok(0) => {
// no bytes written; wait for writable
Done::NoProgress
}
Ok(written) if written < to_send_len => {
// the item was partially sent; update the cursor and wait for writable to send the rest
// (if we tried to send a handle, a partial write means the handle wasn't sent yet.)
Comment thread
robobun marked this conversation as resolved.
Resolved
first.data.cursor += written;
Done::Partial
}
_ => Done::Error,
}
});
match done {
Expand All @@ -1577,7 +1582,13 @@ impl SendQueue {
item.complete(&global_this); // call the callback & deinit
self.continue_send(&global_this, ContinueSendReason::OnWritable);
}
Done::Partial | Done::NoProgress => {}
Done::Partial => {
// libuv completes a request in full or fails. A partial write is
// the i32::MAX cap in `write`, and no writable event follows it.
Comment thread
robobun marked this conversation as resolved.
Resolved
#[cfg(windows)]
self.continue_send(&global_this, ContinueSendReason::OnWritable);
Comment thread
robobun marked this conversation as resolved.
Resolved
}
Comment thread
robobun marked this conversation as resolved.
Resolved
Done::NoProgress => {}
Done::Error => {
// error. close socket.
self.close_socket(CloseReason::Failure, CloseFrom::User);
Expand Down
48 changes: 48 additions & 0 deletions test/js/bun/spawn/spawn.ipc.test.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import { spawn } from "bun";
import { describe, expect, it } from "bun:test";
import { bunEnv, bunExe, gcTick, isWindows } from "harness";
import { totalmem } from "node:os";
import path from "path";

describe.each(["advanced", "json"])("ipc mode %s", mode => {
Expand Down Expand Up @@ -295,3 +296,50 @@ it.skipIf(isWindows)("advanced serialization advertises wire format version 2",
expect(JSON.parse(stdout.trim())).toEqual([1, 2, 0, 0, 0]);
expect(exitCode).toBe(0);
});

// The send queue coalesces every message queued behind a partial write into one
// buffer. When the peer has not read yet, that buffer can pass 2 GiB. The write
// completion compared the socket's i32 write count against that length as an i32
// and panicked with "int cast: TryFromIntError(PosOverflow)". The child peaks at
// 4 to 6 GB of RSS (the 2.2 GB buffer, its last capacity doubling, freed payloads).
// Inside a container os.totalmem() reports the host's RAM;
// process.constrainedMemory() reports the cgroup limit there.
const memory = Math.min(totalmem(), process.constrainedMemory() || Infinity);
describe.skipIf(isWindows || memory < 16 * 1024 ** 3)("send queue past 2 GiB", () => {
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
// Debug and ASAN builds copy the 2.2 GB about four times before the first message.
const timeout = 60_000;
it(
"keeps sending after more than 2 GiB is queued before the first writable event",
async () => {
const chunkLength = 100 * 1024 * 1024;
// 22 chunks = 2200 MiB, all queued in one synchronous tick, so the head item
// passes i32::MAX bytes before the event loop sees the socket writable.
const childSource = `
const chunk = Buffer.alloc(${chunkLength}, "x").toString();
for (let i = 0; i < 22; i++) process.send(chunk);
process.on("message", () => process.exit(0));
`;
const { promise, resolve, reject } = Promise.withResolvers<number>();
let received = 0;
await using child = spawn([bunExe(), "-e", childSource], {
env: bunEnv,
stdio: ["ignore", "inherit", "inherit"],
serialization: "advanced",
ipc(message, subprocess) {
// The first complete message proves the child kept writing past the cap.
// Stop the child there instead of draining the whole queue.
if (received++ === 0) {
resolve(message.length);
subprocess.send("stop");
}
},
onExit(_subprocess, exitCode, signalCode) {
reject(new Error(`child exited (${exitCode}, ${signalCode}) before the first message`));
},
});
expect(await promise).toBe(chunkLength);
expect(await child.exited).toBe(0);
},
timeout,
Comment thread
robobun marked this conversation as resolved.
Resolved
);
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
});
Loading