ハンズオンNode.jsを進める【第4章 マルチプロセス、マルチスレッド】

sysctl -n hw.ncpu

Completed requests:  1084
Total errors:        10
Total time:          10.006 s
Mean latency:        2820.5 ms
Effective rps:       108

Completed requests:  3972
Total errors:        0
Total time:          10.01 s
Mean latency:        944.9 ms
Effective rps:       397

node multi-process.js
メインプロセス 3123
/Users/yuuki_haga/repos/node/node-practice/chapter_4/src/multi-process
{ cpuCount: 8 }
サブプロセス 3137
サブプロセス 3138
サブプロセス 3139
サブプロセス 3140
サブプロセス 3141
サブプロセス 3142
サブプロセス 3143
サブプロセス 3144

// サブプロセスが実行するファイルの指定
setupMaster({
  exec: `${__dirname}/web-app`,
  //  IPCでやりとりするメッセージをJSON形式にシリアライズするわけではなく、構造化クローンアルゴリズムというアルゴリズムでシリアライズするようにする
  serialization: "advanced",
})

"use strict"

const { Worker, threadId } = require("worker_threads")

console.log("メインスレッド", threadId);

// CPUコアの数だけスレッドを起動
const cpuCount = require("os").cpus().length

for(let i = 0; i < cpuCount; i++) {
  // サブスレッドで実行するファイルのパスを指定してWorkerをnew
  // Workerインスタンスを生成すると、指定したファイルパスのファイルが別スレッドで実行される
  // このWorkerはcluster.Workerとは別物
  const worker = new Worker(`${__dirname}/web-app.js`)
  console.log("サブスレッド", worker.threadId);
}

node multi-thread.js
メインスレッド 0
サブスレッド 1
サブスレッド 2
サブスレッド 3
サブスレッド 4
サブスレッド 5
サブスレッド 6
サブスレッド 7
サブスレッド 8

"use strict"

const http = require("http")
const { Worker } = require("worker_threads")

http.createServer((req, res) => {
  // リクエストがきた際の処理
  const url = new URL(req.url, `http://${req.headers.host}`);
  const pathname = url.pathname;
  const n = Number(pathname.slice(1))

  if (Number.isNaN(n)) {
    // Number.isNaNで数値かどうかを判定し、数値でなかった場合は無視
    return res.end()
  }

  // コンストラクタの第二引数でサブスレッドに値を渡しつつ、サブスレッドを生成
  // サブスレッドからworker_threads.parentPort.postMessage()で送信されたメッセージは、Workerインスタンスの
  // messageイベントで取得できる
  // フィボナッチ数の計算は別スレッドで行う。Webアプリケーションを動すしているスレッドとは、別のスレッドである。
  new Worker(`${__dirname}/fibonacci.js`, { workerData: n })
    .on("message", result => res.end(result.toString()))
}).listen(8000)

"use strict"

// 指定された位置のフィボナッチ数を返す
function fibonacci(n) {
  // nが1以下の場合は、それ以外の場合は直前の2つのフィボナッチ数の和を返す
  return n <= 1 ? n : fibonacci(n - 1) + fibonacci(n - 2)
}

const { workerData, parentPort } = require("worker_threads")

// フィボナッチ数の計算結果をメインスレッドに送信
// workerDataでメインスレッドから入力を受け取り、その値に対応するフィボナッチ数を計算して
// parentPort.postMessage()でメインスレッドに返す。
parentPort.postMessage(fibonacci(workerData))

module.exports = {
  fibonacci,
}

Completed requests:  576
Total errors:        7
Total time:          10.012 s
Mean latency:        4868.6 ms
Effective rps:       58

Completed requests:  1084
Total errors:        10
Total time:          10.006 s
Mean latency:        2820.5 ms
Effective rps:       108

const { parentPort } = require("worker_threads")

// messageイベントの監視により、メインスレッドからのメッセージの受信を待機
// 受信したらフィボナッチ数を計算して結果をメインスレッドに送信
parentPort.on("message", n => parentPort.postMessage(fibonacci(n)))

"use strict"

// 指定された位置のフィボナッチ数を返す
function fibonacci(n) {
  // nが1以下の場合は、それ以外の場合は直前の2つのフィボナッチ数の和を返す
  return n <= 1 ? n : fibonacci(n - 1) + fibonacci(n - 2)
}

const { parentPort } = require("worker_threads")

// messageイベントの監視により、メインスレッドからのメッセージの受信を待機
// 受信したらフィボナッチ数を計算して結果をメインスレッドに送信
// parentPortオブジェクトには、onメソッドが生えており、
// 'message'イベントを処理するイベントハンドラーを登録することで、メインスレッドからのデータを受信できるようになります:
parentPort.on("message", n => parentPort.postMessage(fibonacci(n)))

"use strict"

