1/**
2 * Socket read/write cost against raw `node:net`, `ws` and in-memory baselines.
3 *
4 * Two workloads run against plain Node servers, so only the client transport
5 * differs between tasks:
6 *
7 * - `stream`: the client requests a fixed payload and reads it back, measuring
8 * the read path under a burst of large frames.
9 * - `ping-pong`: the client writes a small frame and waits for the echo,
10 * measuring per-message cost on both paths.
11 *
12 * The `node:net` and `ws` transports go over loopback, which is dominated by
13 * syscalls: a CPU profile of the ping-pong run is roughly 20% idle and 35%
14 * kernel writes, so anything smaller than a microsecond is invisible there.
15 * The `duplex` transport is a cross-wired `PassThrough` pair with no kernel in
16 * the path, which is where allocation-level changes to the Socket layer show
17 * up. Its raw baseline is the same duplex read with `data` events.
18 *
19 * Run with `pnpm --dir packages/platform/node-shared benchmark:socket`.
20 */
21import * as NodeSocket from "@effect/platform-node-shared/NodeSocket"
22import type { NonEmptyReadonlyArray } from "effect/Array"
23import * as Effect from "effect/Effect"
24import * as Exit from "effect/Exit"
25import * as Scope from "effect/Scope"
26import * as Socket from "effect/socket/Socket"
27import { once } from "node:events"
28import * as Net from "node:net"
29import { Duplex } from "node:stream"
30import { Bench } from "tinybench"
31import { WebSocket, WebSocketServer } from "ws"
32
33const streamFrameSize = 64 * 1024
34const streamFrameCount = 16
35const streamBytes = streamFrameSize * streamFrameCount
36
37const pingPongCount = 256
38const pingPongSize = 64
39
40const streamFrame = new Uint8Array(streamFrameSize).fill(7)
41const streamFrames: NonEmptyReadonlyArray<Uint8Array> = [
42 streamFrame,
43 ...Array.from({ length: streamFrameCount - 1 }, () => streamFrame)
44]
45const request: NonEmptyReadonlyArray<Uint8Array> = [new Uint8Array([1])]
46const ping: NonEmptyReadonlyArray<Uint8Array> = [new Uint8Array(pingPongSize).fill(3)]
47
48// Servers are plain Node so that only the client transport is measured.
49
50const startNetServer = async (onConnection: (conn: Net.Socket) => void): Promise<Net.Server> => {
51 const server = Net.createServer((conn) => {
52 conn.on("error", () => {})
53 conn.setNoDelay(true)
54 onConnection(conn)
55 })
56 server.listen(0, "127.0.0.1")
57 await once(server, "listening")
58 return server
59}
60
61const startWebSocketServer = async (onConnection: (ws: WebSocket) => void): Promise<WebSocketServer> => {
62 const server = new WebSocketServer({ host: "127.0.0.1", port: 0 })
63 await once(server, "listening")
64 server.on("connection", (ws) => {
65 ws.on("error", () => {})
66 onConnection(ws)
67 })
68 return server
69}
70
71const portOf = (server: Net.Server | WebSocketServer) => (server.address() as Net.AddressInfo).port
72
73const netEchoServer = await startNetServer((conn) => {
74 conn.pipe(conn)
75})
76const netStreamServer = await startNetServer((conn) => {
77 conn.on("data", () => {
78 conn.cork()
79 for (const frame of streamFrames) conn.write(frame)
80 conn.uncork()
81 })
82})
83const wsEchoServer = await startWebSocketServer((ws) => {
84 ws.on("message", (data: Buffer) => ws.send(data))
85})
86const wsStreamServer = await startWebSocketServer((ws) => {
87 ws.on("message", () => {
88 for (const frame of streamFrames) ws.send(frame)
89 })
90})
91
92// A duplex whose write side feeds the read side, so a client talks to itself
93// with no kernel in the path. The read side is unbounded, so this measures the
94// Socket layer rather than transport backpressure.
95const loopbackDuplex = (respond: (chunk: Buffer, push: (data: Uint8Array) => void) => void): Duplex => {
96 const duplex: Duplex = new Duplex({
97 read() {},
98 write(chunk: Buffer, _encoding, callback) {
99 respond(chunk, (data) => {
100 duplex.push(data)
101 })
102 callback()
103 }
104 })
105 return duplex
106}
107
108const duplexEcho = (): Duplex => loopbackDuplex((chunk, push) => push(chunk))
109
110const duplexStream = (): Duplex =>
111 loopbackDuplex((_chunk, push) => {
112 for (const frame of streamFrames) push(frame)
113 })
114
115/**
116 * Tracks how many bytes have arrived so a read of `n` bytes resolves once the
117 * cumulative total passes what earlier reads already claimed.
118 */
119const makeReadCounter = () => {
120 let received = 0
121 let consumed = 0
122 let pending: { readonly target: number; readonly resolve: () => void } | undefined
123 return {
124 push(size: number) {
125 received += size
126 if (pending !== undefined && received >= pending.target) {
127 const resolve = pending.resolve
128 pending = undefined
129 resolve()
130 }
131 },
132 read(bytes: number): Promise<void> {
133 consumed += bytes
134 if (received >= consumed) return Promise.resolve()
135 const target = consumed
136 return new Promise<void>((resolve) => {
137 pending = { target, resolve }
138 })
139 }
140 }
141}
142
143interface RawClient {
144 readonly write: (frames: NonEmptyReadonlyArray<Uint8Array>) => Promise<void>
145 readonly read: (bytes: number) => Promise<void>
146 readonly close: () => void
147}
148
149const rawNetClient = async (port: number): Promise<RawClient> => {
150 const conn = Net.createConnection({ host: "127.0.0.1", port })
151 await once(conn, "connect")
152 conn.setNoDelay(true)
153 const counter = makeReadCounter()
154 conn.on("data", (chunk: Buffer) => counter.push(chunk.length))
155 return {
156 write(frames) {
157 conn.cork()
158 let flushed = true
159 for (const frame of frames) flushed = conn.write(frame)
160 conn.uncork()
161 return flushed ? Promise.resolve() : once(conn, "drain").then(() => {})
162 },
163 read: counter.read,
164 close: () => conn.destroy()
165 }
166}
167
168const rawWebSocketClient = async (url: string): Promise<RawClient> => {
169 const ws = new WebSocket(url)
170 await once(ws, "open")
171 const counter = makeReadCounter()
172 ws.on("message", (data: Buffer) => counter.push(data.length))
173 return {
174 write(frames) {
175 return new Promise<void>((resolve, reject) => {
176 for (let i = 0; i < frames.length - 1; i++) ws.send(frames[i])
177 ws.send(frames[frames.length - 1], (error) => error ? reject(error) : resolve())
178 })
179 },
180 read: counter.read,
181 close: () => ws.terminate()
182 }
183}
184
185const rawDuplexClient = (conn: Duplex): RawClient => {
186 const counter = makeReadCounter()
187 conn.on("data", (chunk: Buffer) => counter.push(chunk.length))
188 return {
189 write(frames) {
190 let flushed = true
191 for (const frame of frames) flushed = conn.write(frame)
192 return flushed ? Promise.resolve() : once(conn, "drain").then(() => {})
193 },
194 read: counter.read,
195 close: () => conn.destroy()
196 }
197}
198
199interface SocketClient {
200 readonly write: (frames: NonEmptyReadonlyArray<Uint8Array>) => Effect.Effect<void>
201 readonly read: (bytes: number) => Effect.Effect<void>
202}
203
204const socketClient = Effect.fnUntraced(function*(socket: Socket.Socket) {
205 const writer = yield* socket.writer
206 const pull = yield* Socket.readerBytes(socket)
207 const read = Effect.orDie(pull)
208 let received = 0
209 let consumed = 0
210 const drain: Effect.Effect<void> = Effect.flatMap(read, (frames) => {
211 for (let i = 0; i < frames.length; i++) received += frames[i].length
212 return received >= consumed ? Effect.void : drain
213 })
214 const client: SocketClient = {
215 write: (frames) => Effect.orDie(writer.writeAll(frames)),
216 read: (bytes) =>
217 Effect.suspend(() => {
218 consumed += bytes
219 return received >= consumed ? Effect.void : drain
220 })
221 }
222 return client
223})
224
225const scope = await Effect.runPromise(Scope.make())
226const runScoped = <A, E>(effect: Effect.Effect<A, E, Scope.Scope>) => Effect.runPromise(Scope.provide(effect, scope))
227
228const duplexSocketClient = (open: () => Duplex) =>
229 runScoped(Effect.flatMap(NodeSocket.fromDuplex(Effect.sync(open)), socketClient))
230
231const netSocketClient = (port: number) =>
232 runScoped(Effect.flatMap(NodeSocket.makeNet({ host: "127.0.0.1", port, noDelay: true }), socketClient))
233
234const webSocketSocketClient = (url: string) =>
235 runScoped(
236 Effect.flatMap(Socket.makeWebSocket(url), socketClient).pipe(
237 Effect.provideService(Socket.WebSocketConstructor, (url, options) => new WebSocket(url, options as any))
238 )
239 )
240
241const netEchoPort = portOf(netEchoServer)
242const netStreamPort = portOf(netStreamServer)
243const wsEchoUrl = `ws://127.0.0.1:${portOf(wsEchoServer)}`
244const wsStreamUrl = `ws://127.0.0.1:${portOf(wsStreamServer)}`
245
246const rawClients = [
247 await rawNetClient(netStreamPort),
248 await rawWebSocketClient(wsStreamUrl),
249 await rawNetClient(netEchoPort),
250 await rawWebSocketClient(wsEchoUrl)
251]
252const [rawNetStream, rawWsStream, rawNetEcho, rawWsEcho] = rawClients
253
254const rawDuplexStream = rawDuplexClient(duplexStream())
255const rawDuplexEcho = rawDuplexClient(duplexEcho())
256const socketDuplexStream = await duplexSocketClient(duplexStream)
257const socketDuplexEcho = await duplexSocketClient(duplexEcho)
258
259const socketNetStream = await netSocketClient(netStreamPort)
260const socketWsStream = await webSocketSocketClient(wsStreamUrl)
261const socketNetEcho = await netSocketClient(netEchoPort)
262const socketWsEcho = await webSocketSocketClient(wsEchoUrl)
263
264const rawStream = (client: RawClient) => async () => {
265 await client.write(request)
266 await client.read(streamBytes)
267}
268
269const rawPingPong = (client: RawClient) => async () => {
270 for (let i = 0; i < pingPongCount; i++) {
271 await client.write(ping)
272 await client.read(pingPongSize)
273 }
274}
275
276const socketStream = (client: SocketClient) => {
277 const effect = Effect.flatMap(client.write(request), () => client.read(streamBytes))
278 return () => Effect.runPromise(effect)
279}
280
281const socketPingPong = (client: SocketClient) => {
282 // A plain counter loop rather than `Effect.repeat`, whose Schedule reads the
283 // clock every iteration and would cost more than the round-trip it wraps.
284 const round = Effect.flatMap(client.write(ping), () => client.read(pingPongSize))
285 let remaining = 0
286 const loop: Effect.Effect<void> = Effect.flatMap(round, () => --remaining > 0 ? loop : Effect.void)
287 return () => {
288 remaining = pingPongCount
289 return Effect.runPromise(loop)
290 }
291}
292
293const options = {
294 iterations: 16,
295 time: 2_000,
296 warmupIterations: 8,
297 warmupTime: 500,
298 timestampProvider: "hrtimeNow" as const
299}
300
301const runSuite = async (
302 name: string,
303 shape: string,
304 unit: string,
305 unitsPerRun: number,
306 bytesPerRun: number,
307 tasks: ReadonlyArray<readonly [name: string, run: () => Promise<void>]>
308) => {
309 const bench = new Bench(options)
310 for (const [taskName, run] of tasks) {
311 bench.add(taskName, run)
312 }
313 await bench.run()
314 console.log(`\n${name} (${shape})`)
315 console.table(bench.table((task) => {
316 const result = task.result
317 if (result?.state !== "completed") {
318 return { Transport: task.name, State: result?.state ?? "missing result" }
319 }
320 return {
321 Transport: task.name,
322 [`${unit}/s`]: Math.round(result.throughput.mean * unitsPerRun),
323 "MiB/s": (result.throughput.mean * bytesPerRun / (1024 * 1024)).toFixed(1),
324 [`Latency (µs/${unit})`]: (result.latency.mean * 1_000 / unitsPerRun).toFixed(2),
325 RME: `${result.latency.rme.toFixed(2)}%`,
326 Samples: result.latency.samplesCount
327 }
328 }))
329}
330
331await runSuite(
332 "stream read",
333 `${streamFrameCount} x ${streamFrameSize / 1024} KiB frames per run`,
334 "frame",
335 streamFrameCount,
336 streamBytes,
337 [
338 ["node:net (raw)", rawStream(rawNetStream)],
339 ["node:net (Socket)", socketStream(socketNetStream)],
340 ["ws (raw)", rawStream(rawWsStream)],
341 ["ws (Socket)", socketStream(socketWsStream)],
342 ["duplex (raw)", rawStream(rawDuplexStream)],
343 ["duplex (Socket)", socketStream(socketDuplexStream)]
344 ]
345)
346
347await runSuite(
348 "ping-pong",
349 `${pingPongCount} sequential ${pingPongSize} B round-trips per run`,
350 "round-trip",
351 pingPongCount,
352 pingPongCount * pingPongSize * 2,
353 [
354 ["node:net (raw)", rawPingPong(rawNetEcho)],
355 ["node:net (Socket)", socketPingPong(socketNetEcho)],
356 ["ws (raw)", rawPingPong(rawWsEcho)],
357 ["ws (Socket)", socketPingPong(socketWsEcho)],
358 ["duplex (raw)", rawPingPong(rawDuplexEcho)],
359 ["duplex (Socket)", socketPingPong(socketDuplexEcho)]
360 ]
361)
362
363for (const client of rawClients) client.close()
364rawDuplexStream.close()
365rawDuplexEcho.close()
366await Effect.runPromise(Scope.close(scope, Exit.void))
367netEchoServer.close()
368netStreamServer.close()
369wsEchoServer.close()
370wsStreamServer.close()