Подробное объяснение библиотеки многнитевых потоков Nexus.js на JavaScript

Если вы не знакомы с этим проектом, рекомендуется сначала прочитать серию статей, которые я написал ранее. Если вы не хотите читать их, не волнуйтесь. В этом тексте также будут затронуты эти темы.

Теперь давайте начнем.

В прошлом году я начал реализовывать Nexus.js, это библиотека для выполнения JavaScript на сервере, основанная на ядре Webkit/JavaScript и поддерживающая многопоточность. В какой-то момент я оставил это, по причинам, которые я не могу контролировать и не хочу обсуждать здесь, в основном: я не мог заставить себя работать долго.

Итак, давайте начнем с обсуждения архитектуры Nexus и того, как он работает.

Цикл событий

Нет событий цикла

Существует пул нит с задачами (без блокировки)

Каждый раз, когда вызывается setTimeout или setImmediate или создается Promise, задача добавляется в очередь задач.

Каждый раз, когда запланирована задача, первая доступная нить выбирает задачу и выполняет её.

Обработка Promise на ядре CPU. Вызов Promise.all() решает Promise параллельно.

ES6

Поддержка async/await и рекомендуется к использованию

Поддержка for await(...)

Поддержка деструктуризации

Поддержка async try/catch/finally

Модуль

Не поддерживается CommonJS. (require(...) и module.exports)

Все модули используют синтаксис import/export ES6

Поддержка динамического импорта через import('file-or-packge').then(...)

Поддержка import.meta, например: import.meta.filename и import.meta.dirname и т.д.

Дополнительные возможности: поддержка импорта напрямую из URL, например:

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

EventEmitter

Nexus реализует класс EventEmitter на основе Promise

Обработчики событий сортируются по всем потокам и выполняются параллельно.

Возвратное значение EventEmitter.emit(...) - это Promise, которая может быть решена в виде массива значений, возвращаемых обработчиками событий.

например:

class EmitterTest extends Nexus.EventEmitter {
 constructor() {
  super();
  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);

I/O

Все вводно-выводные операции выполняются через три атома: Device, Filter и Stream.

Все вводно-выводные атомы реализуют класс EventEmitter.

Чтобы использовать Device, вам нужно создать ReadableStream или WritableStream поверх Device.

Чтобы управлять данными, добавьте Filters в ReadableStream или WritableStream.

В конце концов, используйте source.pipe(...destinationStreams), затем ждите source.resume() для обработки данных.

Все операции ввода/вывода выполняются с использованием объекта ArrayBuffer.

Filter пытался использовать метод process(buffer) для обработки данных.

Например: использовать два независимых выводных файла для преобразования UTF-8 в UTF6.

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(`завершено за ${(Date.now() * startTime) / 1000} секунд!`);
 }
  console.error('Произошла ошибка: ', e);
 }
}
start().catch(console.error);

TCP/UDP

Nexus.js предоставляет класс Acceptor, который отвечает за привязку IP-адреса/порта и прослушивание соединений

При каждом получении запроса на соединение событие connection инициируется и предоставляет устройство Socket.

Каждый экземпляр Socket является полнодуплексным устройством ввода/вывода.

Вы можете использовать ReadableStream и WritableStream для работы с Socket.

Самый базовый пример: (отправка “Hello World” клиенту)

const acceptor = new Nexus.Net.TCP.Acceptor();
let count = 0;
acceptor.on('connection', (socket, endpoint) => {
 const connId = count++;
 console.log(`соединение #${connId} от ${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(`получено сообщение: ${buffer}`));
 rstream.resume().catch(e => console.log(`клиент #${connId} в ${endpoint.address}:${endpoint.port} отключился!`));
 console.log(`отправка приветствия клиенту #${connId}!`);
 wstream.write(buffer);
});
acceptor.bind('127.0.0.1', 10000);
acceptor.listen();
console.log('сервер готов');

Http

Nexus предоставляет класс Nexus.Net.HTTP.Server, который в основном наследует от TCPAcceptor

Некоторые базовые интерфейсы

Когда серверная сторона завершает анализ/проверку основных HTTP-заголовков传入ного подключения, с помощью этой же информации и подключения вызывается событие connection

Каждый экземпляр подключения имеет объект запроса и ответа. Это входные/выходные устройства.

Вы можете создать ReadableStream и WritableStream для манипулирования запросом/ответом.

Если вы подключаете через трубопровод к объекту Response, входной поток будет использовать режим блокового кодирования. В противном случае, вы можете использовать response.write() для записи обычной строки.

Сложный пример: (базовый HTTP-сервер и блоковое кодирование, детали опущены)

....
/**
 * Создает поток ввода из пути.
 * @param path
 * @returns {Promise<ReadableStream>}
 */
async function createInputStream(path) {
 if (path.startsWith('/')) // Если путь начинается с '/', пропустите его.
  path = path.substr(1);
 if (path.startsWith('.')) // Если путь начинается с '.', отклоните его.
  throw new NotFoundError(path);
 if (path === '/' || !path) // Если путь пуст, установите его в index.html.
  path = 'index.html';
 /**
  * `import.meta.dirname` и `import.meta.filename` заменяют старые CommonJS `__dirname` и `__filename`.
  */
 const filePath = Nexus.FileSystem.join(import.meta.dirname, 'server_root', path);
 try {
  // Получите статус целевого пути.
  const {type} = await Nexus.FileSystem.stat(filePath);
  if (type === Nexus.FileSystem.FileType.Directory) // Если это директория, верните её 'index.html'
   return createInputStream(Nexus.FileSystem.join(filePath, 'index.html'));
  else if (type === Nexus.FileSystem.FileType.Unknown || type === Nexus.FileSystem.FileType.NotFound)
   // Если не найдено, выбросьте NotFound.
   throw new NotFoundError(path);
 }
  if (e.code)
   throw e;
  throw new NotFoundError(path);
 }
 try {
  // Сначала мы создаем устройство.
  const fileDevice = new Nexus.IO.FilePushDevice(filePath);
  // Затем мы возвращаем новый ReadableStream, созданный с использованием нашего источника устройства.
  return new Nexus.IO.ReadableStream(fileDevice);
 }
  throw new InternalServerError(e.message);
 }
}
/**
 * Счетчик подключений.
 */