const { Worker } = require("worker_threads")

class ThreadPool {
  // 空きスレッド、キューを初期化

  // インスタンスのメンバ
  availableWorkers = []
  queue = []

  // sizeは生成するスレッドの数
  // Workerコンストラクタの第一引数(サブスレッドで実行するファイルパス)と第二引数(オプション)
  // は利用元から指定できるようにしておく。
  // リクエストを受け付けた時、空いているスレッドがあればそれを使って処理をするが、なければリクエストをキューに積む。
  // それぞれのスレッドは、割り当てられた処理が完了したら、キューからリクエストを(あれば)取り出して処理する
  constructor(size, filePath, options) {
    // 引数で指定された通りにスレッドを生成してプール
    for(let i = 0; i < size; i++) {
      // この別スレッドの生成時にメッセージイベントを登録していた
      // Workerインスタンスはされ続けている限り、GCで消されたりしないはず
      this.availableWorkers.push(new Worker(filePath, options))
    }
  }

  // 外部からの処理要求を受け付けるメソッド(このメソッドは非同期APIだね)
  executeInThread(arg) {
    return new Promise(resolve => {
      const request = { resolve, arg }
      // 空きスレッドがあればリクエストを処理しなければキューにつむ
      const worker = this.availableWorkers.pop()
      // jsでは#をつけるとプライベートになるそう
      // TSのprivateの方がまだわかりやすいな。#ってなんやねんってなりそう
      worker ? this.#process(worker, request) : this.queue.push(request)
    })
  }

  // 実際にスレッドで処理を実行するprivateメソッド
  // resolve, argはスレッド呼び出し時のパラメータ
  #process(worker, { resolve, arg }) {
    // 一回限りのリスナを登録している
    worker.once("message", result => {
      // リクエスト元に結果を返す
      // このresultがフィボナッチ数。
      resolve(result)

      // キューに積まれたリクエストがあれば処理をして、なければ空きスレッドに戻す
      const request = this.queue.shift()
      request
        ? this.#process(worker, request)
        : this.availableWorkers.push(worker)
    })
    // workerにargを返す
    // filePathを実行する予定のワーカーに対して、メッセージを送る
    //
    worker.postMessage(arg)
  }
}

module.exports = {
  ThreadPool,
}

"use strict"

const http = require("http")
const { Worker } = require("worker_threads")
const cpuCount = require("os").cpus().length
const { ThreadPool } = require("./thread-pool")

// CPUコア数と同じサイズのスレッドプールを作成
const threadPool = new ThreadPool(cpuCount, `${__dirname}/fibonacci.js`)

// リクエストを受け付ける処理自体は非同期的な処理(async)だから、前の処理が後続のリクエストが捌けないとかはなさそう
http.createServer(async (req, res) => {
  // リクエストがきた際の処理
  const url = new URL(req.url, `http://${req.headers.host}`);
  const pathname = url.pathname;
  const n = Number(pathname.slice(1))

  if (Number.isNaN(n)) {
    // Number.isNaNで数値かどうかを判定し、数値でなかった場合は無視
    return res.end()
  }

  // ここでawaitされるのか
  const result = await threadPool.executeInThread(n)
  res.end(result.toString())

}).listen(3000)

Completed requests:  4162
Total errors:        98
Total time:          10.003 s
Mean latency:        908.9 ms
Effective rps:       416

"use strict"

const { parentPort, workerData } = require("worker_threads")

// メインスレッドに返す
parentPort.postMessage(
  workerData.buffer,
  // postMessage()の第二引数に転送対象オブジェクトを指定
  workerData.transfer ? [workerData.buffer] : []
)

"use strict"

const perf_hooks = require("perf_hooks")
const { Worker } = require("worker_threads")

function useMaybeTransfer(transfer) {
  // 1GBのArrayBufferを生成
  const buffer = new ArrayBuffer(1024 * 1024 * 1024)
  // 現在時刻を記録
  const start = perf_hooks.performance.now()
  new Worker(
    "./maybe-transfer.js",
    {
      workerData: { buffer, transfer },
      // transferListプロパティに転送対象オブジェクトを指定
      transferList: transfer ? [buffer] : []
    }
  ).on("message", () => {
    // サブスレッドから値が戻ってくるまでにかかった時間を出力
    console.log(perf_hooks.performance.now() - start)
  })

  // サブスレッドに渡した値がどう見えるか確認
  console.log(buffer);
}

// 転送を利用する場合
useMaybeTransfer(true)

// 転送を利用しない場合(値のコピー)
useMaybeTransfer(false)

node main.js
ArrayBuffer { (detached), byteLength: 0 }
ArrayBuffer {
  [Uint8Contents]: <00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ... 1073741724 more bytes>,
  byteLength: 1073741824
}
1174.1023473739624
3604.235155105591