-
Notifications
You must be signed in to change notification settings - Fork 31k
/
Copy pathtcp-raw-pipe.js
146 lines (120 loc) Β· 3.29 KB
/
tcp-raw-pipe.js
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
// In this benchmark, we connect a client to the server, and write
// as many bytes as we can in the specified time (default = 10s)
'use strict';
const common = require('../common.js');
const util = require('util');
// If there are --dur=N and --len=N args, then
// run the function with those settings.
// if not, then queue up a bunch of child processes.
const bench = common.createBenchmark(main, {
len: [102400, 1024 * 64 * 16],
type: ['utf', 'asc', 'buf'],
dur: [5],
}, {
test: { len: 1024 },
flags: [ '--expose-internals', '--no-warnings' ],
});
function main({ dur, len, type }) {
const {
TCP,
TCPConnectWrap,
constants: TCPConstants,
} = common.binding('tcp_wrap');
const { WriteWrap } = common.binding('stream_wrap');
const PORT = common.PORT;
function fail(err, syscall) {
throw util._errnoException(err, syscall);
}
// Server
const serverHandle = new TCP(TCPConstants.SERVER);
let err = serverHandle.bind('127.0.0.1', PORT);
if (err)
fail(err, 'bind');
err = serverHandle.listen(511);
if (err)
fail(err, 'listen');
serverHandle.onconnection = function(err, clientHandle) {
if (err)
fail(err, 'connect');
clientHandle.onread = function(buffer) {
// We're not expecting to ever get an EOF from the client.
// Just lots of data forever.
if (!buffer)
fail('read');
const writeReq = new WriteWrap();
writeReq.async = false;
err = clientHandle.writeBuffer(writeReq, Buffer.from(buffer));
if (err)
fail(err, 'write');
writeReq.oncomplete = function(status, handle, err) {
if (err)
fail(err, 'write');
};
};
clientHandle.readStart();
};
// Client
let chunk;
switch (type) {
case 'buf':
chunk = Buffer.alloc(len, 'x');
break;
case 'utf':
chunk = 'ΓΌ'.repeat(len / 2);
break;
case 'asc':
chunk = 'x'.repeat(len);
break;
default:
throw new Error(`invalid type: ${type}`);
}
const clientHandle = new TCP(TCPConstants.SOCKET);
const connectReq = new TCPConnectWrap();
let bytes = 0;
err = clientHandle.connect(connectReq, '127.0.0.1', PORT);
if (err)
fail(err, 'connect');
clientHandle.onread = function(buffer) {
if (!buffer)
fail('read');
bytes += buffer.byteLength;
};
connectReq.oncomplete = function(err) {
if (err)
fail(err, 'connect');
bench.start();
clientHandle.readStart();
setTimeout(() => {
// Multiply by 2 since we're sending it first one way
// then back again.
bench.end(2 * (bytes * 8) / (1024 * 1024 * 1024));
process.exit(0);
}, dur * 1000);
while (clientHandle.writeQueueSize === 0)
write();
};
function write() {
const writeReq = new WriteWrap();
writeReq.oncomplete = afterWrite;
let err;
switch (type) {
case 'buf':
err = clientHandle.writeBuffer(writeReq, chunk);
break;
case 'utf':
err = clientHandle.writeUtf8String(writeReq, chunk);
break;
case 'asc':
err = clientHandle.writeAsciiString(writeReq, chunk);
break;
}
if (err)
fail(err, 'write');
}
function afterWrite(err, handle) {
if (err)
fail(err, 'write');
while (clientHandle.writeQueueSize === 0)
write();
}
}