Remove multithreading

Author: tshemsedinov

lib/application.js

@@ -3,7 +3,6 @@
 const { node, npm } = require('./dependencies.js');
 const { path, events, vm, fs, fsp } = node;
 const { common } = npm;
-
 const security = require('./security.js');
 
 const SCRIPT_OPTIONS = { timeout: 5000 };

lib/auth.js

@@ -124,5 +124,5 @@ module.exports = () => {
       .select('SystemUser', ['Id', 'Password'], { login })
       .then(([user]) => user);
 
-  return Object.freeze({ start, restore, remove, save, registerUser, getUser });
+  return { start, restore, remove, save, registerUser, getUser };
 };

lib/database.js

@@ -1,6 +1,7 @@
 'use strict';
 
-const { Pool } = require('pg');
+const { npm } = require('./dependencies.js');
+const { pg } = npm;
 const application = require('./application.js');
 
 const OPERATORS = ['>=', '<=', '<>', '>', '<'];
@@ -47,7 +48,7 @@ const updates = (delta, firstArgIndex = 1) => {
 
 class Database {
   constructor(config) {
-    this.pool = new Pool(config);
+    this.pool = new pg.Pool(config);
   }
 
   query(sql, values) {

lib/logger.js

@@ -13,10 +13,10 @@ const COLORS = {
 const DATETIME_LENGTH = 19;
 
 class Logger {
-  constructor(logPath, threadId = 0) {
+  constructor(logPath) {
     this.path = logPath;
     const date = new Date().toISOString().substring(0, 10);
-    const filePath = path.join(logPath, `${date}-W${threadId}.log`);
+    const filePath = path.join(logPath, `${date}.log`);
     this.stream = fs.createWriteStream(filePath, { flags: 'a' });
     this.home = path.dirname(this.path);
   }

lib/security.js

@@ -81,4 +81,4 @@ const validatePassword = (password, hash = defaultHash) =>
     );
   });
 
-module.exports = Object.freeze({ hashPassword, validatePassword });
+module.exports = { hashPassword, validatePassword };

lib/semaphore.js

@@ -1,10 +1,10 @@
 'use strict';
 
 const { node, npm } = require('./dependencies.js');
-const { http, https, worker } = node;
+const { http, https } = node;
 const { common, ws } = npm;
-
-const Semaphore = require('./semaphore.js');
+const { Channel } = require('./channel.js');
+const { application } = require('./application.js');
 
 const SHUTDOWN_TIMEOUT = 5000;
 const SHORT_TIMEOUT = 500;
@@ -19,61 +19,46 @@ const receiveBody = async (req) => {
 };
 
 class Server {
-  constructor(config, { application, Channel }) {
+  constructor(config) {
     this.config = config;
-    this.application = application;
-    this.Channel = Channel;
     this.channels = new Map();
-    const { host, balancer, protocol, ports, concurrency, queue } = config;
-    this.semaphore = new Semaphore(concurrency, queue.size, queue.timeout);
-    const { threadId } = worker;
-    this.balancer = balancer && threadId === 1;
-    const skipBalancer = balancer ? 1 : 0;
-    this.port = this.balancer ? balancer : ports[threadId - skipBalancer - 1];
-    const transport = protocol === 'http' || this.balancer ? http : https;
+    const { port, host, protocol, cert } = config;
+    this.port = port;
+    this.host = host;
+    this.protocol = protocol;
+    const transport = protocol === 'http' ? http : https;
     const listener = this.listener.bind(this);
-    this.server = transport.createServer({ ...application.cert }, listener);
+    this.server = transport.createServer(cert, listener);
     this.ws = new ws.Server({ server: this.server });
     this.ws.on('connection', async (connection, req) => {
       const channel = await new Channel(req, null, connection, application);
       connection.on('message', (data) => {
         channel.message(data);
       });
     });
-    this.protocol = protocol;
-    this.host = host;
     this.server.listen(this.port, host);
   }
 
   async listener(req, res) {
-    const { channels, Channel } = this;
     let finished = false;
     const { url, connection } = req;
-    const channel = await new Channel(req, res, null, this.application);
-    channels.set(connection, channel);
+    const channel = await new Channel(req, res, null, application);
+    this.channels.set(connection, channel);
 
     const timer = setTimeout(() => {
       if (finished) return;
       finished = true;
-      channels.delete(connection);
+      this.channels.delete(connection);
       channel.error(504);
     }, LONG_RESPONSE);
 
     res.on('close', () => {
       if (finished) return;
       finished = true;
       clearTimeout(timer);
-      channels.delete(connection);
+      this.channels.delete(connection);
     });
 
-    if (this.balancer) {
-      const host = common.parseHost(req.headers.host);
-      const port = common.sample(this.config.ports);
-      const { protocol } = this.config;
-      channel.redirect(`${protocol}://${host}:${port}/`);
-      return;
-    }
-
     if (url.startsWith('/api')) this.request(channel);
     else channel.static();
   }
@@ -117,7 +102,7 @@ class Server {
 
   async close() {
     this.server.close((err) => {
-      if (err) this.application.logger.error(err.stack);
+      if (err) application.logger.error(err.stack);
     });
     if (this.channels.size === 0) {
       await common.timeout(SHORT_TIMEOUT);

lib/server.js

@@ -1,40 +1,61 @@
 'use strict';
 
-const { Worker } = require('worker_threads');
-const path = require('path');
-
+const { node } = require('./lib/dependencies.js');
+const { fsp, path } = node;
+const application = require('./lib/application.js');
 const Config = require('./lib/config.js');
-
-const PATH = process.cwd();
-const CFG_PATH = path.join(PATH, 'application/config');
-
-const options = { trackUnmanagedFds: true };
-
-(async () => {
-  const config = await new Config(CFG_PATH);
-  const { balancer, ports = [], workers = {} } = config.server;
-  const count = ports.length + (balancer ? 1 : 0) + (workers.pool || 0);
-  let active = count;
-  const threads = new Array(count);
-
-  const start = (id) => {
-    const worker = new Worker('./lib/worker.js', options);
-    threads[id] = worker;
-    worker.on('exit', (code) => {
-      if (code !== 0) start(id);
-      else if (--active === 0) process.exit(0);
-    });
+const Logger = require('./lib/logger.js');
+const Database = require('./lib/database.js');
+const Server = require('./lib/server.js');
+const Channel = require('./lib/channel.js');
+const initAuth = require('./lib/auth.js');
+
+const loadCert = async (certPath) => {
+  const key = await fsp.readFile(path.join(certPath, 'key.pem'));
+  const cert = await fsp.readFile(path.join(certPath, 'cert.pem'));
+  return { key, cert };
+};
+
+const main = async () => {
+  const configPath = path.join(application.path, 'config');
+  const config = await new Config(configPath);
+  const logPath = path.join(application.root, 'log');
+  const logger = await new Logger(logPath);
+  Object.assign(application, { config, logger });
+
+  const logError = (err) => {
+    logger.error(err ? err.stack : 'No exception stack available');
   };
 
-  for (let id = 0; id < count; id++) start(id);
+  process.on('uncaughtException', logError);
+  process.on('warning', logError);
+  process.on('unhandledRejection', logError);
+
+  const certPath = path.join(application.path, 'cert');
+  const cert = await loadCert(certPath).catch(() => {
+    logError(new Error('Can not load TLS certificates'));
+  });
+
+  application.db = new Database(config.database);
+  application.auth = initAuth();
+  application.sandboxInject('auth', application.auth);
 
-  const stop = async () => {
-    console.log();
-    for (const worker of threads) {
-      worker.postMessage({ name: 'stop' });
-    }
+  const { ports } = config.server;
+  const options = { cert, application, Channel };
+  for (const port of ports) {
+    application.server = new Server(config.server, options);
+    logger.system(`Listen port ${port}`);
+  }
+
+  await application.init();
+  logger.system('Application started');
+
+  const stop = () => {
+    process.exit(0);
   };
 
   process.on('SIGINT', stop);
   process.on('SIGTERM', stop);
-})();
+};
+
+main();