feat: use snappy-wasm

packages/beacon-node/package.json

@@ -147,6 +147,7 @@
     "multiformats": "^11.0.1",
     "prom-client": "^15.1.0",
     "qs": "^6.11.1",
+    "snappy": "^7.2.2",
     "snappyjs": "^0.7.0",
     "strict-event-emitter-types": "^2.0.0",
     "systeminformation": "^5.17.12",

packages/beacon-node/src/network/gossip/encoding.ts

@@ -1,4 +1,5 @@
-import {compress, uncompress} from "snappyjs";
+import * as snappyjs from "snappyjs";
+import * as snappy from "snappy";
 import xxhashFactory from "xxhash-wasm";
 import {Message} from "@libp2p/interface";
 import {digest} from "@chainsafe/as-sha256";
@@ -61,6 +62,11 @@ export function msgIdFn(gossipTopicCache: GossipTopicCache, msg: Message): Uint8
   return Buffer.from(digest(Buffer.concat(vec))).subarray(0, 20);
 }
 
+/** Snappyjs is faster at compressing small buffers */
+const SNAPPYJS_COMPRESS_THRESHOLD = 100;
+/** Snappyjs is faster at uncompressing small buffers */
+const SNAPPYJS_UNCOMPRESS_THRESHOLD = 200;
+
 export class DataTransformSnappy {
   constructor(
     private readonly gossipTopicCache: GossipTopicCache,
@@ -74,7 +80,12 @@ export class DataTransformSnappy {
    * - `outboundTransform()`: compress snappy payload
    */
   inboundTransform(topicStr: string, data: Uint8Array): Uint8Array {
-    const uncompressedData = uncompress(data, this.maxSizePerMessage);
+    let uncompressedData: Uint8Array;
+    if (data.length < SNAPPYJS_UNCOMPRESS_THRESHOLD) {
+      uncompressedData = snappyjs.uncompress(data, this.maxSizePerMessage);
+    } else {
+      uncompressedData = snappy.uncompressSync(Buffer.from(data)) as Uint8Array;
+    }
 
     // check uncompressed data length before we extract beacon block root, slot or
     // attestation data at later steps
@@ -101,6 +112,10 @@ export class DataTransformSnappy {
       throw Error(`ssz_snappy encoded data length ${length} > ${this.maxSizePerMessage}`);
     }
     // No need to parse topic, everything is snappy compressed
-    return compress(data);
+    if (data.length < SNAPPYJS_COMPRESS_THRESHOLD) {
+      return snappyjs.compress(data);
+    } else {
+      return snappy.compressSync(Buffer.from(data));
+    }
   }
 }

packages/beacon-node/test/perf/network/gossip/snappy.test.ts

@@ -0,0 +1,61 @@
+import {randomBytes} from "node:crypto";
+import * as snappyjs from "snappyjs";
+import * as snappy from "snappy";
+import {itBench} from "@dapplion/benchmark";
+
+describe("network / gossip / snappy", () => {
+  const msgLens = [100, 200, 300, 400, 500, 1000, 10000, 100000];
+  describe("compress", () => {
+    for (const msgLen of msgLens) {
+      const uncompressed = randomBytes(msgLen);
+      const RUNS_FACTOR = 1000;
+
+      itBench({
+        id: `${msgLen} bytes - compress - snappyjs`,
+        runsFactor: RUNS_FACTOR,
+        fn: () => {
+          for (let i = 0; i < RUNS_FACTOR; i++) {
+            snappyjs.compress(uncompressed);
+          }
+        },
+      });
+
+      itBench({
+        id: `${msgLen} bytes - compress - snappy`,
+        runsFactor: RUNS_FACTOR,
+        fn: () => {
+          for (let i = 0; i < RUNS_FACTOR; i++) {
+            snappy.compressSync(uncompressed);
+          }
+        },
+      });
+    }
+  });
+  describe("uncompress", () => {
+    for (const msgLen of [100, 200, 300, 400, 500, 1000, 10000, 100000]) {
+      const uncompressed = randomBytes(msgLen);
+      const compressed = snappyjs.compress(uncompressed);
+      const RUNS_FACTOR = 1000;
+
+      itBench({
+        id: `${msgLen} bytes - uncompress - snappyjs`,
+        runsFactor: RUNS_FACTOR,
+        fn: () => {
+          for (let i = 0; i < RUNS_FACTOR; i++) {
+            snappyjs.uncompress(compressed);
+          }
+        },
+      });
+
+      itBench({
+        id: `${msgLen} bytes - uncompress - snappy`,
+        runsFactor: RUNS_FACTOR,
+        fn: () => {
+          for (let i = 0; i < RUNS_FACTOR; i++) {
+            snappy.uncompressSync(compressed);
+          }
+        },
+      });
+    }
+  });
+});