Detalhamento da biblioteca de execução multithreading Nexus.js em JavaScript

First, if you are not familiar with this project, it is recommended to read a series of articles written before. If you don't want to read these, don't worry. This will also involve that content.

Now, let's get started.

Last year, I started implementing Nexus.js, which is based on Webkit/A multi-threaded server-side JavaScript runtime library for the JavaScript core. For a while, I gave up on doing this due to some reasons beyond my control, which I won't discuss here, mainly: I can't get myself to work for a long time.

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

Event loop

No event loop

There is a thread pool with (lock-free) task objects

Each call to setTimeout or setImmediate or creation of a Promise will queue the task to the task queue.

When a scheduled task is planned, the first available thread will select the task and execute it.

Processing Promises on CPU cores. The call to Promise.all() will resolve Promises in parallel.

ES6

Support for async/await, and it is recommended to use

Support for for await(...)

Support for destructuring

Support for async try/catch/finally

Module

Does not support CommonJS. (require(...) and module.exports)

All modules use ES6of import/Export syntax

Support for dynamic imports through import('file-or-packge').then(...)

Support for import.meta, for example: import.meta.filename and import.meta.dirname, etc.

Additional features: support direct import from URL, for example:

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

EventEmitter

Nexus implements an EventEmitter class based on Promise

The event handler is sorted across all threads and executed in parallel.

O valor retornado por EventEmitter.emit(...) é um Promise, que pode ser resolvido em um array de valores retornados pelo manipulador de eventos.

por exemplo:

class EmitterTest extends Nexus.EventEmitter {
 constructor() {
  super();
  for(let i = 0; i < 4; i++)
   this.on('test', value => { console.log(`acionado teste ${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('primeiro teste concluído!');
 await test.emit('test', { payload: 'test 2});
 console.log('segundo teste concluído!');
 const values = await test.emit('returns-a-value', 10);
 console.log('terceiro teste concluído, os valores retornados são:'); console.inspect(values);
}
start().catch(console.error);

I/O

todos os inputs/todas as saídas são realizadas por três primitivos: Device, Filter e Stream.

todos os inputs/os primitivos de saída implementam a classe EventEmitter

Para usar Device, você precisa criar um ReadableStream ou WritableStream sobre o Device

Para operar dados, você pode adicionar Filters a um ReadableStream ou WritableStream.

Por fim, use source.pipe(...destinationStreams) e aguarde source.resume() para processar os dados.

todos os inputs/todas as operações de saída são realizadas usando objetos ArrayBuffer.

Filter tentou usar o método process(buffer) para processar os dados.

por exemplo: usando2um arquivo de saída independente contém UTF-8convertendo para UTF6。

const startTime = Date.now();
 tentar {
  const device = new Nexus.IO.FilePushDevice('enwik}8');
  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(`concluído em ${(Date.now() * startTime) / 1000} segundos!`);
 }
  console.error('Ocorreu um erro: ', e);
 }
}
start().catch(console.error);

TCP/UDP

O Nexus.js fornece uma classe Acceptor, responsável por vincular o endereço IP/Porta e escuta de conexão

Cada vez que um pedido de conexão é recebido, o evento connection é acionado e fornece um dispositivo Socket.

Cada instância de Socket é bidirecional./O dispositivo.

Você pode usar ReadableStream e WritableStream para operar o Socket.

O exemplo mais básico: (enviar “Hello World” para o cliente)

const acceptor = new Nexus.Net.TCP.Acceptor();
let count = 0;
acceptor.on('connection', (socket, endpoint) => {
 const connId = count++;
 console.log(`conexão #${connId} de ${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(`obteve mensagem: ${buffer}`));
 rstream.resume().catch(e => console.log(`o cliente #${connId} em ${endpoint.address}:${endpoint.port} desconectou!`));
 console.log(`enviando saudação para #${connId}!`);
 wstream.write(buffer);
});
acceptor.bind('127.0.0.1', 10000);
acceptor.listen();
console.log('server ready');

Http

O Nexus fornece a classe Nexus.Net.HTTP.Server, que basicamente herda de TCPAcceptor

Alguns interfaces básicos

Quando o servidor completar a análise básica dos cabeçalhos Http recebidos/Durante a verificação, o evento connection será acionado usando a conexão e as mesmas informações

Cada instância de conexão possui um objeto request e response. Esses são os inputs/Dispositivo de saída.

Você pode construir ReadableStream e WritableStream para manipular request/response。

Se você conectar a um objeto Response por meio de um canal, o fluxo de entrada será codificado em modo de bloco. Caso contrário, você pode usar response.write() para escrever uma string convencional.

Exemplo complexo: (servidor HTTP básico e codificação de bloco, detalhes omitidos)}

....
/**
 * Cria um fluxo de entrada a partir de um caminho.
 * @param path
 * @returns {Promise<ReadableStream>}
 */
async function createInputStream(path) {
 if (path.startsWith('/)) // Se começar com '/, omita-o.
  path = path.substr(1);
 if (path.startsWith('.')) // Se começar com '.', rejeite-o.
  throw new NotFoundError(path);
 if (path === '/|| !path) // Se estiver vazio, defina como index.html.
  path = 'index.html';
 /**
  * `import.meta.dirname` e `import.meta.filename` substituem os antigos `__dirname` e `__filename` do CommonJS.
  */
 const filePath = Nexus.FileSystem.join(import.meta.dirname, 'server_root', path);
 tentar {
  // Estatística do caminho de destino.
  const {type} = await Nexus.FileSystem.stat(filePath);
  if (type === Nexus.FileSystem.FileType.Directory) // Se for um diretório, retorne seu 'index.html'
   return createInputStream(Nexus.FileSystem.join(filePath, 'index.html'));
  else if (type === Nexus.FileSystem.FileType.Unknown || type === Nexus.FileSystem.FileType.NotFound)
   // Se não for encontrado, lance NotFound.
   throw new NotFoundError(path);
 }
  if (e.code)
   throw e;
  throw new NotFoundError(path);
 }
 tentar {
  // Primeiro, criamos um dispositivo.
  const fileDevice = new Nexus.IO.FilePushDevice(filePath);
  // Então retornamos um novo ReadableStream criado usando nosso dispositivo de origem.
  return new Nexus.IO.ReadableStream(fileDevice);
 }
  throw new InternalServerError(e.message);
 }
}
/**
 * Contador de conexões.
 */
let connections = 0;
/**
 * Criar um novo servidor HTTP.
 * @type {Nexus.Net.HTTP.Server}
 */
const server = new Nexus.Net.HTTP.Server();
// Um erro do servidor significa que ocorreu um erro enquanto o servidor estava escutando conexões.
// Podemos ignorar tais erros na maioria das vezes, mas ainda assim os exibimos.
server.on('error', e => {
 console.error(FgRed + Brilhante + 'Erro do Servidor: ' + e.message + '\n' + e.stack, Reset);
});
/**
 * Escutar conexões.
 */
server.on('connection', async (connection, peer) => {
 // Iniciar com um ID de conexão de 0, incrementar com cada nova conexão.
 const connId = connections++;
 // Gravar o tempo de início para esta conexão.
 const startTime = Date.now();
 // A desestruturação é suportada, por que não usá-la?
 const { request, response } = connection;
 // Analise as partes da URL.
 const { path } = parseURL(request.url);
 // Aqui guardaremos quaisquer erros que ocorram durante a conexão.
 const errors = [];
 // inStream é nossa fonte de arquivo ReadableStream, outStream é nossa resposta (dispositivo) envolvida em um WritableStream.
 let inStream, outStream;
 tentar {
  // Registre a solicitação.
  console.log(`> #${FgCyan` + connId + Redefinir} ${Brilhante + peer.address}:${peer.port + Redefinir} ${
   FgGreen + request.method + Reset} "${FgYellow}${path}${Reset}"`, Reset);
  // Defina o cabeçalho 'Server'.
  response.set('Server', `nexus.js`)/0.1.1`);
  // Crie nosso fluxo de entrada.
  inStream = await createInputStream(path);
  // Crie nosso fluxo de saída.
  outStream = new Nexus.IO.WritableStream(response);
  // Conecte todos os eventos `error`, adicione quaisquer erros ao nosso 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); });
  // Definir o tipo de conteúdo e o status da solicitação.
  resposta
   .definir('Conteúdo-Type', mimeType(path))
   .status(200);
  // Conecte a entrada à saída(s).
  const desconectar = inStream.pipe(outStream);
  tentar {
   // Resuma nosso fluxo de arquivo, isso faz com que o fluxo mude para codificação em chunks HTTP.
   // Isso retornará uma promessa que só será resolvida após a última byte (chunk HTTP) ser escrito.
   esperar inStream.resumir();
  }
   // Capture qualquer erro que ocorra durante o streaming.
   erros.push(e);
  }
  // Desconecte todos os callbacks criados por `.pipe()`.
  retornar desconectar();
 }
  // Se ocorrer um erro, empurre-o para o array.
  erros.push(e);
  // Defina o tipo de conteúdo, status e escreva uma mensagem básica.
  resposta
   .definir('Conteúdo-Tipo', 'texto/plano')
   .status(e.codigo || 500)
   .enviar(e.mensagem || 'Ocorreu um erro.');
 }
  // Feche os streams manualmente. Isso é importante porque podemos esgotar os handles de arquivo de outra forma.
  se (inStream)
   esperar inStream.close();
  se (outStream)
   esperar outStream.close();
  // Feche a conexão, não tem efeito real com keep-conexões ativas.
  esperar conexão.close();
  // Pegar o status da resposta.
  let status = response.status();
  // Determinar qual cor deve ser exibida no terminal.
  const statusColors = {
   '200': Brilhante + FgGreen, // Verde para 200 (OK),
   '404': Brilhante + FgYellow, // Amarelo para 404 (Não Encontrado)
   '500': Brilhante + FgRed // Vermelho para 500 (Erro Interno do Servidor)
  };
  let statusColor = statusColors[status];
  if (statusColor)
   status = statusColor + status + Redefinir;
  // Registrar a conexão (e o tempo para completar) no console.
  console.log(`< #${FgCyan + connId + Redefinir} ${Brilhante + peer.address}:${peer.port + Redefinir} ${
   FgGreen + request.method + Redefinir} "${FgYellow}${path}${Redefinir}" ${status} ${(Date.now() * startTime)}ms` +
   (errors.length ? " " + FgRed + Brilhante + errors.map(error => error.message).join(', ') + Redefinir : Redefinir));
 }
});
/**
 * IP e porta para escutar.
 */
