update

Author: tsctx

bench.mjs

@@ -76,4 +76,4 @@ await run()
 
 for (const ws of connections) {
   ws.close()
-}
+}

benchmarks/_util/runner.js

@@ -0,0 +1,128 @@
+//@ts-check
+
+"use strict";
+
+class Info {
+  /**@type {string} */
+  #name;
+  /**@type {bigint} */
+  #current;
+  /**@type {bigint} */
+  #finish;
+  /**@type {(...args: any[]) => any} */
+  #callback;
+  /**@type {boolean} */
+  #finalized = false;
+
+  /**
+   * @param {string} name
+   * @param {(...args: any[]) => any} callback
+   */
+  constructor(name, callback) {
+    this.#name = name;
+    this.#callback = callback;
+  }
+
+  get name() {
+    return this.#name;
+  }
+
+  start() {
+    if (this.#finalized) {
+      throw new TypeError("called after finished.");
+    }
+    this.#current = process.hrtime.bigint();
+  }
+
+  end() {
+    if (this.#finalized) {
+      throw new TypeError("called after finished.");
+    }
+    this.#finish = process.hrtime.bigint();
+    this.#finalized = true;
+    this.#callback();
+  }
+
+  diff() {
+    return Number(this.#finish - this.#current);
+  }
+}
+
+/**
+ * @typedef BenchMarkHandler
+ * @type {(ev: { name: string; start(): void; end(): void; }) => any}
+ */
+
+/**
+ * @param {Record<string, BenchMarkHandler>} experiments
+ * @param {{}} [options]
+ * @returns {Promise<{ name: string; average: number; samples: number; fn: BenchMarkHandler; min: number; max: number }[]>}
+ */
+async function bench(experiments, options = {}) {
+  const names = Object.keys(experiments);
+
+  /**@type {{ name: string; average: number; samples: number; fn: BenchMarkHandler; min: number; max: number }[]} */
+  const results = [];
+
+  async function waitMaybePromiseLike(p) {
+    if (
+      (typeof p === "object" || typeof p === "function") &&
+      p !== null &&
+      typeof p.then === "function"
+    ) {
+      await p;
+    }
+  }
+
+  for (let i = 0; i < names.length; ++i) {
+    const name = names[i];
+    const fn = experiments[name];
+    const samples = [];
+
+    let timing = 0;
+
+    for (let j = 0; j < 128 || timing < 800_000_000; ++j) {
+      let resolve = (value) => {},
+        reject = (reason) => {},
+        promise = new Promise(
+          (done, fail) => ((resolve = done), (reject = fail))
+        );
+
+      const info = new Info(name, resolve);
+
+      try {
+        const p = fn(info);
+
+        await waitMaybePromiseLike(p);
+      } catch (err) {
+        reject(err);
+      }
+
+      await promise;
+
+      samples.push({ time: 1e6 * info.diff() });
+
+      timing += info.diff();
+    }
+
+    const average =
+      samples.map((v) => v.time).reduce((a, b) => a + b, 0) / samples.length;
+
+    results.push({
+      name: names[i],
+      average: average,
+      samples: samples.length,
+      fn: fn,
+      min: samples.reduce((a, acc) => Math.min(a, acc.time), samples[0].time),
+      max: samples.reduce((a, acc) => Math.max(a, acc.time), samples[0].time),
+    });
+  }
+
+  return results;
+}
+
+function print() {
+  
+}
+
+module.exports = { bench };

benchmarks/_util/websocket-simple-server.js

@@ -6,29 +6,6 @@ const stream = require('node:stream')
 
 const uid = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
 
-function safeBufferConcat (chunks, length) {
-  if (length === undefined) {
-    length = 0
-    for (let i = 0; i < chunks.length; ++i) {
-      length += chunks[i].length
-    }
-  }
-  const buffer = Buffer.allocUnsafeSlow(length)
-
-  let offset = 0
-  for (let i = 0; i < chunks.length; ++i) {
-    const chunk = chunks[i]
-    buffer.set(chunk, offset)
-    offset += chunk.length
-  }
-
-  if (length !== offset) {
-    buffer.fill(0, offset, buffer.length)
-  }
-
-  return buffer
-}
-
 class ws {
   /**
    * @param {number} opcode
@@ -72,10 +49,10 @@ class ws {
   }
 
   /**
-   * @param {Uint8Array} buffer
+   * @param {Uint8Array | null} buffer
    */
   static parseFrame (buffer) {
-    if (buffer.length < 2) {
+    if (buffer === null || buffer.length < 2) {
       return null
     }
 
@@ -125,6 +102,7 @@ class ws {
   }
 
   static Stream = class extends stream.Writable {
+    /** @type {Uint8Array | null} */
     #head = null
     #receivedLength = 0
 
@@ -133,7 +111,7 @@ class ws {
         if (this.#head === null) {
           this.#head = chunk
         } else {
-          this.#head = safeBufferConcat([this.#head, chunk])
+          this.#head = Buffer.concat([this.#head, chunk])
         }
         const head = this.#head
         const parsed = ws.parseFrame(head)
@@ -172,7 +150,7 @@ class ws {
         if (this.#head === null) {
           this.#head = chunk
         } else if (this.#head.length < 2) {
-          this.#head = safeBufferConcat([this.#head, chunk])
+          this.#head = Buffer.concat([this.#head, chunk])
           merged = true
         } else {
           this.#receivedLength += chunk.length
@@ -187,7 +165,7 @@ class ws {
               const start = length - (parsed.offset + parsed.length)
               if (chunk.length < start) {
                 if (merged) throw new Error('fatal error')
-                this.#head = safeBufferConcat([this.#head, chunk]).subarray(
+                this.#head = Buffer.concat([this.#head, chunk]).subarray(
                   start
                 )
               } else {
@@ -268,21 +246,21 @@ class ws {
 
     writeFrame (frame) {
       if (this.#socket.writable) {
-        return new Promise((resolve, reject) => {
+        return /** @type {Promise<void>} */(new Promise((resolve, reject) => {
           this.#socket.write(frame, (err) => {
             if (err) {
               reject(err)
             } else {
               resolve()
             }
           })
-        })
+        }))
       }
     }
 
     async close () {
       if (this.#socket.writable) {
-        await this.writeFrame(ws.createFrame(ws.opcode.CLOSE))
+        await this.writeFrame(ws.createFrame(ws.opcode.CLOSE, new Uint8Array(0)))
         this.#socket.end()
       }
     }
@@ -297,7 +275,9 @@ class ws {
  */
 function setup ({ onConnection, parseBody }) {
   const server = http.createServer((_req, res) => {
-    res.end('')
+    // Http handler
+    res.writeHead(404)
+    res.end('404 Not Found')
   })
 
   server.on('upgrade', (req, socket, _head) => {

benchmarks/server-websocket.mjs

@@ -0,0 +1 @@
+import './websocket/server/simple.mjs'

benchmarks/websocket-benchmark.mjs

@@ -1,3 +1,94 @@
-import { cronometro } from "cronometro"
+//@ts-check
 
-// TODO
+import { bench } from "./_util/runner.js";
+import { WebSocket } from "../index.js";
+import { randomBytes } from "node:crypto";
+const __BINARY_SIZE__ = 1024 * 256;
+
+const binary = randomBytes(__BINARY_SIZE__);
+
+/**
+ * @param {{
+ *   send(buffer: Uint8Array | string): void;
+ *   addEventListener(name: string, listener: (...args: any) => void, options?: { once: boolean }): void
+ * }} socket
+ * @param {Uint8Array | string} data
+ * @param {(...args: any) => any} callback
+ */
+function waitWrite(socket, data, callback) {
+  return /** @type {Promise<void>} */(new Promise((resolve, reject) => {
+    socket.send(data);
+
+    socket.addEventListener(
+      "message",
+      (ev) => {
+        resolve();
+        callback();
+      },
+      { once: true }
+    );
+  }))
+}
+
+/**@type {Record<string, import('./_util/runner.js').BenchMarkHandler} */
+const experiments = {};
+
+/**@type {WebSocket | null} */
+let ws;
+
+experiments["sending 256Kib (undici)"] = async function (ev) {
+  if (ws === null) {
+    throw new Error("called before initialized");
+  }
+  ev.start();
+  await waitWrite(ws, binary, () => {
+    ev.end();
+  });
+};
+
+async function connect() {
+  ws = new WebSocket("ws://localhost:5001");
+
+  await /**@type {Promise<void>} */ (
+    new Promise((resolve, reject) => {
+      if (ws === null) {
+        return void reject(new Error("called before initialized"));
+      }
+      function onOpen() {
+        resolve();
+        this.removeEventListener("open", onOpen);
+        this.removeEventListener("error", onError);
+      }
+      function onError(err) {
+        reject(err);
+        this.removeEventListener("open", onOpen);
+        this.removeEventListener("error", onError);
+      }
+      ws.addEventListener("open", onOpen);
+      ws.addEventListener("error", onError);
+    })
+  );
+}
+
+connect()
+  .then(() => bench(experiments, {}))
+  .then((results) => {
+    if (ws === null) {
+      throw new Error("called before initialized");
+    }
+
+    ws.close();
+
+    console.log(results);
+  })
+  .catch((err) => {
+    process.nextTick((err) => {
+      throw err;
+    }, err);
+  });
+
+function print(results) {
+  
+}
+
+export {};

benchmarks/websocket/server/simple.mjs

@@ -8,7 +8,7 @@ if (cluster.isPrimary) {
     cluster.fork()
   }
 } else {
-  const emptyFrame = ws.createFrame(ws.opcode.BINARY, Buffer.allocUnsafe(0))
+  const emptyFrame = ws.createFrame(ws.opcode.BINARY, new Uint8Array(0))
 
   const server = setup({
     onConnection (ctrl) {