Nexu.js is a multithreaded JavaScript runtime library

First, if you are not familiar with the project, it is recommended to read a series of articles written before. If you don’t want to read this, don’t worry. That’s going to be in there as well.

Now, let’s get started.

Last year, I started implementing Nexu.js, a multi-threaded server-side JavaScript runtime based on the Webkit/JavaScript kernel. I stopped doing it for a while, for reasons beyond my control that I’m not going to discuss here, mainly: I couldn’t bring myself to work long hours.

So let’s start by discussing the architecture of the Nexus and how it works.

Event loop

• No event loop
• There is a thread pool with (unlocked) task objects
• Each time setTimeout or setImmediate is called or a Promise is created, the task is queued up to the task queue clock.
• Whenever a task is scheduled, the first available thread selects the task and executes it.
• Processing promises on the CPU kernel. The call to promise.all () resolves the Promise in parallel.

ES6

• Async /await is supported and recommended
• Support for await (…).
• Support the deconstruction
• Support the async try/catch/finally

The module

• CommonJS is not supported. (the require (…). And the module exports)
• All modules use ES6’s import/export syntax
• Support dynamic import via import(‘file-or-packge’).then(…)
• Support import.meta, such as import.meta.filename and import.meta.dirname
• Additional features: Support direct import from URL, for example:

import { h } from 'https://unpkg.com/preact/dist/preact.esm.js';
Copy the code

EventEmitter

• Nexus implements the EventEmitter class based on Promise
• Event handlers sort on all threads and process execution in parallel.
• EventEmitter.emit(…) The return value of “is a Promise, which can be parsed into an array of returned values in the event handler.

Such as:

class EmitterTest extends Nexus.EventEmitter {
  constructor() {
    super(a);for(let i = 0; i < 4; i++)
      this.on('test', value => { console.log(`fired test ${i}! `); console.inspect(value); });
    for(let i = 0; i < 4; i++)
      this.on('returns-a-value', v => `${v + i}`); }}const test = new EmitterTest();

async function start() {
  await test.emit('test', { payload: 'test 1' });
  console.log('first test done! ');
  await test.emit('test', { payload: 'test 2' });
  console.log('second test done! ');
  const values = await test.emit('returns-a-value'.10);
  console.log('third test done, returned values are:'); console.inspect(values);
}

start().catch(console.error);
Copy the code

I/O

• All input/output is done through three primitives: Device, Filter, and Stream.
• All input/output primitives implement the EventEmitter class
• To use Device, you need to create a ReadableStream or WritableStream on top of Device
• To manipulate data, you can add Filters to either ReadableStream or WritableStream.
• Finally, use source.pipe(… DestinationStreams), and waits for source.resume() to process the data.
• All input/output operations are done using ArrayBuffer objects.
• Filter tries the process(buffer) method to process the data.

For example, convert UTF-8 to UTF6 using two separate output files.

  const startTime = Date.now();
  try {
    const device = new Nexus.IO.FilePushDevice('enwik8');
    const stream = new Nexus.IO.ReadableStream(device);

    stream.pushFilter(new Nexus.IO.EncodingConversionFilter("UTF-8", "UTF-16LE"));

    const wstreams = [0,1,2,3]
      .map(i => new Nexus.IO.WritableStream(new Nexus.IO.FileSinkDevice('enwik16-' + i)));

    console.log('piping...');

    stream.pipe(...wstreams);

    console.log('streaming...');

    await stream.resume();

    await stream.close();

    await Promise.all(wstreams.map(stream => stream.close()));

    console.log(`finished in ${(Date.now() * startTime) / 1000} seconds!`);
  } catch (e) {
    console.error('An error occurred: ', e);
  }
}

start().catch(console.error);

Copy the code

TCP/UDP

• Nexu.js provides an Acceptor class that binds IP addresses/ports and listens for connections
• Each time a connection request is received, the Connection event is raised and a Socket device is provided.
• Each Socket instance is a full-duplex I/O device.
• You can use ReadableStream and WritableStream to operate sockets.

The most basic example :(sending “Hello World” to the client)

const acceptor = new Nexus.Net.TCP.Acceptor();
let count = 0;

