From 3ededbbc350f1a041e4f4bc67784d66a1759b9bb Mon Sep 17 00:00:00 2001
From: Julian Simioni <julian@simioni.org>
Date: Fri, 21 Oct 2022 14:43:21 -0400
Subject: [PATCH] feat(cluster): Allow multiple worker processes with cluster
 module

This adds the ability for the API to use multiple worker processes with
Node.js's builtin [`cluster`](https://nodejs.org/api/cluster.html)
module.

Historically, I've been opposed to adding this functionality, preferring
that Pelias users manage parallelism on their own such as with
Kubernetes or something else.

But there are enough use cases where that sort of orchestration isn't
worth the complexity, and you want more than one CPU's worth of API
requests.

This implementation is essentially lifted from the one in Placeholder,
with mostly cosmetic changes only.

The biggest difference is the default. For backwards compatibility,
without specifing the new `CPUS` environment variable, the API will
_not_ use the cluster module and operate as a single process just like
before.

_With_ the `CPUS` variable set to a number, that many worker processes
will be launched, up to the number of CPUs detected on the machine.
---
 README.md |  1 +
 index.js  | 81 ++++++++++++++++++++++++++++++++++++++++++++++++-------
 2 files changed, 73 insertions(+), 9 deletions(-)

diff --git a/README.md b/README.md
index 7643121d8..873512579 100644
--- a/README.md
+++ b/README.md
@@ -220,6 +220,7 @@ Most Pelias configuration is done through pelias-config, however the API has add
 | --- | --- | --- |
 | HOST | `undefined` | The network interface the Pelias API will bind to. Defaults to whatever the current Node.js default is, which is currently to listen on all interfaces. See the [Node.js Net documentation](https://nodejs.org/api/net.html#net_server_listen_port_host_backlog_callback) for more info. |
 | PORT | 3100 | The TCP port the Pelias API will use for incoming network connections. |
+| CPUS | 1 | The maximum number of worker processes to be launched. Will not launch more than one process per detected CPU |
 
 ## Contributing
 
diff --git a/index.js b/index.js
index 6b6cfb095..041e73fbb 100644
--- a/index.js
+++ b/index.js
@@ -1,3 +1,6 @@
+const os = require('os');
+const cluster = require('cluster');
+
 const logger = require('pelias-logger').get('api');
 const type_mapping = require('./helper/type_mapping');
 
@@ -5,21 +8,81 @@ const app = require('./app'),
     port = ( process.env.PORT || 3100 ),
     host = ( process.env.HOST || undefined );
 
+// determine the number of processes to launch
+// by default launch only a single process,
+// but if the CPUS environment variable is set, launch up to one process per CPU detected
+const envCpus = parseInt( process.env.CPUS, 10 );
+const cpus = Math.min( envCpus || 1 , os.availableParallelism() );
+
 let server;
+let terminating = false;
+
+logger.info('Starting Pelias API using %d CPUs', cpus);
+
+// simple case where cluster module is disabled with CPUS=1
+// or if this is a worker
+if ( cpus === 1 || cluster.isWorker ) {
+  startServer();
+// if using the cluster module, do more work to set up all the workers
+} else if ( cluster.isMaster ) {
+  // listen to worker ready message and print a message
+  cluster.on('online', (worker) => {
+    if (Object.keys(cluster.workers).length === cpus) {
+      logger.info( `pelias is now running on http://${host || `::`}:${port}` );
+    }
+  });
+
+  // set up worker exit event that prints error message
+  cluster.on('exit', (worker, code, signal) => {
+    if (!terminating) {
+      logger.error('[master] worker died', worker.process.pid);
+    }
+  });
 
-// load Elasticsearch type mappings before starting web server
-type_mapping.load(() => {
-  server = app.listen( port, host, () => {
-    // ask server for the actual address and port its listening on
-    const listenAddress = server.address();
-    logger.info( `pelias is now running on http://${listenAddress.address}:${listenAddress.port}` );
+  // create a handler that prints when a new worker is created via fork
+  cluster.on('fork', (worker, code, signal) => {
+    logger.info('[master] worker forked', worker.process.pid);
   });
-});
 
+  // call fork to create the desired number of workers
+  for( let c = 0; c < cpus; c++ ){
+    cluster.fork();
+  }
+}
+
+// an exit handler that either closes the local Express server
+// or, if using the cluster module, forwards the signal to all workers
 function exitHandler() {
-  logger.info('Pelias API shutting down');
+  terminating = true;
+
+  if (cluster.isPrimary) {
+    logger.info('Pelias API shutting down');
+    for (const id in cluster.workers) {
+      cluster.workers[id].send('shutdown');
+      cluster.workers[id].disconnect();
+    }
+  }
 
-  server.close();
+  if (server) {
+    server.close();
+  }
+}
+
+function startServer() {
+  // load Elasticsearch type_mapping before starting the web server
+  // This has to be run on each forked worker because unlike "real"
+  // unix `fork`, these forks don't share memory with other workers
+  type_mapping.load(() => {
+    server = app.listen( port, host, () => {
+      // ask server for the actual address and port its listening on
+      const listenAddress = server.address();
+      if (cluster.isMaster) {
+        logger.info( `pelias is now running on http://${listenAddress.address}:${listenAddress.port}` );
+      } else {
+        logger.info( `pelias worker ${process.pid} online` );
+      }
+    });
+  });
 }
 
 process.on('SIGINT', exitHandler);