feat: chapter 11

Author: lmammino

11-advanced-recipes/09-cpu-bound/README.md

@@ -0,0 +1,42 @@
+# 09-cpu-bound
+
+This sample demonstrates how to run a CPU-bound task in Node.js using
+setImmediate or multiple processes.
+
+## Run
+
+To start the server, run:
+
+```bash
+node index.js
+```
+
+Then try to send one or more request to trigger the subset sum task:
+
+```shell script
+curl -G http://localhost:8000/subsetSum --data-urlencode "data=[116,119,101,101,-116,109,101,-105,-102,117,-115,-97,119,-116,-104,-105,115]" --data-urlencode "sum=0"
+```
+
+While a subset sum task is running, you can check the responsiveness of the
+server with a command like this:
+
+```shell script
+curl -G http://localhost:8000
+```
+
+If the subset sum takes too long on your machine or is too fast for checking the
+responsiveness of the server, try to reduce or increase the number of items of
+the array given as input.
+
+In the file `index.js`, try to swap between the various implementations of the
+`SubsetSum` API to compare the differences in the responsiveness of the server.
+
+```js
+// index.js
+
+// ...
+//import { SubsetSum } from './subsetSum.js'
+//import { SubsetSum } from './subsetSumDefer.js'
+//import { SubsetSum } from './subsetSumFork.js'
+//import { SubsetSum } from './subsetSumThreads.js'
+```

11-advanced-recipes/09-cpu-bound/index.js

@@ -0,0 +1,25 @@
+import { createServer } from 'node:http'
+import { SubsetSum } from './subsetSum.js'
+// import { SubsetSum } from './subsetSumDefer.js'
+// import { SubsetSum } from './subsetSumFork.js'
+// import { SubsetSum } from './subsetSumThreads.js'
+
+createServer((req, res) => {
+  const url = new URL(req.url, 'http://localhost')
+  if (url.pathname !== '/subsetSum') {
+    res.writeHead(200)
+    return res.end("I'm alive!\n")
+  }
+
+  const data = JSON.parse(url.searchParams.get('data'))
+  const sum = JSON.parse(url.searchParams.get('sum'))
+  res.writeHead(200)
+  const subsetSum = new SubsetSum(sum, data)
+  subsetSum.on('match', match => {
+    res.cork()
+    res.write(`Match: ${JSON.stringify(match)}\n`)
+    res.uncork()
+  })
+  subsetSum.on('end', () => res.end())
+  subsetSum.start()
+}).listen(8000, () => console.log('Server started'))

11-advanced-recipes/09-cpu-bound/package.json

@@ -0,0 +1,15 @@
+{
+  "name": "09-cpu-bound",
+  "version": "1.0.0",
+  "description": "This sample demonstrates how to run a CPU-bound task in Node.js using setImmediate or multiple processes.",
+  "type": "module",
+  "scripts": {},
+  "engines": {
+    "node": ">=23"
+  },
+  "engineStrict": true,
+  "keywords": [],
+  "author": "Luciano Mammino and Mario Casciaro",
+  "license": "MIT",
+  "dependencies": {}
+}

11-advanced-recipes/09-cpu-bound/processPool.js

@@ -0,0 +1,50 @@
+import { fork } from 'node:child_process'
+
+export class ProcessPool {
+  constructor(file, poolMax) {
+    this.file = file
+    this.poolMax = poolMax
+    this.pool = []
+    this.active = []
+    this.waiting = []
+  }
+
+  acquire() {
+    return new Promise((resolve, reject) => {
+      let worker
+      if (this.pool.length > 0) {
+        worker = this.pool.pop()
+        this.active.push(worker)
+        return resolve(worker)
+      }
+
+      if (this.active.length >= this.poolMax) {
+        return this.waiting.push({ resolve, reject })
+      }
+
+      worker = fork(this.file)
+      worker.once('message', message => {
+        if (message === 'ready') {
+          this.active.push(worker)
+          return resolve(worker)
+        }
+        worker.kill()
+        reject(new Error('Improper process start'))
+      })
+      worker.once('exit', code => {
+        console.log(`Worker exited with code ${code}`)
+        this.active = this.active.filter(w => worker !== w)
+        this.pool = this.pool.filter(w => worker !== w)
+      })
+    })
+  }
+
+  release(worker) {
+    if (this.waiting.length > 0) {
+      const { resolve } = this.waiting.shift()
+      return resolve(worker)
+    }
+    this.active = this.active.filter(w => worker !== w)
+    this.pool.push(worker)
+  }
+}

11-advanced-recipes/09-cpu-bound/subsetSum.js

@@ -0,0 +1,31 @@
+import { EventEmitter } from 'node:events'
+
+export class SubsetSum extends EventEmitter {
+  constructor(sum, set) {
+    super()
+    this.sum = sum
+    this.set = set
+    this.totalSubsets = 0
+  }
+
+  _combine(set, subset) {
+    for (let i = 0; i < set.length; i++) {
+      const newSubset = subset.concat(set[i])
+      this._combine(set.slice(i + 1), newSubset)
+      this._processSubset(newSubset)
+    }
+  }
+
+  _processSubset(subset) {
+    console.log('Subset', ++this.totalSubsets, subset)
+    const res = subset.reduce((prev, item) => prev + item, 0)
+    if (res === this.sum) {
+      this.emit('match', subset)
+    }
+  }
+
+  start() {
+    this._combine(this.set, [])
+    this.emit('end')
+  }
+}

