feat: support reusePort on server listen

.github/workflows/nodejs.yml

@@ -12,6 +12,6 @@ jobs:
     uses: node-modules/github-actions/.github/workflows/node-test.yml@master
     with:
       os: 'ubuntu-latest, macos-latest'
-      version: '14, 16, 18, 20, 22'
+      version: '14, 16, 18, 20, 22, 23'
     secrets:
       CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}

README.md

@@ -8,8 +8,8 @@
 
 [npm-image]: https://img.shields.io/npm/v/egg-cluster.svg?style=flat-square
 [npm-url]: https://npmjs.org/package/egg-cluster
-[codecov-image]: https://codecov.io/github/eggjs/egg-cluster/coverage.svg?branch=master
-[codecov-url]: https://codecov.io/github/eggjs/egg-cluster?branch=master
+[codecov-image]: https://codecov.io/github/eggjs/cluster/coverage.svg?branch=master
+[codecov-url]: https://codecov.io/github/eggjs/cluster?branch=master
 [snyk-image]: https://snyk.io/test/npm/egg-cluster/badge.svg?style=flat-square
 [snyk-url]: https://snyk.io/test/npm/egg-cluster
 [download-image]: https://img.shields.io/npm/dm/egg-cluster.svg?style=flat-square

lib/agent_worker.js

@@ -24,7 +24,7 @@ if (options.startMode === 'worker_threads') {
   AgentWorker = require('./utils/mode/impl/process/agent').AgentWorker;
 }
 
-const debug = require('util').debuglog('egg-cluster');
+const debug = require('util').debuglog('egg-cluster:agent_worker');
 const ConsoleLogger = require('egg-logger').EggConsoleLogger;
 const consoleLogger = new ConsoleLogger({ level: process.env.EGG_AGENT_WORKER_LOGGER_LEVEL });
 

lib/app_worker.js

@@ -16,8 +16,9 @@ if (options.startMode === 'worker_threads') {
   AppWorker = require('./utils/mode/impl/process/app').AppWorker;
 }
 
+const os = require('os');
 const fs = require('fs');
