JavaScript의 멀티스레드 런타임 라이브러리 Nexus.js 상세 설명

먼저, 이 프로젝트에 익숙하지 않다면, 이전에 쓴 일련의 기사를 먼저 읽는 것을 추천합니다. 이를 읽고 싶지 않다면 걱정하지 마세요. 이곳에도 이러한 내용이 포함됩니다.

그제서야 시작해 보겠습니다.

지난 해, Nexus.js를 구현하기 시작했습니다. 이는 Webkit을 기반으로 한 것입니다./JavaScript 커널의 멀티 스레드 서버 JavaScript 런타임 라이브러리입니다. 얼마 동안 이 작업을 포기했습니다. 원인은 제가 제어할 수 없는 몇 가지가 있었으며, 주로: 장시간 일할 수 없었습니다.

따라서, Nexus의 아키텍처와 그 작동 방식에 대해 논의하기 시작해 보겠습니다.

이벤트 루프

이벤트 루프가 없습니다.

(불릭스) 작업 객체를 가진 스레드 풀이 있습니다.

setTimeout이나 setImmediate를 호출하거나 Promise를 생성할 때마다, 작업이 작업 큐에排队됩니다.

스케줄링 작업이 있을 때마다, 첫 번째 사용 가능한 스레드가 작업을 선택하고 실행합니다.

CPU 코어에서 Promise를 처리합니다. Promise.all() 호출은 병행적으로 Promise를 해결합니다.

ES6

async를 지원합니다./await를 지원하며, 추천 사용.

for await(...)를 지원합니다.

구조화를 지원합니다.

async try를 지원합니다./catch/finally

모듈

CommonJS를 지원하지 않습니다. (require(...)와 module.exports)

모든 모듈은 ES6의 import/export 문법

import('file'를 통해 동적 임포트를 지원합니다.-또는-packge').then(...)

import.meta를 지원합니다. 예를 들어, import.meta.filename 및 import.meta.dirname 등.

추가 기능: URL에서 직접 가져오기를 지원합니다. 예를 들어:

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

EventEmitter

Nexus는 Promise를 기반으로 한 EventEmitter 클래스를 구현했습니다.

이벤트 처리 프로그램이 모든 스레드에서 정렬되고 병행 처리가 실행됩니다.

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，你需要在Device之上创建一个ReadableStream或WritableStream

要操作数据，可以将Filters添加到ReadableStream或WritableStream中。

最后，使用source.pipe(...destinationStreams)，然后等待source.resume()来处理数据。

所有的输入/输出操作都是使用ArrayBuffer对象完成的。

Filter试了了process(buffer)方法来处理数据。

例如：使用2个独立的输出文件将UTF-8转换为UTF6。

const startTime = Date.now();
 try {
  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(`완료되었습니다: ${(Date.now() * startTime) / 1000} 초!`);
 } catch (e) {
  console.error('오류가 발생했습니다: ', e);
 }
}
start().catch(console.error);

TCP/UDP

Nexus.js는 ip 주소를 바인딩하는 Acceptor 클래스를 제공합니다./포트와 연결 감시

연결 요청을 받을 때마다 connection 이벤트가 발생하고, Socket 장치를 제공합니다.

각 Socket 인스턴스는 완전한 양방향의 I/O 장치.

ReadableStream과 WritableStream을 사용하여 소켓을操作할 수 있습니다。

가장 기본적인 예제：（클라이언트에 '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(`message: ${buffer}`));
 rstream.resume().catch(e => console.log(`client #${connId} at ${endpoint.address}:${endpoint.port} disconnected!`));
 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 객체에 연결된 경우, 입력 스트림은 블록 인코딩 모드로 사용됩니다. 그렇지 않으면 response.write()를 사용하여 일반 문자열을 입력할 수 있습니다.

복잡한 예제: (기본적인 Http 서버와 블록 인코딩, 자세한 내용은 생략)

....
/**
 * 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)
   // 찾지 못하면 NotFound을 throw합니다.
   throw new NotFoundError(path);
 } catch(e) {
  if (e.code)
   throw e;
  throw new NotFoundError(path);
 }
 try {
  // 먼저, 장치를 생성합니다.
  const fileDevice = new Nexus.IO.FilePushDevice(filePath);
  // 그런 다음 우리는 소스 장치를 사용하여 생성된 새로운 ReadableStream을 반환합니다.
  return new Nexus.IO.ReadableStream(fileDevice);
 } catch(e) {
  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 + 밝은 + '서버 오류: ' + e.message + '\n' + e.stack, Reset);
});
/**
 * 연결을 듣습니다.
 */
server.on('connection', async (connection, peer) => {
 // 0번의 연결 ID로 시작하고, 새로운 연결마다 증가합니다.
 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 + 리셋} ${Bright + peer.address}:${peer.port + 리셋} ${
   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);
  // 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();
  // 응답의 상태를 가져옵니다.
  let status = response.status();
  // 터미널에 출력할 색상을 결정합니다。
  const statusColors = {
   '200': 밝은 + FgGreen, // 녹색은 200 (OK),
   '404': 밝은 + FgYellow, // 노란색은 404 (찾을 수 없음)
   '500': 밝은 + FgRed // 빨간색은 500 (내부 서버 오류)
  };
  let statusColor = statusColors[status];
  if (statusColor)
   status = statusColor + status + 리셋;
  // 连接（以及完成时间）记录到控制台。
  console.log(`< #${FgCyan + connId + 리셋} ${Bright + peer.address}:${peer.port + 리셋} ${
   FgGreen + request.method + 리셋} "${FgYellow}${path}${Reset}" ${status} ${(Date.now() * startTime)}ms` +
   (errors.length ? " " + FgRed + 밝은 + errors.map(error => error.message).join(', ') + 리셋 : 리셋));
 }
});
/**
 * 监听的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);
/**
 * Happy streaming!
 */
