Descripción detallada de la biblioteca de ejecución de hilos Nexus.js en JavaScript

Primero, si no estás familiarizado con este proyecto, te recomiendo que leas una serie de artículos que escribí anteriormente. Si no quieres leerlos, no te preocupes. Aquí también se abordarán esos temas.

Ahora, comencemos.

El año pasado, comencé a implementar Nexus.js, que es una base en Webkit/Biblioteca de JavaScript para servidores JavaScript con servicio de multihilo en el núcleo de JavaScript. Durante un tiempo, renuncié a hacer esto debido a razones que no podía controlar, principalmente: no podía mantenerme trabajando por mucho tiempo.

Por lo tanto, comencemos discutiendo la arquitectura de Nexus y cómo funciona.

Bucle de eventos

No hay bucle de eventos

Hay un grupo de hebras con objetos de tareas (sin bloqueo)

Cada vez que se llama a setTimeout o setImmediate o se crea una Promise, la tarea se coloca en la cola de tareas.

Cada vez que se planifica una tarea, la primera hebra disponible selecciona la tarea y la ejecuta.

Procesa Promises en el núcleo del CPU. La llamada a Promise.all() resolverá de manera paralela las Promises.

ES6

Soporte para async/await, y se recomienda su uso

Soporte para for await(...)

Soporte para desestructuración

Soporte para async try/catch/finally

Módulo

No admite CommonJS. (require(...) y module.exports)

Todos los módulos utilizan ES6del import/la sintaxis de export

Soporte para importación dinámica a través de import('file-o-packge').then(...)

Soporte para import.meta, por ejemplo: import.meta.filename y import.meta.dirname, entre otros

Funcionalidades adicionales: soporte para importar directamente desde la URL, por ejemplo:

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

EventEmitter

Nexus implementa la clase EventEmitter basada en Promise

El gestor de eventos se ordena en todas las hebras y ejecuta el procesamiento paralelo.

El valor de retorno de EventEmitter.emit(...) es un Promise, que se puede resolver en un array de valores devueltos por el procesador de eventos.

por ejemplo:

class EmitterTest extends Nexus.EventEmitter {
 constructor() {
  super();
  for(let i = 0; i < 4; i++)
   this.on('test', value => { console.log(`fue disparado el 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('primer test terminado!');
 await test.emit('test', { payload: 'test 2});
 console.log('segundo test terminado!');
 const values = await test.emit('returns-a-value', 10);
 console.log('tercer test terminado, los valores devueltos son:'); console.inspect(values);
}
start().catch(console.error);

I/O

Todos los datos de entrada/Las salidas se completan a través de tres operadores: Device, Filter y Stream.

Todos los datos de entrada/Los operadores de salida implementan la clase EventEmitter

Para utilizar Device, debe crear un ReadableStream o WritableStream sobre Device

Para operar con los datos, puede agregar Filters a ReadableStream o WritableStream.

Finalmente, utilice source.pipe(...destinationStreams) y luego espere a source.resume() para procesar los datos.

Todos los datos de entrada/Las operaciones de salida se completan utilizando el objeto ArrayBuffer.

Filter intentó utilizar el método process(buffer) para procesar los datos.

por ejemplo: usar2se generan 35 archivos de salida independientes que contienen UTF-8convertir a UTF6。

const startTime = Date.now();
 try {
  const dispositivo = 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('enrutando...');
  stream.pipe(...wstreams);
  console.log('transmitiendo...');
  await stream.resume();
  await stream.close();
  await Promise.all(wstreams.map(stream => stream.close()));
  console.log(`finalizado en ${(Date.now * startTime) / 1000} segundos!`);
 }
  console.error('Ocurrió un error: ', e);
 }
}
start().catch(console.error);

TCP/UDP

Nexus.js proporciona una clase Acceptor, responsable de enlazar la dirección IP/Puerto y escucha de conexiones

Cada vez que se recibe una solicitud de conexión, se desencadena el evento connection y se proporciona un dispositivo Socket.

Cada instancia de Socket es bidireccional./Dispositivo O.

Puedes usar ReadableStream y WritableStream para operar con Socket.

El ejemplo más básico: (enviar “Hola Mundo” al cliente)

const acceptor = new Nexus.Net.TCP.Acceptor();
let count = 0;
acceptor.on('connection', (socket, endpoint) => {
 const connId = count++;
 console.log(`conexión #${connId} desde ${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(`obtuve el mensaje: ${buffer}`));
 rstream.resume().catch(e => console.log(`el cliente #${connId} en ${endpoint.address}:${endpoint.port} se desconectó!`));
 console.log(`enviando saludo a #${connId}!`);
 wstream.write(buffer);
});
acceptor.bind('127.0.0.1', 10000);
acceptor.listen();
console.log('server ready');

Http

Nexus proporciona una clase Nexus.Net.HTTP.Server, que básicamente hereda de TCPAcceptor

algunas interfaces básicas

Cuando el servidor ha completado la解析传入连接的基本Http cabezal/Durante la verificación, se utilizará la conexión y la misma información para desencadenar el evento connection

Cada instancia de conexión tiene un objeto request y response. Estos son las entradas/dispositivo de salida。

Puedes construir ReadableStream y WritableStream para manipular request/response。

Si conectas a un objeto Response a través de un tubo, el flujo de entrada utilizará el modo de codificación en bloques. De lo contrario, puedes usar response.write() para escribir una cadena convencional.

Ejemplo complejo: (servidor HTTP básico y codificación de bloques, detalles omitidos)}

....
/**
 * Creates an input stream from a path.
 * @param path
 * @returns {Promise<ReadableStream>}
 */
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)
   // Si no se encuentra, lance NotFound.
   throw new NotFoundError(path);
 }
  if (e.code)
   throw e;
  throw new NotFoundError(path);
 }
 try {
  // Primero, creamos un dispositivo.
  const fileDevice = new Nexus.IO.FilePushDevice(filePath);
  // Luego devolvemos un nuevo Nexus.IO.ReadableStream creado usando nuestro dispositivo de origen.
  return new Nexus.IO.ReadableStream(fileDevice);
 }
  throw new InternalServerError(e.message);
 }
}
/**
 * Contador de conexiones.
 */