acceptor.on('connection', (socket, endpoint) => {
  const connId = count++;
  console.log(`connection #${connId} from ${endpoint.address}:${endpoint.port}`);
  const rstream = new Nexus.IO.ReadableStream(socket);
  const wstream = new Nexus.IO.WritableStream(socket);
  const buffer = new Uint8Array(13);
  const message = 'Hello World! \n';
  for(let i = 0; i < 13; i++)
    buffer[i] = message.charCodeAt(i);
  rstream.pushFilter(new Nexus.IO.UTF8StringFilter());
  rstream.on('data', buffer => console.log(`got message: ${buffer}`));
  rstream.resume().catch(e= > console.log(`client #${connId} at ${endpoint.address}:${endpoint.port}disconnected! `));
  console.log(`sending greeting to #${connId}! `);
  wstream.write(buffer);
});

acceptor.bind('127.0.0.1'.10000);
acceptor.listen();

console.log('server ready');
Copy the code

Http

• Nexus provides a Nexus.Net.HTTP.Server class that basically inherits TCPAcceptor
• Some basic interfaces
• When the server has finished parsing/validating the basic Http headers for the incoming connection, the Connection event is fired with the connection and the same information
• Each connection instance has a request and a Response object. These are input/output devices.
• You can construct ReadableStream and WritableStream to manipulate request/ Response.
• If you pipe to a Response object, the input stream will use a block-coded pattern. Otherwise, you can use Response.write () to write a regular string.

Complex examples :(basic Http server with block encoding, details omitted)

./** * Creates an input stream from a path. * @param path * @returns {Promise
      
       } */
      
async function createInputStream(path) {
  if (path.startsWith('/')) // If it starts with '/', omit it.
    path = path.substr(1);
  if (path.startsWith('. ')) // If it starts with '.', reject it.
    throw new NotFoundError(path);
  if (path === '/'| |! path)// If it's empty, set to index.html.
    path = 'index.html';
  /** * `import.meta.dirname` and `import.meta.filename` replace the old CommonJS `__dirname` and `__filename`. */
  const filePath = Nexus.FileSystem.join(import.meta.dirname, 'server_root', path);
  try {
    // Stat the target path.
    const {type} = await Nexus.FileSystem.stat(filePath);
    if (type === Nexus.FileSystem.FileType.Directory) // If it's a directory, return its 'index.html'
      return createInputStream(Nexus.FileSystem.join(filePath, 'index.html'));
    else if (type === Nexus.FileSystem.FileType.Unknown || type === Nexus.FileSystem.FileType.NotFound)
      // If it's not found, throw NotFound.
      throw new NotFoundError(path);
  } catch(e) {
    if (e.code)
      throw e;
    throw new NotFoundError(path);
  }
  try {
    // First, we create a device.
    const fileDevice = new Nexus.IO.FilePushDevice(filePath);
    // Then we return a new ReadableStream created using our source device.
    return new Nexus.IO.ReadableStream(fileDevice);
  } catch(e) {
    throw newInternalServerError(e.message); }}/** * Connections counter. */
let connections = 0;

/** * Create a new HTTP server. * @type {Nexus.Net.HTTP.Server} */
const server = new Nexus.Net.HTTP.Server();

// A server error means an error occurred while the server was listening to connections.
// We can mostly ignore such errors, we display them anyway.
server.on('error', e => {
  console.error(FgRed + Bright + 'Server Error: ' + e.message + '\n' + e.stack, Reset);
});

