Node.jsにworkerが入った

1
// child_process
2

3
// parent
4
const { fork } = require("child_process");
5

6
const child = fork("child.js");
7

8
child.on("message", (message) => {
9
  console.log("message", message);
10
  child.send("from parent");
11
});
12

13
// child
14
process.on("message", (message) => {
15
  console.log("message", message);
16
});
17

18
if (process.send) process.send("from child");

1
require("worker_threads");
2

3
const { port1, port2 } = new MessageChannel();
4

5
port1.on("message", (message) => console.log(message));
6
port2.postMessage("hi");
7

8
port2.on("message", (message) => console.log(message));
9
port1.postMessage("bye");

1
const {
2
  threadId, isMainThread,
3
  Worker, workerData, parentPort
4
  MessageChannel, MessagePort
5
} = require('worker_threads');

1
const { Worker, isMainThread, workerData } = require("worker_threads");
2

3
let current = 0;
4

5
function counter(title, cnt) {
6
  console.log(`| ${title} |: ${cnt}`);
7
}
8

9
if (isMainThread) {
10
  console.log("Main Thread");
11

12
  // 4つ作成する
13
  for (let i = 0; i < 4; i++) {
14
    // 相対パスはダメ
15
    // この場合(__filename)はworkerも同じこのファイルを参照する
16
    // 第二引数はグローバルパラメーター
17
    new Worker(__filename, { workerData: i });
18
  }
19

20
  setInterval(
21
    (title) => {
22
      counter(title, ++current);
23
    },
24
    1000,
25
    "MainThread",
26
  );
27
} else {
28
  console.log(`worker: ${workerData}`);
29

30
  setInterval(
31
    (title) => {
32
      counter(title, ++current);
33
    },
34
    1000,
35
    `worker: ${workerData}`,
36
  );
37
}

1
worker: 1 worker: 2 worker: 3 | MainThread |: 1 | worker: 0 |: 1 | worker: 1 |:
2
1 | worker: 2 |: 1 | worker: 3 |: 1 | MainThread |: 2 | worker: 0 |: 2 | worker:
3
1 |: 2 | worker: 2 |: 2 | worker: 3 |: 2 | MainThread |: 3 | worker: 0 |: 3 |
4
worker: 1 |: 3 | worker: 2 |: 3 | worker: 3 |: 3 ```
5

6
### スレッド間でメッセージングを行う
7

8
```javascript
9
// parent.js
10
// main thread
11

12
const { resolve } = require("path");
13
const { Worker, workerData } = require("worker_threads");
14

15
console.log("| Main Thread |");
16

17
function createWorker(path, cb) {
18
  const worker = new Worker(path, { workerData: null });
19

20
  worker.on("message", (msg) => {
21
    cb(null, msg);
22
  });
23
  worker.on("error", cb);
24
  worker.on("exit", (code) => {
25
    if (code !== 0) throw new Error("worker stopped");
26
    console.log("| Main Thread | worker stopped");
27
  });
28

29
  return worker;
30
}
31

32
const w = createWorker(resolve("child.js"), (err, res) => {
33
  // workerから結果を受け取る
34
  if (err) {
35
    console.error(err);
36
    process.exit(1);
37
  }
38

39
  console.log(`| Main Thread | execution time: ${res}ms from Worker Thread`);
40

41
  // workerを停止させるためにメッセージを送る
42
  w.postMessage("thx;)");
43
});

1
const { PerformanceObserver, performance } = require("perf_hooks");
2
const { parentPort } = require("worker_threads");
3

4
console.log("| Worker |");
5

6
const obs = new PerformanceObserver((items) => {
7
const time = items.getEntries()[0].duration;
8

9
// main threadへ処理時間の結果を返す
10
parentPort.postMessage(time);
11

12
performance.clearMarks();
13
});
14
obs.observe({ entryTypes: ["measure"] });
15

16
const len = 64 _ 1024 _ 1024;
17
const b = Buffer.allocUnsafe(len);
18
let s = "";
19

20
// 重い処理なためこの箇所の実行時間を計測する
21
performance.mark("A");
22
for (let i = 0; i < 256; ++i) s += String.fromCharCode(i);
23
for (let i = 0; i < 64 _ 1024 _ 1024; i += 256) b.write(s, i, 256, "ascii");
24
for (let i = 0; i < 32; ++i) b.toString("base64");
25
performance.mark("B");
26
performance.measure("A to B", "A", "B");
27

28
// main threadから返答が来たらこのworkerを終了する
29
parentPort.on("message", (msg) => {
30
console.log(`| Worker | ${msg} from Main Thread`);
31
process.exit();
32
});

1
# output
2
$ node --experimental-worker parent.js
3
| Main Thread |
4
| Worker |
5
| Main Thread | execution time: 3128.782205ms from Worker Thread
6
| Worker | thx;) from Main Thread
7
| Main Thread | worker stopped

1
    $ ./node benchmark/cluster/echo.js
2
    cluster/echo.js n=100000 sendsPerBroadcast=1 payload="string" workers=1: 33,647.30473442063
3
    cluster/echo.js n=100000 sendsPerBroadcast=10 payload="string" workers=1: 12,927.907405288383
4
    cluster/echo.js n=100000 sendsPerBroadcast=1 payload="object" workers=1: 28,496.37373941151
5
    cluster/echo.js n=100000 sendsPerBroadcast=10 payload="object" workers=1: 8,975.53747186485
6
    $ ./node --experimental-worker benchmark/worker/echo.js
7
    worker/echo.js n=100000 sendsPerBroadcast=1 payload="string" workers=1: 88,044.32902365089
8
    worker/echo.js n=100000 sendsPerBroadcast=10 payload="string" workers=1: 39,873.33697018837
9
    worker/echo.js n=100000 sendsPerBroadcast=1 payload="object" workers=1: 64,451.29132425621
10
    worker/echo.js n=100000 sendsPerBroadcast=10 payload="object" workers=1: 22,325.635443739284