let connections = 0;
/**
 * Crear un nuevo servidor HTTP.
 * @type {Nexus.Net.HTTP.Server}
 */
const server = new Nexus.Net.HTTP.Server();
// Un error del servidor significa que se produjo un error mientras el servidor estaba escuchando conexiones.
// Podemos ignorar la mayoría de estos errores, de todos modos los mostramos.
server.on('error', e => {
 console.error(FgRed + Brillante + 'Error del servidor: ' + e.message + '\n' + e.stack, Reset);
});
/**
 * Escuchar conexiones.
 */
server.on('connection', async (connection, peer) => {
 // Comienza con un ID de conexión de 0, incrementa con cada nueva conexión.
 const connId = connections++;
 // Grabar la hora de inicio para esta conexión.
 const startTime = Date.now();
 // La desestructuración es compatible, ¿por qué no usarla?
 const { request, response } = connection;
 // Analizar las partes de la URL.
 const { path } = parseURL(request.url);
 // Aquí almacenaremos cualquier error que ocurra durante la conexión.
 const errors = [];
 // inStream es nuestra fuente de archivo ReadableStream, outStream es nuestra respuesta (dispositivo) envuelta en un WritableStream.
 let inStream, outStream;
 try {
  // Registrar la solicitud.
  console.log(`> #${FgCyan + connId + Reiniciar} ${Brillante + direcciónDelPar:${puertoDelPar} + Reiniciar} ${
   FgGreen + request.method + Reset} "${FgYellow}${path}${Reset}"`, Reset);
  // Establecer la cabecera 'Server'.
  response.set('Server', `nexus.js/0.1.1`);
  // Crear nuestro flujo de entrada.
  inStream = await createInputStream(path);
  // Crear nuestro flujo de salida.
  outStream = new Nexus.IO.WritableStream(response);
  // Conectar todos los eventos `error`, agregar cualquier error a nuestro array `errors`.
  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); });
  // Establecer el tipo de contenido y el estado de la solicitud.
  respuesta
   .establecer('Contenido-Type', mimeType(path))
   .estado(200);
  // Conectar la entrada a la salida(s).
  const desconectar = entradaFlujo.pipe(salidaFlujo);
  try {
   // Reanudar nuestro flujo de archivo, esto hace que el flujo cambie a la codificación de trozos HTTP.
   // Esto devolverá una promesa que solo se resolverá después de que se escriba el último byte (trozo HTTP).
   esperar que entradaFlujo.reanudar();
  }
   // Capturar cualquier error que ocurra durante el flujo.
   errores.púsh(e);
  }
  // Desconectar todos los callbacks creados por `.pipe()`.
  volver a conectar();
 }
  // Si se produjo un error, púshalo al array.
  errores.púsh(e);
  // Establecer el tipo de contenido, estado y escribir un mensaje básico.
  respuesta
   .establecer('Contenido-Tipo', 'texto/plano')
   .estado(e.código || 500)
   .enviar(e.mensaje || 'Se ha producido un error.');
 }
  // Cerrar los flujos manualmente. Esto es importante porque podríamos quedarnos sin handles de archivo de otro modo.
  si (entradaFlujo)
   esperar que entradaFlujo.cierre();
  si (salidaFlujo)
   esperar que salidaFlujo.cierre();
  // Cerrar la conexión, no tiene efecto real con mantener.-conexiones activas.
  esperar que conexión.cierre();
  // Obtener el estado de la respuesta.
  let estado = respuesta.estado();
  // Determinar qué color mostrar en la terminal.
  const coloresDelEstado = {
   '200': Brillante + FgGreen, // Verde para 200 (OK),
   '404': Brillante + FgYellow, // Amarillo para 404 (No encontrado)
   '500': Brillante + FgRed // Rojo para 500 (Error Interno del Servidor)
  };
  let colorDelEstado = coloresDelEstado[estado];
  if (colorDelEstado)
   estado = colorDelEstado + estado + Reiniciar;
  // Registrar la conexión (y el tiempo para completar) en la consola.
  console.log(`< #${FgCyan + connId + Reiniciar} ${Brillante + direcciónDelPar:${puertoDelPar} + Reiniciar} ${
   FgGreen + request.method + Reiniciar} "${FgYellow}${path}${Reiniciar}" ${estado} ${(Date.now() * startTime)}ms` +
   (errors.length ? " " + FgRed + Brillante + errors.map(error => error.message).join(', ') + Reiniciar : Reiniciar));
 }
});
/**
 * IP y puerto para escuchar.
 */
