javascript - 强制~同步 Node.js IPC

Question

我有一个 Node 服务器，它使用 IPC 创建一个带有 fork() 的子进程。在某些时候， child 会以大约 10Hz 的频率将结果发送回 parent ，作为长期运行任务的一部分。当传递给 process.send() 的有效负载很小时，一切正常:我发送的每条消息都会立即收到并由父级处理。

但是，当有效负载“大”时——我还没有确定确切的大小限制——而不是立即被父级接收所有有效负载首先发送，并且只有在子级完成其长时间运行的任务后，父级才会接收并处理消息。

tl；视觉博士:

好 (发生在小负载上):

child:  send()
parent: receive()
child:  send()
parent: receive()
child:  send()
parent: receive()
...

错误 (发生在大负载时):

child:  send()
child:  send()
child:  send()
(repeat many times over many seconds)
...
parent: receive()
parent: receive()
parent: receive()
parent: receive()
...

这是一个错误吗？ (编辑:行为只发生在 OS X，而不是 Windows 或 Linux)

有什么办法可以避免这种情况，除了尽量保持我的 IPC 有效载荷小吗？

编辑 2 :下面的示例代码使用时间和迭代计数器来选择何时发送更新。 (在我的实际代码中，也可以在 n 次迭代或循环达到某些结果之后发送更新。)因此，重写代码以使用 setInterval/ setTimeout 而不是循环对我来说是最后的手段，因为它要求我删除功能。

编辑 :这是重现问题的测试代码。但是，它只能在 OS X 上重现，不能在 Windows 或 Linux 上重现:

server.js

const opts = {stdio:['inherit', 'inherit', 'inherit', 'ipc']};
const child = require('child_process').fork('worker.js', [], opts);

child.on('message', msg => console.log(`parent: receive() ${msg.data.length} bytes`, Date.now()));

require('http').createServer((req, res) => {
   console.log(req.url);
   const match = /\d+/.exec(req.url);
   if (match) {
      child.send(match[0]*1);
      res.writeHead(200, {'Content-Type':'text/plain'});
      res.end(`Sending packets of size ${match[0]}`);
   } else {
      res.writeHead(404, {'Content-Type':'text/plain'});
      res.end('what?');
   }
}).listen(8080);

worker.js

if (process.send) process.on('message', msg => run(msg));

function run(messageSize) {
   const msg = new Array(messageSize+1).join('x');
   let lastUpdate = Date.now();
   for (let i=0; i<1e7; ++i) {
      const now = Date.now();
      if ((now-lastUpdate)>200 || i%5000==0) {
         console.log(`worker: send()  > ${messageSize} bytes`, now);
         process.send({action:'update', data:msg});
         lastUpdate = Date.now();
      }
      Math.sqrt(Math.random());
   }
   console.log('worker done');
}

大约 8k 左右出现问题。例如查询 http://localhost:8080/15 vs http://localhost:8080/123456

/15
worker: send()  > 15 bytes 1571324249029
parent: receive() 15 bytes 1571324249034
worker: send()  > 15 bytes 1571324249235
parent: receive() 15 bytes 1571324249235
worker: send()  > 15 bytes 1571324249436
parent: receive() 15 bytes 1571324249436
worker done
/123456
worker: send()  > 123456 bytes 1571324276973
worker: send()  > 123456 bytes 1571324277174
worker: send()  > 123456 bytes 1571324277375
child done
parent: receive() 123456 bytes 1571324277391
parent: receive() 123456 bytes 1571324277391
parent: receive() 123456 bytes 1571324277393

在 Node v12.7 和 v12.12 上都有经验。

Answer 1

l 在 Node 中与套接字或文件描述符相结合的长时间运行和阻塞的while循环总是表明某些事情做错了。

在无法测试整个设置的情况下，很难判断我的声明是否真的正确，但短消息可能会直接以一个 block 的形式传递给操作系统，然后再传递给另一个进程。对于较大的消息 Node 需要等到操作系统可以接收更多数据，所以发送排队，并且因为你有一个阻塞 while发送是队列直到 while loop结束了。

所以对于你的问题，这不是一个错误。

当您使用最新的 nodejs 版本时，我会使用 await和 async而不是并创建一个非阻塞 while类似于 sleep在 this answer . await如果 processSome 将允许 Node 事件循环拦截返回待处理的 Promise。

对于没有真正反射(reflect)真实用例的代码，很难说出如何正确解决它。如果你在 processSome 中不做任何异步操作这将允许 I/O 拦截然后您需要定期手动执行此操作，例如一个 await new Promise(setImmediate); .

async function run() {
  let interval = setInterval(() => {
    process.send({action:'update', data:status()});
    console.log('child:  send()');
  }, 1/10)

  while(keepGoing()) {
    await processSome();
  }

  clearInterval(interval)
}