ホームページ > 記事 > ウェブフロントエンド > JavaScriptによるマルチスレッドランタイムライブラリNexus.jsの詳細説明（添付コードの一部）
JavaScriptによるマルチスレッドランタイムライブラリNexus.jsの詳細説明（添付コードの一部）
- 亚连オリジナル
- 2018-05-18 17:49:141613ブラウズ
この記事では主に JavaScript マルチスレッド ランタイム ライブラリ Nexus.js の学習体験とコード共有を紹介します。必要な友達はそれを参照して一緒に学ぶことができます。
イベントループ
イベントループはありません
(ロックフリー)タスクオブジェクトを含むスレッドプールがあります
setTimeoutまたはsetImmediateが呼び出されるか、Promiseが作成されるたびに、タスクはキューに入れられますタスクキューのクロック。
タスクがスケジュールされると、最初に利用可能なスレッドがタスクを選択して実行します。
CPU コアで Promise を処理します。 Promise.all() を呼び出すと、Promise が並行して解決されます。 eS6
async/awaitsをサポートします。 CommonJS はサポートされていません。 (require(...) および module.exports)
すべてのモジュールは ES6 インポート/エクスポート構文を使用します
import('file-or-packge').then(...) による動的インポートをサポートします
import をサポートします。
追加機能: 次のような URL からの直接インポートをサポート:
import { h } from 'https://unpkg.com/preact/dist/preact.esm.js';EventEmitterNexus は 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
すべての入出力は、デバイス、フィルター、ストリームの 3 つのプリミティブを通じて行われます。
すべての入出力プリミティブは EventEmitter クラスを実装します
Device を使用するには、Device の上に ReadableStream または WritableStream を作成する必要があります
データを操作するには、ReadableStream または WritableStream にフィルターを追加できます。
最後に、source.pipe(...destinationStreams) を使用し、source.resume() がデータを処理するのを待ちます。
すべての入出力操作は ArrayBuffer オブジェクトを使用して行われます。 フィルターはデータを処理するために process(buffer) メソッドを試みました。 例: 2 つの別々の出力ファイルを使用して 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(`finished in ${(Date.now() * startTime) / 1000} seconds!`);
} catch (e) {
console.error('An error occurred: ', e);
}
}
start().catch(console.error);
TCP/UDP
Nexus.js は、IP アドレス/ポートのバインドと接続の監視を担当する Acceptor クラスを提供します
接続リクエストが受信されるたびに、接続イベントがトリガーされて提供されますソケットデバイス。
各 Socket インスタンスは全二重 I/O デバイスです。
Socket の操作には ReadableStream と WritableStream を使用できます。
最も基本的な例: (クライアントに「Hello World」を送信)
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');Http
Nexus は、基本的に TCPAcceptor を継承する Nexus.Net.HTTP.Server クラスを提供します
基本インターフェース
いつサーバーが受信接続の基本的な HTTP ヘッダーの解析/検証を完了すると、接続と同じ情報を使用して接続イベントがトリガーされます
各接続インスタンスにはリクエストと応答オブジェクトがあります。これらは入出力デバイスです。
ReadableStream と WritableStream を構築してリクエスト/レスポンスを操作できます。
パイプ経由で 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)
// 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 new InternalServerError(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/0.1.1`);
// 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} ${(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);ベンチマーク
これまでに実装したものはすべてカバーできたと思います。それでは、パフォーマンスについて話しましょう。
これは、アピール HTTP サーバーの現在のベンチマークです。同時接続数 100、リクエスト数合計 10000 です:
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: nexus.js/0.1.1
Server Hostname: localhost
Server Port: 3000
Document Path: /
Document Length: 8673 bytes
Concurrency Level: 100
Time taken for tests: 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)1 秒あたり 1000 リクエスト。古い i7 では、ベンチマーク ソフトウェア、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 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:
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: 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.029 s Connection rate: 975.1 conn/s (1.0 ms/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.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 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 0
ご覧のとおり、まだ動作します。ただし、一部の接続はストレスによりタイムアウトします。この問題の原因については現在も調査中です。
上記は私があなたのためにまとめたものです。
関連記事:
JS操作のDOMツリートラバーサル方法のまとめ JSは配列重複排除アルゴリズムを実装 JSは最小公倍数と最大公約数を求める以上がJavaScriptによるマルチスレッドランタイムライブラリNexus.jsの詳細説明（添付コードの一部）の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。