/** * Listen to connections. */
server.on('connection'.async (connection, peer) => {
  // Start with a connection ID of 0, increment with every new connection.
  const connId = connections++;
  // Record the start time for this connection.
  const startTime = Date.now();
  // Destructuring is supported, why not use it?
  const { request, response } = connection;
  // Parse the URL parts.
  const { path } = parseURL(request.url);
  // Here we'll store any errors that occur during the connection.
  const errors = [];
  // inStream is our ReadableStream file source, outStream is our response (device) wrapped in a WritableStream.
  let inStream, outStream;
  try {
    // Log the request.
    console.log(` > #${FgCyan + connId + Reset} ${Bright + peer.address}:${peer.port + Reset} ${ FgGreen + request.method + Reset} "${FgYellow}${path}${Reset}"`, Reset);
    // Set the 'Server' header.
    response.set('Server'.` nexus. Js/while `);
    // Create our input stream.
    inStream = await createInputStream(path);
    // Create our output stream.
    outStream = new Nexus.IO.WritableStream(response);
    // Hook all `error` events, add any errors to our `errors` array.
    inStream.on('error', e => { errors.push(e); });
    request.on('error', e => { errors.push(e); });
    response.on('error', e => { errors.push(e); });
    outStream.on('error', e => { errors.push(e); });
    // Set content type and request status.
    response
      .set('Content-Type', mimeType(path))
      .status(200);
    // Hook input to output(s).
    const disconnect = inStream.pipe(outStream);
    try {
      // Resume our file stream, this causes the stream to switch to HTTP chunked encoding.
      // This will return a promise that will only resolve after the last byte (HTTP chunk) is written.
      await inStream.resume();
    } catch (e) {
      // Capture any errors that happen during the streaming.
      errors.push(e);
    }
    // Disconnect all the callbacks created by `.pipe()`.
    return disconnect();
  } catch(e) {
    // If an error occurred, push it to the array.
    errors.push(e);
    // Set the content type, status, and write a basic message.
    response
      .set('Content-Type'.'text/plain')
      .status(e.code || 500)
      .send(e.message || 'An error has occurred.');
  } finally {
    // Close the streams manually. This is important because we may run out of file handles otherwise.
    if (inStream)
      await inStream.close();
    if (outStream)
      await outStream.close();
    // Close the connection, has no real effect with keep-alive connections.
    await connection.close();
    // Grab the response's status.
    let status = response.status();
    // Determine what colour to output to the terminal.
    const statusColors = {
      '200': Bright + FgGreen, // Green for 200 (OK),
      '404': Bright + FgYellow, // Yellow for 404 (Not Found)
      '500': Bright + FgRed // Red for 500 (Internal Server Error)
    };
    let statusColor = statusColors[status];
    if (statusColor)
      status = statusColor + status + Reset;
    // Log the connection (and time to complete) to the console.
    console.log(` < #${FgCyan + connId + Reset} ${Bright + peer.address}:${peer.port + Reset} ${ FgGreen + request.method + Reset} "${FgYellow}${path}${Reset}" ${status} The ${(Date.now() * startTime)}ms` +
      (errors.length ? "" + FgRed + Bright + errors.map(error= > error.message).join(', ') + Reset : Reset)); }});/** * IP and port to listen on. */