const ip = '0.0.0.0', port = 3000;
/**
 * Si se debe o no establecer la bandera `reuse`. (opcional, predeterminado=false)
 */
const portReuse = true;
/**
 * Número máximo permitido de conexiones concurrentes. El valor predeterminado es 128 en mi sistema. (opcional, específico del sistema)
 * @type {number}
 */
const maxConcurrentConnections = 1000;
/**
 * Enlazar la dirección y el puerto seleccionados.
 */
server.bind(ip, port, portReuse);
/**
 * Comenzar a escuchar solicitudes.
 */
server.listen(maxConcurrentConnections);
/**
 * ¡Disfruta del streaming!
 */
console.log(FgGreen + `Nexus.js HTTP server listening at ${ip}:${port}` + Reiniciar);

Benchmark

Creo que ya he cubierto todo lo que se ha implementado hasta ahora. Entonces, ahora hablemos de rendimiento.

Este es el benchmark actual del servidor HTTP mencionado anteriormente, hay100 conexiones concurrentes y un total de10000 solicitudes:

Este es ApacheBench, Versión 2.3 <$Revisión: 1796539 $>
Derechos de autor 1996 Adam Twiss, Zeus Technology Ltd, http://www.zeustech.net/
Licenciado a la Fundación de Software Apache, http://www.apache.org/
Medición de localhost (sea paciente).....hecho
Software del servidor:    nexus.js/0.1.1
Nombre del host del servidor:    localhost
Puerto del servidor:      3000
Ruta del documento:     /
Longitud del documento:    8673 bytes
Nivel de concurrencia:   100
Tiempo tomado para las pruebas:  9.991 segundos
Solicitudes completadas:   10000
Solicitudes fallidas:    0
Total transferido:   87880000 bytes
HTML transferido:    86730000 bytes
Solicitudes por segundo:  1000.94 [#/seg] (promedio)
Tiempo por solicitud:    99.906 [ms] (promedio)
Tiempo por solicitud:    0.999 [ms] (media, a través de todas las solicitudes concurrentes)
Tasa de transferencia:     8590.14 [Kbytes/seg] recibido
Tiempo de conexión (ms)
       min media[+/-sd] mediana max
Conexión:    0  0  0.1   0    1
Procesando:   6  99 36.6   84   464
Esperando:    5  99 36.4   84   463
Total:     6 100 36.6   84   464
Porcentaje de solicitudes servidas en un tiempo determinado (ms)
 50%   84
 66%   97
 75%  105
 80%  112
 90%  134
 95%  188
 98%  233
 99%  238
 100%  464 (solicitud más larga)

por segundo1000 solicitudes. En un viejo i7encima, se ejecutó incluyendo este software de prueba de rendimiento, uno que ocupó5G de memoria IDE, así como el servidor en sí.

voodooattack@voodooattack:~$ cat /proc/cpuinfo 
procesador: 0
id proveedor: GenuineIntel
familia cpu: 6
modelo: 60
nombre modelo: Intel(R) Core(TM) i7-4770 CPU @ 3.40GHz
pasos: 3
microcode: 0x22
MHz cpu: 3392.093
tamaño caché: 8192 KB
id físico: 0
hermanos: 8
id núcleo: 0
núcleos cpu: 4
apicid: 0
apicid inicial: 0
fpu: sí
fpu_exception: sí
nivel cpuid: 13
wp: sí
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
tamaño clflush : 64
alineación de caché : 64
tamaños de dirección : 39 bits físicos, 48 bits virtuales
gestión de energía:

Intenté1000 solicitudes concurrentes, pero ApacheBench se timeouts debido a que se abrieron muchos sockets. Intenté httperf, aquí están los resultados:

voodooattack@voodooattack:~$ httperf --puerto=3000 --número-conexiones=10000 --tasa=1000
httperf --cliente=0/1 --servidor=localhost --puerto=3000 --uri=/ --tasa=1000 --enviar-búfer=4096 --recibir-búfer=16384 --número-conexiones=10000 --número-llamadas=1
httperf: advertencia: límite de archivos abiertos > FD_SETSIZE; limitando el número máximo de archivos abiertos a FD_SETSIZE
Longitud máxima de ráfaga de conexión: 262
Total: conexiones 9779 solicitudes 9779 respuestas 9779 prueba-duración 10.029 s
Tasa de conexión: 975.1 conn/s (1.0 ms/conn, <=1022 conexiones concurrentes)
Tiempo de conexión [ms]: mínimo 0.5 prom 337.9 máx 7191.8 mediana 79.5 stddev 848.1
Tiempo de conexión [ms]: conectar 207.3
Longitud de conexión [respuestas]/conn]: 1.000
Tasa de solicitud: 975.1 solicitud/s (1.0 ms/solicitud)
Tamaño de solicitud [B]: 62.0
Tasa de respuesta [respuestas/s]: mín 903.5 prom 974.6 máx 1045.7 stddev 100.5 (2 muestra)
Tiempo de respuesta [ms]: respuesta 129.5 transferencia 1.1
Tamaño de respuesta [B]: encabezado 89.0 contenido 8660.0 pie de página 2.0 (total 8751.0)
Estado de respuesta: 1xx=0 2xx=9779 3xx=0 4xx=0 5xx=0
Tiempo de CPU [s]: usuario 0.35 sistema 9.67 (usuario 3.5% sistema 96.4% total 99.9%)
Net I/O: 8389.9 KB/s (68.7*10^6 bps)
Errores: total 221 client-timo 0 socket-timo 0 connrefused 0 connreset 0
Errores: fd-unavail 221 addrunavail 0 ftab-full 0 other 0

Como puedes ver, aún funciona. A pesar de la presión, algunos enlaces pueden superar el tiempo de espera. Sigo investigando la causa de este problema.

Esto es todo el contenido sobre el aprendizaje de Nexus.js, si tienes alguna pregunta, puedes dejar un comentario a continuación para discutir, gracias por tu apoyo a la serie de tutoriales de grito.

Declaración: el contenido de este artículo se ha obtenido de la red, pertenece al propietario original, ha sido contribuido y subido por los usuarios de Internet de manera autónoma. Este sitio no posee los derechos de propiedad, no ha sido editado artificialmente y no asume ninguna responsabilidad legal relacionada. Si encuentra contenido sospechoso de violación de derechos de autor, por favor envíe un correo electrónico a: notice#oldtoolbag.com (al enviar un correo electrónico, reemplace # con @ para denunciar y proporcione evidencia relevante. Una vez confirmado, este sitio eliminará inmediatamente el contenido sospechoso de infracción de derechos de autor.)