11-advanced-recipes/09-cpu-bound/subsetSumDefer.js

@@ -0,0 +1,41 @@
+import { EventEmitter } from 'node:events'
+
+export class SubsetSum extends EventEmitter {
+  constructor(sum, set) {
+    super()
+    this.sum = sum
+    this.set = set
+    this.totalSubsets = 0
+  }
+
+  _combineInterleaved(set, subset) {
+    this.runningCombine++
+    setImmediate(() => {
+      this._combine(set, subset)
+      if (--this.runningCombine === 0) {
+        this.emit('end')
+      }
+    })
+  }
+
+  _combine(set, subset) {
+    for (let i = 0; i < set.length; i++) {
+      const newSubset = subset.concat(set[i])
+      this._combineInterleaved(set.slice(i + 1), newSubset)
+      this._processSubset(newSubset)
+    }
+  }
+
+  _processSubset(subset) {
+    console.log('Subset', ++this.totalSubsets, subset)
+    const res = subset.reduce((prev, item) => prev + item, 0)
+    if (res === this.sum) {
+      this.emit('match', subset)
+    }
+  }
+
+  start() {
+    this.runningCombine = 0
+    this._combineInterleaved(this.set, [])
+  }
+}

11-advanced-recipes/09-cpu-bound/subsetSumFork.js

@@ -0,0 +1,34 @@
+import { EventEmitter } from 'node:events'
+import { join } from 'node:path'
+import { ProcessPool } from './processPool.js'
+
+const workerFile = join(
+  import.meta.dirname,
+  'workers',
+  'subsetSumProcessWorker.js'
+)
+const workers = new ProcessPool(workerFile, 2)
+
+export class SubsetSum extends EventEmitter {
+  constructor(sum, set) {
+    super()
+    this.sum = sum
+    this.set = set
+  }
+
+  async start() {
+    const worker = await workers.acquire()
+    worker.send({ sum: this.sum, set: this.set })
+
+    const onMessage = msg => {
+      if (msg.event === 'end') {
+        worker.removeListener('message', onMessage)
+        workers.release(worker)
+      }
+
+      this.emit(msg.event, msg.data)
+    }
+
+    worker.on('message', onMessage)
+  }
+}

11-advanced-recipes/09-cpu-bound/subsetSumThreads.js

@@ -0,0 +1,34 @@
+import { EventEmitter } from 'node:events'
+import { join } from 'node:path'
+import { ThreadPool } from './threadPool.js'
+
+const workerFile = join(
+  import.meta.dirname,
+  'workers',
+  'subsetSumThreadWorker.js'
+)
+const workers = new ThreadPool(workerFile, 2)
+
+export class SubsetSum extends EventEmitter {
+  constructor(sum, set) {
+    super()
+    this.sum = sum
+    this.set = set
+  }
+
+  async start() {
+    const worker = await workers.acquire()
+    worker.postMessage({ sum: this.sum, set: this.set })
+
+    const onMessage = msg => {
+      if (msg.event === 'end') {
+        worker.removeListener('message', onMessage)
+        workers.release(worker)
+      }
+
+      this.emit(msg.event, msg.data)
+    }
+
+    worker.on('message', onMessage)
+  }
+}

11-advanced-recipes/09-cpu-bound/threadPool.js

@@ -0,0 +1,46 @@
+import { Worker } from 'node:worker_threads'
+
+export class ThreadPool {
+  constructor(file, poolMax) {
+    this.file = file
+    this.poolMax = poolMax
+    this.pool = []
+    this.active = []
+    this.waiting = []
+  }
+
+  acquire() {
+    return new Promise((resolve, reject) => {
+      let worker
+      if (this.pool.length > 0) {
+        worker = this.pool.pop()
+        this.active.push(worker)
+        return resolve(worker)
+      }
+
+      if (this.active.length >= this.poolMax) {
+        return this.waiting.push({ resolve, reject })
+      }
+
+      worker = new Worker(this.file)
+      worker.once('online', () => {
+        this.active.push(worker)
+        resolve(worker)
+      })
+      worker.once('exit', code => {
+        console.log(`Worker exited with code ${code}`)
+        this.active = this.active.filter(w => worker !== w)
+        this.pool = this.pool.filter(w => worker !== w)
+      })
+    })
+  }
+
+  release(worker) {
+    if (this.waiting.length > 0) {
+      const { resolve } = this.waiting.shift()
+      return resolve(worker)
+    }
+    this.active = this.active.filter(w => worker !== w)
+    this.pool.push(worker)
+  }
+}

11-advanced-recipes/09-cpu-bound/workers/subsetSumProcessWorker.js

@@ -0,0 +1,17 @@
+import { SubsetSum } from '../subsetSum.js'
+
+process.on('message', msg => {
+  const subsetSum = new SubsetSum(msg.sum, msg.set)
+
+  subsetSum.on('match', data => {
+    process.send({ event: 'match', data: data })
+  })
+
+  subsetSum.on('end', data => {
+    process.send({ event: 'end', data: data })
+  })
+
+  subsetSum.start()
+})
+
+process.send('ready')