-const debug = require('util').debuglog('egg-cluster');
+const debug = require('util').debuglog('egg-cluster:app_worker');
 const ConsoleLogger = require('egg-logger').EggConsoleLogger;
 const consoleLogger = new ConsoleLogger({
   level: process.env.EGG_APP_WORKER_LOGGER_LEVEL,
@@ -38,6 +39,13 @@ const port = options.port = options.port || listenConfig.port;
 const debugPort = options.debugPort;
 const protocol = (httpsOptions.key && httpsOptions.cert) ? 'https' : 'http';
 
+let reusePort = options.reusePort = options.reusePort || listenConfig.reusePort;
+if (reusePort && os.platform() !== 'linux') {
+  // Currently only linux is supported
+  reusePort = false;
+  debug('platform %s is not support currently, set reusePort to false', os.platform());
+}
+
 AppWorker.send({
   to: 'master',
   action: 'realport',
@@ -121,10 +129,21 @@ function startServer(err) {
         exitProcess();
         return;
       }
-      const args = [ port ];
-      if (listenConfig.hostname) args.push(listenConfig.hostname);
-      debug('listen options %s', args);
-      server.listen(...args);
+      if (reusePort) {
+        const listenOptions = { port, reusePort };
+        if (listenConfig.hostname) {
+          listenOptions.host = listenConfig.hostname;
+        }
+        debug('listen options %s', listenOptions);
+        server.listen(listenOptions);
+      } else {
+        const args = [ port ];
+        if (listenConfig.hostname) {
+          args.push(listenConfig.hostname);
+        }
+        debug('listen options %s', args);
+        server.listen(...args);
+      }
     }
     if (debugPortServer) {
       debug('listen on debug port: %s', debugPort);

lib/master.js

@@ -34,6 +34,7 @@ class Master extends EventEmitter {
    *  - {Object} [plugins] - customized plugins, for unittest
    *  - {Number} [workers] numbers of app workers, default to `os.cpus().length`
    *  - {Number} [port] listening port, default to 7001(http) or 8443(https)
+   *  - {Boolean} [reusePort] setting `reusePort` to `true` allows multiple sockets on the same host to bind to the same port. Incoming connections are distributed by the operating system to listening sockets. This option is available only on some platforms, such as Linux 3.9+, DragonFlyBSD 3.6+, FreeBSD 12.0+, Solaris 11.4, and AIX 7.2.5+. **Default:** `false`.
    *  - {Number} [debugPort] listening a debug port on http protocol
    *  - {Object} [https] https options, { key, cert, ca }, full path
    *  - {Array|String} [require] will inject into worker/agent process

lib/utils/options.js

@@ -12,6 +12,7 @@ module.exports = function(options) {
     framework: '',
     baseDir: process.cwd(),
     port: options.https ? 8443 : null,
+    reusePort: false,
     workers: null,
     plugins: null,
     https: false,
@@ -68,7 +69,6 @@ module.exports = function(options) {
 
   const isDebug = process.execArgv.some(argv => argv.includes('--debug') || argv.includes('--inspect'));
   if (isDebug) options.isDebug = isDebug;
-
   return options;
 };
 

package.json

@@ -8,7 +8,7 @@
     "test": "npm run lint -- --fix && npm run test-local",
     "test-local": "egg-bin test --ts false",
     "cov": "egg-bin cov --prerequire --timeout 100000 --ts false",
-    "ci": "npm run lint && npm run cov"
+    "ci": "npm run lint && node test/reuseport_cluster.js && npm run cov"
   },
   "files": [
     "index.js",

test/app_worker.test.js

@@ -232,6 +232,35 @@ describe('test/app_worker.test.js', () => {
         .expect(200);
     });
 
+    it('should set reusePort=true in config', async () => {
+      app = utils.cluster('apps/app-listen-reusePort');
+      // app.debug();
+      await app.ready();
+
+      app.expect('code', 0);
+      app.expect('stdout', /egg started on http:\/\/127.0.0.1:17010/);
+
+      await request('http://0.0.0.0:17010')
+        .get('/')
+        .expect('done')
+        .expect(200);
+
+      await request('http://127.0.0.1:17010')
+        .get('/')
+        .expect('done')
+        .expect(200);
+
+      await request('http://localhost:17010')
+        .get('/')
+        .expect('done')
+        .expect(200);
+
+      await request('http://127.0.0.1:17010')
+        .get('/port')
+        .expect('17010')
+        .expect(200);
+    });
+
     it('should use hostname in config', async () => {
       const url = address.ip() + ':17010';
 

test/fixtures/apps/app-listen-reusePort/app.js

@@ -0,0 +1,6 @@
+'use strict';
+
+module.exports = app => {
+  // don't use the port that egg-mock defined
+  app._options.port = undefined;
+};

test/fixtures/apps/app-listen-reusePort/app/router.js

@@ -0,0 +1,9 @@
+module.exports = app => {
+  app.get('/', ctx => {
+    ctx.body = 'done';
+  });
+
+  app.get('/port', ctx => {
+    ctx.body = ctx.app._options.port;
+  });
+};

test/fixtures/apps/app-listen-reusePort/config/config.default.js

@@ -0,0 +1,9 @@
+module.exports = {
+  keys: '123',
+  cluster: {
+    listen: {
+      port: 17010,
+      reusePort: true,
+    },
+  },
+};

test/fixtures/apps/app-listen-reusePort/package.json

@@ -0,0 +1,3 @@
+{
+  "name": "app-listen-reusePort"
+}

test/master.test.js

@@ -29,6 +29,17 @@ describe('test/master.test.js', () => {
         .end(done);
     });
 
+    it('start success with reusePort=true', done => {
+      mm.env('local');
+      app = utils.cluster('apps/master-worker-started', { reusePort: true });
+
+      app.expect('stdout', /egg start/)
+        .expect('stdout', /egg started/)
+        .notExpect('stdout', /\[master\] agent_worker#1:\d+ start with clusterPort:\d+/)
+        .expect('code', 0)
+        .end(done);
+    });
+
     it('start success in prod env', done => {
       mm.env('prod');
       app = utils.cluster('apps/mock-production-app').debug(false);

test/reuseport_cluster.js

@@ -0,0 +1,70 @@
+const cluster = require('node:cluster');
+const http = require('node:http');
+const numCPUs = require('node:os').availableParallelism();
+const process = require('node:process');
+
+function request(index) {
+  http.get('http://localhost:17001/', res => {
+    const { statusCode } = res;
+    console.log(index, res.statusCode, res.headers);
+    let error;
+    // Any 2xx status code signals a successful response but
+    // here we're only checking for 200.
+    if (statusCode !== 200) {
+      error = new Error('Request Failed.\n' +
+                        `Status Code: ${statusCode}`);
+    }
+    if (error) {
+      console.error(error.message);
+      // Consume response data to free up memory
+      res.resume();
+      return;
+    }
+    res.setEncoding('utf8');
+    let rawData = '';
+    res.on('data', chunk => { rawData += chunk; });
+    res.on('end', () => {
+      try {
+        console.log(rawData);
+      } catch (e) {
+        console.error(e.message);
+      }
+    });
+  }).on('error', e => {
+    console.error(`Got error: ${e.stack}`);
+  });
+}
+
+if (cluster.isPrimary) {
+  console.log(`Primary ${process.pid} is running`);
+
+  // Fork workers.
+  for (let i = 0; i < numCPUs; i++) {
+    cluster.fork();
+  }
+
+  cluster.on('exit', (worker, code, signal) => {
+    console.log(`worker ${worker.process.pid} died, code: ${code}, signal: ${signal}`);
+  });
+
+  setTimeout(() => {
+    for (let i = 0; i < 20; i++) {
+      request(i);
+    }
+  }, 2000);
+  setTimeout(() => {
+    process.exit(0);
+  }, 5000);
+} else {
+  // Workers can share any TCP connection
+  // In this case it is an HTTP server
+  http.createServer((req, res) => {
+    res.writeHead(200);
+    res.end('hello world\n');
+  }).listen({
+    port: 17001,
+    reusePort: true,
+  });
+
+  console.log(`Worker ${process.pid} started`);
+}