how to handle backpressure on child_process.fork when sending a lot of messages

[MichaelBelousov]

env: node v18.5.0 Ubuntu 22.04

I have a node.js program that uses child_process.fork to create a worker, and then sends it a lot of messages as it performs work. I use a separate process because I am using a synchronous javascript API for dealing with a binary format, to read from one file and then the same API to write to another file. So the program would be much faster if every time the analysis on the read side was ready, it pushed a message of what to write to the write side, and then continued processing data from the read side rather than waiting for the writing to be done.

But I seem to be running up against backpressure in the ipc between a node.js parent and forked child. Following is a minimal viable reproduction, where after a certain amount of messages, the child seems to stop receiving message events. In fact I can add a signal handler like process.on("SIGUSR2", () => { debugger; }, then kill -s SIGUSR2 pid, and attach with devtools to both parent and child, and manually run process.send and its equivalent on the parent, but breakpoints in the child's on("message") handler will not be hit. Even attaching with strace --attach, I don't see a difference in the child's recvmsg syscall usage when sending a message manually, but the handler isn't called.

I can't seem to myself find other people dealing seeing this issue, are there any resources for handling backpressure in a child_process.fork'ed node.js process? I see some for streams, but I don't think it applies.

parent.js

const child_process = require("child_process");
const crypto = require("crypto");

async function main() {
  const child = child_process.fork(
    "./child.js",
    [],
    {
      execArgv: [
        process.env.INSPECT_CHILD && `--inspect=${process.env.INSPECT_CHILD}`,
        process.env.INSPECT_BRK_CHILD && `--inspect=${process.env.INSPECT_BRK_CHILD}`,
      ].filter(Boolean),
      serialization: 'advanced',
    }
  );

  child.on("disconnect", () => console.log(`parent sees disconnect from child`));
  child.on("close", () => console.log(`parent sees close from child`));
  child.on("error", (err) => console.log(`parent sees error from child: ${err}`));
  child.on("exit", (code, signal) => console.log(`parent sees child exiting with ${code ?? signal}`));
  child.on("message", (message) => console.error(`parent received ${message}`));

  for (let i = 0; i < 100_000; ++i) {
    const data = crypto.pseudoRandomBytes(10_000);
    const shouldContinue = child.send({ id: i, data });
    if (!shouldContinue) console.log(`parent got shouldn't continue for ${i}`);
    console.error(`parent sending ${i}`);
    await new Promise(r => process.nextTick(r));
  }
  console.log("parent sent all");
  //child.disconnect();
}

if (module === require.main)
  main().catch(console.error);

process.on("SIGTERM", (code) => console.log(`parent got ${SIGTERM}`));
process.on("exit", (code) => console.log(`parent exiting with ${code}`));

child.js

async function main() {
  process.on("SIGTERM", () => console.log(`child got SIGTERM`));
  process.on("exit", (code) => console.log(`child exiting with ${code}`));
  process.on("disconnect", () => console.log("parent disconnected"));

  process.on("message", (message) => {
    console.error("child received", message.id);
  })
}

Note, in my larger project that uses this architecture, it stops sending messages to the child for 200ms every time it encounters a false result from .send, but that seems to not help, even if I wait much longer.

[MichaelBelousov]

related? nodejs/node#7657

[MichaelBelousov]

I have elected to just manually set up ipc, and use the v8 serialization api via require("v8").[de]serialize to send messages with array buffers between my processes. This way I can use the readable interface on the ipc primitives in the node:net module (or just process.stdin/out) and handle back pressure. It also continues to work even when I don't handle backpressure (although ofc the performance gets bad on large data if I don't)