let connections = 0;
/**
 * Создать новый HTTP сервер.
 * @type {Nexus.Net.HTTP.Server}
 */
const server = new Nexus.Net.HTTP.Server();
// Ошибка сервера означает, что произошла ошибка во время прослушивания сервером подключений.
// Такие ошибки можно в основном игнорировать, мы их все равно отображаем.
server.on('error', e => {
 console.error(FgRed + Bright + 'Ошибка сервера: ' + e.message + '\n' + e.stack, Reset);
});
/**
 * Слушать подключения.
 */
server.on('connection', async (connection, peer) => {
 // Начать с идентификатора связи 0, увеличивать каждый раз при новой связи.
 const connId = connections++;
 // Записать время начала для этой связи.
 const startTime = Date.now();
 // Деструктурирование поддерживается, почему бы не использовать его?
 const { request, response } = connection;
 // Парсинг частей URL.
 const { path } = parseURL(request.url);
 // Здесь мы будем хранить любые ошибки, которые могут возникнуть во время подключения.
 const errors = [];
 // inStream是我们的 Читаемый поток файлового источника, outStream是我们的 ответ (устройство), обернутый в WritableStream.
 let inStream, outStream;
 try {
  // Записать запрос.
  console.log(`> #${FgCyan + connId + Reset} ${Bright + peer.address}:${peer.port + Reset} ${
   FgGreen + request.method + Reset} "${FgYellow}${path}${Reset}"`, Reset);
  // Установить заголовок 'Server'.
  response.set('Server', `nexus.js/0.1.1`);
  // Создать наш поток ввода.
  inStream = await createInputStream(path);
  // Создать наш поток вывода.
  outStream = new Nexus.IO.WritableStream(response);
  // Подключить все события `error`, добавить любые ошибки в наш массив `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); });
  // Установите тип содержимого и статус запроса.
  response
   .set('Content-Type', mimeType(path))
   .status(200);
  // Подключите вход к выходу(ам).
  const disconnect = inStream.pipe(outStream);
  try {
   // Восстановите наш файловый поток, это вызывает переключение потока на HTTP chunked encoding.
   // Это вернёт promise, которая будет разрешена только после записи последнего байта (HTTP chunk).
   await inStream.resume();
  }
   // Уловите все ошибки, которые могут произойти во время потоковой передачи.
   errors.push(e);
  }
  // Разорвите все callbacks, созданные с помощью `.pipe()`.
  return disconnect();
 }
  // Если произошла ошибка, добавьте её в массив.
  errors.push(e);
  // Установите тип содержимого, статус и напишите базовое сообщение.
  response
   .set('Content-Type', 'text/plain')
   .status(e.code || 500)
   .send(e.message || 'Произошла ошибка.');
 }
  // Прекратите потоки вручную. Это важно, так как мы можем исчерпать файловые дескрипторы.
  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} ${(Date.now() * startTime)}ms` +
   (errors.length ? " " + FgRed + Bright + errors.map(error => error.message).join(', ') + Reset : Reset));
 }
});
/**
 * IP и порт для прослушивания.
 */
const ip = '0.0.0.0', port = 3000;
/**
 * Установить ли флаг `reuse`. (необязательное, по умолчанию=false)
 */
const portReuse = true;
/**
 * Максимальное количество разрешенных параллельных соединений. По умолчанию 128 на моей системе. (необязательное, система специфичное)
 * @type {number}
 */
const maxConcurrentConnections = 1000;
/**
 * Привязать выбранный адрес и порт.
 */
server.bind(ip, port, portReuse);
/**
 * Начать прослушивание запросов.
 */
server.listen(maxConcurrentConnections);
/**
 * радостное стриминг!
 */
console.log(FgGreen + `Nexus.js HTTP server listening at ${ip}:${port}` + Reset);

Базовый уровень

Я думаю, что я уже охватил все, что было реализовано до сих пор. Давайте теперь поговорим о производительности.

Это текущий базовый уровень Http сервера, с 100 параллельными соединениями и всего 10000 запросами:

Это ApacheBench, Версия 2.3 <$Revision: 1796539 $>
Авторские права 1996 Адам Твис, Zeus Technology Ltd, http://www.zeustech.net/
Лицензия принадлежит Фонду программного обеспечения Apache, http://www.apache.org/
Бenchmarking localhost (подождите).....готово
Софт сервера:    nexus.js/0.1.1
Имя хоста сервера:    localhost
Порт сервера:      3000
Путь документа:     /
Длина документа:    8673 байт
Уровень параллелизма:   100
Время на тестирование:  9.991 секунд
Полностью выполненные запросы:   10000
Неуспешные запросы:    0
Общее количество переданного:   87880000 байт
Переданный HTML:    86730000 байт
Запросов в секунду:  1000.94 [#/сек] (среднее)
Время на каждый запрос:    99.906 [мс] (среднее)
Время на каждый запрос:    0.999 [мс] (среднее, для всех параллельных запросов)
Скорость передачи:     8590.14 [Кбайт/с] получено
Время соединения (мс)
       мин среднее[+/-sd] медиана  макс
Соединение:    0  0  0.1   0    1
Обработка:   6  99 36.6   84   464
В ожидании:    5  99 36.4   84   463
Общее количество:     6 100 36.6   84   464
Процент запросов, обслуженных в течение определенного времени (мс)
 50%   84
 66%   97
 75%  105
 80%  112
 90%  134
 95%  188
 98%  233
 99%  238
 100%  464 (самый длинный запрос)

Каждую секунду 1000 запросов. На старом i7, на котором работает этот тестовый софт, занимающий 5 ГБ оперативной памяти IDE, и сам сервер.

voodooattack@voodooattack:~$ cat /proc/cpuinfo 
процессор: 0
идентификатор производителя: GenuineIntel
семейство cpu: 6
модель: 60
название модели: Intel(R) Core(TM) i7-4770 CPU @ 3.40GHz
шаг: 3
микрокод: 0x22
cpu MHz: 3392.093
размер кэша: 8192 КБ
физический идентификатор: 0
братья: 8
идентификатор ядра: 0
число ядер cpu: 4
apicid: 0
начальный apicid: 0
fpu: да
исключение fpu: да
уровень cpuid: 13
wp: да
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: 64
Алиасирование кэша: 64
Размеры адресов: 39 бит физические, 48 бит виртуальные
Управление питанием:

Я попробовал 1000 одновременных запросов, но ApacheBench из-за большого количества открытых套овых превысил время ожидания. Я попробовал httperf, вот результат:

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: предупреждение: ограничение на количество открытых файлов > FD_SETSIZE; ограничение максимального количества открытых файлов до FD_SETSIZE
Максимальная длина всплеска подключений: 262
Общее количество: подключений 9779 запросов 9779 ответов 9779 время тестирования 10.029 с
Скорость подключения: 975.1 подключений/с (1.0 мс/подключение, <=1022 одновременных подключений)
Время подключения [мс]: мин 0.5 avg 337.9 max 7191.8 median 79.5 stddev 848.1
Время подключения [мс]: соединение 207.3
Длина подключения [ответов/подключение]: 1.000
Скорость запросов: 975.1 запросов/с (1.0 мс/запрос)
Размер запроса [Б]: 62.0
回复速率[回复/秒]：最小 903.5 平均 974.6 最大 1045.7 标准差 100.5 (2个样本)
回复时间[ms]：响应 129.5 传输 1.1
回复大小[B]：头部 89.0 内容 8660.0 尾部 2.0 (总计 8751.0)
回复状态：1xx=0 2xx=9779 3xx=0 4xx=0 5xx=0
CPU时间[s]：用户 0.35 系统 9.67 (用户 3.5% 系统 96.4% 总计 99.9%)
网络I/O：8389.9 KB/s (68.7*10^6 bps)
错误：总计 221 客户端超时 0 套接字超时 0 连接被拒绝 0 连接重置 0
错误：fd-unavail 221 addrunavail 0 ftab-full 0 other 0

正如你所看到的，它仍然能工作。尽管由于压力，有些连接会超时。我仍在研究导致这个问题的原因。

以上就是关于Nexus.js学习知识的全部内容，大家有问题可以在下方留言讨论，感谢对呐喊教程的支持。

声明：本文内容来自网络，版权归作者所有，内容由互联网用户自发贡献自行上传，本网站不拥有所有权，未进行人工编辑处理，也不承担相关法律责任。如果您发现涉嫌版权的内容，欢迎发送邮件至：notice#oldtoolbag.com（发邮件时，请将#更换为@）进行举报，并提供相关证据，一经查实，本站将立即删除涉嫌侵权内容。