const ip = '0.0.0.0', port = 3000;
/**
 * Se deve ou não definir a bandeira `reuse`. (opcional, padrão=false)
 */
const portReuse = true;
/**
 * Máximo de conexões simultâneas permitidas. O padrão é 128 no meu sistema. (opcional, específico do sistema)
 * @type {number}
 */
const maxConcurrentConnections = 1000;
/**
 * Ligar o endereço e a porta selecionados.
 */
server.bind(ip, port, portReuse);
/**
 * Começar a ouvir solicitações.
 */
server.listen(maxConcurrentConnections);
/**
 * Streaming feliz!
 */
console.log(FgGreen + `Nexus.js HTTP server listening at ${ip}:${port}` + Reiniciar);

Benchmark

Acho que já cobri tudo que foi implementado até agora. Então, vamos falar sobre o desempenho.

Este é o benchmark atual do servidor HTTP mencionado acima, com100 conexões concorrentes e um total de10000 solicitações:

Este é o ApacheBench, Versão 2.3 <$Revisão: 1796539 $>
Direitos autorais 1996 Adam Twiss, Zeus Technology Ltd, http://www.zeustech.net/
Licenciado para The Apache Software Foundation, http://www.apache.org/
Medindo localhost (tenha paciência).....concluído
Software do servidor:    nexus.js/0.1.1
Nome do servidor:    localhost
Porta do servidor:      3000
Caminho do documento:     /
Tamanho do documento:    8673 bytes
Nível de concorrência:   100
Tempo gasto em testes:  9.991 segundos
Solicitações completas:   10000
Solicitações falhadas:    0
Total transferido:   87880000 bytes
HTML transferido:    86730000 bytes
Solicitações por segundo:  1000.94 [#/seg] (média)
Tempo por solicitação:    99.906 [ms] (média)
Tempo por solicitação:    0.999 [ms] (média, ao longo de todos os pedidos concorrentes)
Taxa de transferência:     8590.14 [Kbytes/seg] recebido
Tempos de Conexão (ms)
       mínimo média[+/-sd] médio  máx
Conexão:    0  0  0.1   0    1
Processando:   6  99 36.6   84   464
Aguardando:    5  99 36.4   84   463
Total:     6 100 36.6   84   464
Percentagem de pedidos servidos dentro de um tempo determinado (ms)
 50%   84
 66%   97
 75%  105
 80%  112
 90%  134
 95%  188
 98%  233
 99%  238
 100%  464 (pedido mais longo)

por segundo1000 pedidos. Em um i7Acima, está em execução, incluindo este software de benchmark, que ocupou5O IDE G da memória e o servidor em si.

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
tamanho clflush: 64
alinhamento da cache: 64
tamanhos de endereço: 39 bits físicos, 48 bits virtuais
gerenciamento de energia:

Tentei1000 solicitações concorrentes, mas o ApacheBench timeout devido a muitos sockets abertos. Tentei httperf, aqui estão os resultados:

voodooattack@voodooattack:~$ httperf --porta=3000 --número-conexões=10000 --taxa=1000
httperf --cliente=0/1 --servidor=localhost --porta=3000 --uri=/ --taxa=1000 --enviar-buffer=4096 --receber-buffer=16384 --número-conexões=10000 --número-chamadas=1
httperf: aviso: limite de arquivo aberto > FD_SETSIZE; limitando o número máximo de arquivos abertos a FD_SETSIZE
Comprimento máximo de explosão de conexão: 262
Total: conexões 9779 solicitações 9779 respostas 9779 teste-duração 10.029 s
Taxa de conexão: 975.1 conn/s (1.0 ms/conn, <=1022 conexões concorrentes)
Tempo de conexão [ms]: min 0.5 avg 337.9 max 7191.8 média 79.5 stddev 848.1
Tempo de conexão [ms]: conectar 207.3
Connection length [replies/conn]: 1.000
Request rate: 975.1 req/s (1.0 ms/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]: response 129.5 transfer 1.1
Reply size [B]: header 890.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 0

Como você vê, ele ainda funciona. No entanto, devido à pressão, alguns conexões podem expirar. Estou ainda estudando a causa desse problema.

Isso é tudo sobre o conteúdo de estudo de Nexus.js. Se você tiver perguntas, pode deixar comentários abaixo para discutir. Agradecemos o apoio ao tutorial Grito.

Declaração: O conteúdo deste artigo é extraído da Internet, pertence ao respectivo proprietário, foi contribuído e carregado voluntariamente pelos usuários da Internet, o site não possui direitos de propriedade, não foi editado manualmente e não assume responsabilidade legal relevante. Se você encontrar conteúdo suspeito de violação de direitos autorais, por favor, envie um e-mail para: notice#oldtoolbag.com (ao enviar e-mail, substitua # por @ para denunciar e forneça provas relevantes. Apenas após a verificação, o site deletará o conteúdo suspeito de violação de direitos autorais.)