const ip = '0.0.0.0', port = 3000;
/** * Whether or not to set the `reuse` flag. (optional, default=false) */
const portReuse = true;
/** * Maximum allowed concurrent connections. Default is 128 on my system. (optional, system specific) * @type {number} */
const maxConcurrentConnections = 1000;
/** * Bind the selected address and port. */
server.bind(ip, port, portReuse);
/** * Start listening to requests. */
server.listen(maxConcurrentConnections);
/** * Happy streaming! * /
console.log(FgGreen + `Nexus.js HTTP server listening at ${ip}:${port}` + Reset);
Copy the code

The benchmark

I think I’ve covered everything that has been achieved so far. So now let’s talk about performance.

Here is the current benchmark for the appeal Http server, with 100 concurrent connections and a total of 10,000 requests:

This is ApacheBench, Version 2.3 <$Revision: 1796539 $> Copyright 1996 Adam Twiss, Zeus Technology Ltd, http://www.zeustech.net/ Licensed to The Apache Software Foundation, http://www.apache.org/ Benchmarking localhost (be patient)..... Done Server Software: Nexu.js /0.1.1 Server Hostname: Localhost Server Port: 3000 Document Path: / Document Length: 8673 bytes Concurrency Level: 100 Time takenforTests: 9.991 seconds Complete Requests: 10000 Failed requests: 0 Total transferred: 87880000 bytes HTML transferred: 86730000 bytes Requests per second: 1000.94 [#/sec] (mean)Time per Request: 99.906 [MS] (mean) Time per Request: 0.999 [MS] (mean, across all concurrent requests) Transfer rate 8590.14 [Kbytes/ SEC] Received Connection Times (ms) min mean[+/-sd] Median Max Connect: 0 0 0.1 0 1 Processing: 6 99 36.6 84 464 Waiting: 5 99 36.4 84 463 Total: 6 100 36.6 84 464 Percentage of the requests served within a certain time (ms) 50% 84 66% 97 75% 105 80% 112 90% 134 95%  188 98% 233 99% 238 100% 464 (longest request)Copy the code

1000 requests per second. On an old I7, it ran the benchmark software, an IDE that took up 5 gigabytes of memory, and the server itself.

voodooattack@voodooattack:~$ cat /proc/cpuinfo processor : 0 vendor_id : GenuineIntel cpu family : 6 model : 60 Model Name: Intel(R) Core(TM) i7-4770 CPU @ 3.40GHz Stepping: 3 Microcode: 0x22 CPU MHz: 3392.093 Cache Size: 8192 KB physical id : 0 siblings : 8 core id : 0 cpu cores : 4 apicid : 0 initial apicid : 0 fpu : yes fpu_exception : yes cpuid level : 13 wp : yes flags : fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush dts acpi mmx fxsr sse sse2 ss ht tm pbe syscall nx pdpe1gb rdtscp lm constant_tsc arch_perfmon pebs bts rep_good nopl xtopology nonstop_tsc cpuid aperfmperf pni  pclmulqdq dtes64 monitor ds_cpl vmx smx est tm2 ssse3 sdbg fma cx16 xtpr pdcm pcid sse4_1 sse4_2 x2apic movbe popcnt tsc_deadline_timer aes xsave avx f16c rdrand lahf_lm abm cpuid_fault tpr_shadow vnmi flexpriority ept vpid fsgsbase Tsc_adjust bMI1 avx2 smep bMI2 erms Invpcid xSaveopt dtherm IDA Arat PLN PTS Bugs: Bogomips: 6784.18 clFlush size: 64 cache_alignment : 64 address sizes : 39 bits physical, 48 bits virtual power management:Copy the code

Graphical representation of results:

I tried 1000 concurrent requests, but APacheBench timed out due to many sockets being opened. I tried Httperf and here is the result:

voodooattack@voodooattack:~$ httperf --port=3000 --num-conns=10000 --rate=1000
httperf --client=0/1 --server=localhost --port=3000 --uri=/ --rate=1000 --send-buffer=4096 --recv-buffer=16384 --num-conns=10000 --num-calls=1
httperf: warning: open file limit > FD_SETSIZE; limiting max. # of open files to FD_SETSIZE
Maximum connect burst length: 262

Total: connections 9779 requests 9779 replies 9779 test- Duration 10.029s Connection rate: 975.1 CONN /s (1.0ms /conn, <=1022 Concurrent connections) Connection time [ms]: Min 0.5 AVg 337.9 Max 7191.8 median 79.5 stddev 848.1 Connection time [ms]: Connect 207.3 Connection Length [replies/conn]: 1.000 Request rate: 975.1 req/s (1.0ms /req) Request size [B]: 62.0 Reply rate [replies/s]: min 903.5 AVG 974.6 Max 1045.7 STddev 100.5 (2 samples) Reply time [ms]: Reply size [B]: Header 89.0 content 8660.0 Footer 2.0 (total 8751.0) Reply status: 1xx=0 2xx=9779 3xx=0 4XX =0 5XX =0 CPU time [s]: user 0.35 system 9.67 (user 3.5% system 96.4% total 99.9%) Net I/O: 8389.9 KB/s (68.7*10^6 BPS) Errors: Total 221 client-timo 0 socket-timo 0 connrefused 0 connreset 0 Errors: fd-unavail 221 addrunavail 0 ftab-full 0 other 0Copy the code

As you can see, it still works. Although some connections will time out due to stress. I’m still working on the cause of the problem.

The source code for the project is available on GitHub.

The original address: https://dev.to/voodooattack/introducing-nexusjs-a-multi-threaded-javascript-run-time-3g6