console.log(FgGreen + `Nexus.js HTTP 서버 ${ip}:${port}에서 리스닝 중` + 초기화);

기준

현재까지 구현된 모든 것을 다 다루었는지 생각해보겠습니다. 그럼 이제 성능에 대해 이야기해보겠습니다.

이 Http 서버의 현재 기준은 다음과 같습니다. 다음은100개의 동시 연결과 총10000개의 요청:

이것은 ApacheBench, 버전 2.3 <$Revision: 1796539 $>
저작권 1996 Adam Twiss, Zeus Technology Ltd, http://www.zeustech.net/
The Apache Software Foundation에 라이선스됨, http://www.apache.org/
localhost를 기준으로 Benchmarking (잘못될 수 있음).....완료
서버 소프트웨어:    nexus.js/0.1.1
서버 호스트네임:    localhost
서버 포트:      3000
문서 경로:     /
문서 길이:    8673 바이트
콘currency 레벨:   100
테스트에 걸린 시간:  9.991 초
완료 요청:   10000
실패 요청:    0
전송된 총량:   87880000 바이트
전송된 HTML:    86730000 바이트
초당 요청:  1000.94 [#/초] (평균)
요청 당 시간:    99.906 [ms] (평균)
요청 당 시간:    0.999 [ms] (평균, 모든 동시 요청에 걸쳐)
전송 속도:     8590.14 [Kbytes/sec] 수신
연결 시간 (ms)
       최소 평균[+/-sd] 중앙값  최대
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
certain time (ms) 내에 처리된 요청의 비율
 50%   84
 66%   97
 75%  105
 80%  112
 90%  134
 95%  188
 98%  233
 99%  238
 100%  464 (가장 긴 요청)

초당1000개의 요청.7위에, 이 기준 테스트 소프트웨어를 포함하여 실행되고 있는5G内存의 IDE, 그리고 서버 본체.

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 모니터 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
버그    :
bogomips: 6784.18
clflush 크기: 64
캐시 정렬: 64
주소 크기  : 39 물리적 비트, 48 가상 비트
전원 관리:

저는 시도했습니다1000개의 동시 요청, 하지만 ApacheBench가 많은 소켓이 열려 timeout이 발생했습니다. httperf를 시도했고, 이는 결과입니다:

voodooattack@voodooattack:~$ httperf --포트=3000 --수-conns=10000 --비율=1000
httperf --클라이언트=0/1 --서버=localhost --포트=3000 --uri=/ --비율=1000 --전송-버퍼=4096 --수신-버퍼=16384 --수-conns=10000 --수-호출=1
httperf: 경고: 파일 열기 제한 > FD_SETSIZE; 최대 열린 파일 수를 FD_SETSIZE로 제한
최대 연결 폭 길이: 262
총: 연결 9779 요청 9779 답변 9779 테스트-지속 시간 10.029 s
연결 비율: 975.1 conn/s (1.0 ms/conn, <=1022 동시 연결
연결 시간 [ms]: 최소 0.5 평균 337.9 최대 7191.8 중간값 79.5 stddev 848.1
연결 시간 [ms]: connect 207.3
연결 길이 [응답]/conn]: 1.000
요청 비율: 975.1 요청/s (1.0 ms/요청)
요청 크기 [B]: 62.0
응답 비율 [응답/초]: 최소 903.5 평균 974.6 최대 1045.7 stddev 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 시간 [초]: 사용자 0.35 시스템 9.67 (사용자 3.5% 시스템 96.4% 총 99.9%)
Net I/O: 8389.9 KB/s (68.7*10^6 bps)
에러: 총 221 client-timi 0 socket-timi 0 connrefused 0 connreset 0
에러: fd-unavail 221 addrunavail 0 ftab-full 0 other 0

너무나도 볼 수 있듯이, 그것은 여전히 작동합니다. 그러나 압력으로 인해 일부 연결이 시간 초과됩니다. 저는 이 문제를 일으킨 원인을 연구하고 있습니다.

이것이 Nexus.js 학습 지식에 대한 모든 내용입니다. 질문이 있으면 아래에 댓글을 달고 논의해 주세요. 울부짖는 교본에 대한 지지에 감사합니다.

선언: 이 문서의 내용은 인터넷에서 가져왔으며, 저작권은 원작자에게 있으며, 인터넷 사용자가 자발적으로 기여하고 업로드한 내용입니다. 이 사이트는 소유권을 가지지 않으며, 인공 편집을 하지 않았으며, 관련 법적 책임을 부담하지 않습니다. 저작권 위반이 의심되는 내용을 발견하면 메일을 notice#w로 보내 주시기 바랍니다.3codebox.com（메일을 보내는 경우，#을 @으로 변경하십시오. 신고하고 관련 증거를 제공하시면, 사이트가 즉시 위반 내용을 삭제합니다.）