chore: add transfer benchmark (#90)

Adds a benchmark suite for doing various size data transfers between helia and kubo.

Closes #88

benchmarks/add-dir/src/kubo.ts

@@ -1,7 +1,7 @@
 import { createNode } from 'ipfsd-ctl'
 import last from 'it-last'
 import { path as kuboPath } from 'kubo'
-import { create as kuboRpcClient } from 'kubo-rpc-client'
+import { globSource, create as kuboRpcClient } from 'kubo-rpc-client'
 import type { CID } from 'multiformats/cid'
 import fs, { promises as fsPromises } from 'node:fs'
 import nodePath from 'node:path'
@@ -27,8 +27,7 @@ export async function createKuboBenchmark (): Promise<AddDirBenchmark> {
   })).cid
 
   const addDir = async function (dir: string): Promise<CID> {
-    // @ts-expect-error types are messed up
-    const res = await last(controller.api.addAll(goRpcClient.globSource(nodePath.dirname(dir), `${nodePath.basename(dir)}/**/*`)))
+    const res = await last(controller.api.addAll(globSource(nodePath.dirname(dir), `${nodePath.basename(dir)}/**/*`)))
 
     if (res == null) {
       throw new Error('Import failed')

benchmarks/gc/src/helia.ts

@@ -30,7 +30,7 @@ export async function createHeliaBenchmark (): Promise<GcBenchmark> {
       await drain(helia.blockstore.putMany(map(blocks, ({ key, value }) => ({ cid: key, block: value }))))
     },
     async pin (cid) {
-      await helia.pins.add(cid)
+      await drain(helia.pins.add(cid))
     },
     async teardown () {
       await helia.stop()
@@ -39,7 +39,7 @@ export async function createHeliaBenchmark (): Promise<GcBenchmark> {
       const pins = await all(helia.pins.ls())
 
       for (const pin of pins) {
-        await helia.pins.rm(pin.cid)
+        await drain(helia.pins.rm(pin.cid))
       }
 
       return pins.length

benchmarks/gc/src/kubo.ts

@@ -51,13 +51,7 @@ export async function createKuboBenchmark (): Promise<GcBenchmark> {
         paths: cid
       }))
 
-      const isPinned = result[0].type.includes('direct') || result[0].type.includes('indirect') || result[0].type.includes('recursive')
-
-      if (!isPinned) {
-        console.info(result)
-      }
-
-      return isPinned
+      return result[0].type.includes('direct') || result[0].type.includes('indirect') || result[0].type.includes('recursive')
     },
     hasBlock: async (cid) => {
       try {

benchmarks/transfer/package.json

@@ -0,0 +1,40 @@
+{
+  "name": "benchmarks-transfer",
+  "version": "1.0.0",
+  "main": "index.js",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "clean": "aegir clean",
+    "build": "aegir build --bundle false",
+    "lint": "aegir lint",
+    "dep-check": "aegir dep-check",
+    "start": "npm run build && node dist/src/index.js"
+  },
+  "devDependencies": {
+    "@chainsafe/libp2p-noise": "^15.0.0",
+    "@chainsafe/libp2p-yamux": "^6.0.2",
+    "@helia/unixfs": "^3.0.3",
+    "@ipld/dag-pb": "^4.0.2",
+    "@libp2p/websockets": "^8.0.19",
+    "aegir": "^42.2.5",
+    "blockstore-fs": "^1.0.1",
+    "datastore-level": "^10.0.1",
+    "execa": "^8.0.1",
+    "helia": "^4.1.0",
+    "ipfs-unixfs-importer": "^15.1.1",
+    "ipfsd-ctl": "^14.0.0",
+    "it-all": "^3.0.1",
+    "it-buffer-stream": "^3.0.2",
+    "it-drain": "^3.0.1",
+    "it-map": "^3.0.2",
+    "kubo": "^0.28.0",
+    "kubo-rpc-client": "^4.0.0",
+    "libp2p": "^1.4.0",
+    "multiformats": "^13.1.0",
+    "tinybench": "^2.4.0"
+  },
+  "dependencies": {
+    "pretty-bytes": "^6.1.0"
+  }
+}

benchmarks/transfer/src/README.md

@@ -0,0 +1,42 @@
+# Transfer Benchmark
+
+Benchmarks Helia transfer performance against Kubo
+
+To run:
+
+1. Add `benchmarks/*` to the `workspaces` entry in the root `package.json` of this repo
+2. Run
+    ```console
+    $ npm run reset
+    $ npm i
+    $ npm run build
+    $ cd benchmarks/transfer
+    $ npm start
+
+    > benchmarks-gc@1.0.0 start
+    > npm run build && node dist/src/index.js
+
+
+    > benchmarks-transfer@1.0.0 build
+    > aegir build --bundle false
+
+    [14:51:28] tsc [started]
+    [14:51:33] tsc [completed]
+    generating Ed25519 keypair...
+    ┌─────────┬────────────────┬─────────┬───────────┬──────┐
+    │ (index) │ Implementation │  ops/s  │   ms/op   │ runs │
+    ├─────────┼────────────────┼─────────┼───────────┼──────┤
+    //... results here
+    ```
+
+Recently generated graphs:
+
+- Lower numbers are better
+- The legend arrow indicates direction of transfer
+    - e.g. `helia -> kubo` is the equivalent of
+      1. `ipfs.add` executed on Helia
+      2. `ipfs.cat` executed on Kubo which pulls the data from Helia
+
+<img width="595" alt="image" src="https://github.com/ipfs/helia/assets/665810/302c9d42-8979-4cca-a7e7-13ee6fe083fa">
+
+<img width="594" alt="image" src="https://github.com/ipfs/helia/assets/665810/9b25abfe-2cf2-4c5e-89a1-6b1817dee722">

benchmarks/transfer/src/helia.ts

@@ -0,0 +1,77 @@
+import { createHelia } from 'helia'
+import { createLibp2p } from 'libp2p'
+import { tcp } from '@libp2p/tcp'
+import { noise } from '@chainsafe/libp2p-noise'
+import { yamux } from '@chainsafe/libp2p-yamux'
+import type { TransferBenchmark } from './index.js'
+import os from 'node:os'
+import path from 'node:path'
+import fs from 'node:fs/promises'
+import { LevelDatastore } from 'datastore-level'
+import { FsBlockstore } from 'blockstore-fs'
+import drain from 'it-drain'
+import { unixfs } from '@helia/unixfs'
+import { identify } from '@libp2p/identify'
+import { fixedSize } from 'ipfs-unixfs-importer/chunker'
+import { balanced } from 'ipfs-unixfs-importer/layout'
+
+export async function createHeliaBenchmark (): Promise<TransferBenchmark> {
+  const repoPath = path.join(os.tmpdir(), `helia-${Math.random()}`)
+
+  const helia = await createHelia({
+    blockstore: new FsBlockstore(`${repoPath}/blocks`),
+    datastore: new LevelDatastore(`${repoPath}/data`),
+    libp2p: await createLibp2p({
+      addresses: {
+        listen: [
+          '/ip4/127.0.0.1/tcp/0'
+        ]
+      },
+      transports: [
+        tcp()
+      ],
+      connectionEncryption: [
+        noise()
+      ],
+      streamMuxers: [
+        yamux()
+      ],
+      services: {
+        identify: identify()
+      },
+      connectionManager: {
+        minConnections: 0
+      }
+    })
+  })
+
+  return {
+    async teardown () {
+      await helia.stop()
+      await fs.rm(repoPath, {
+        recursive: true,
+        force: true
+      })
+    },
+    async addr () {
+      return helia.libp2p.getMultiaddrs()[0]
+    },
+    async dial (ma) {
+      await helia.libp2p.dial(ma)
+    },
+    async add (content, options) {
+      const fs = unixfs(helia)
+
+      return await fs.addByteStream(content, {
+        ...options,
+        chunker: options.chunkSize != null ? fixedSize({ chunkSize: options.chunkSize }) : undefined,
+        layout: options.maxChildrenPerNode != null ? balanced({ maxChildrenPerNode: options.maxChildrenPerNode }) : undefined
+      })
+    },
+    async get (cid) {
+      const fs = unixfs(helia)
+
+      await drain(fs.cat(cid))
+    }
+  }
+}

benchmarks/transfer/src/index.ts

@@ -0,0 +1,143 @@
+/* eslint-disable no-console */
+
+import type { CID } from 'multiformats/cid'
+import { createHeliaBenchmark } from './helia.js'
+import { createKuboBenchmark } from './kubo.js'
+import bufferStream from 'it-buffer-stream'
+import type { Multiaddr } from '@multiformats/multiaddr'
+import prettyBytes from 'pretty-bytes'
+
+const ONE_MEG = 1024 * 1024
+
+export interface TransferBenchmark {
+  teardown: () => Promise<void>
+  addr: () => Promise<Multiaddr>
+  dial: (multiaddr: Multiaddr) => Promise<void>
+  add: (content: AsyncIterable<Uint8Array>, options: ImportOptions) => Promise<CID>
+  get: (cid: CID) => Promise<void>
+}
+
+export interface ImportOptions {
+  cidVersion?: 0 | 1
+  rawLeaves?: boolean
+  chunkSize?: number
+  maxChildrenPerNode?: number
+}
+
+interface File {
+  name: string
+  options: ImportOptions
+  size: number
+}
+
+const opts: Record<string, ImportOptions> = {
+  'kubo defaults': {
+    chunkSize: 256 * 1024,
+    rawLeaves: false,
+    cidVersion: 0,
+    maxChildrenPerNode: 174
+  },
+  'filecoin defaults': {
+    chunkSize: 1024 * 1024,
+    rawLeaves: true,
+    cidVersion: 1,
+    maxChildrenPerNode: 1024
+  },
+/*  '256KiB block size': {
+    chunkSize: 256 * 1024,
+    rawLeaves: true,
+    cidVersion: 1,
+    maxChildrenPerNode: 174
+  },
+  '512KiB block size': {
+    chunkSize: 256 * 1024 * 2,
+    rawLeaves: true,
+    cidVersion: 1,
+    maxChildrenPerNode: 174
+  },
+  '1MB block size': {
+    chunkSize: 1024 * 1024,
+    rawLeaves: true,
+    cidVersion: 1,
+    maxChildrenPerNode: 174
+  },
+  '2MB block size': {
+    chunkSize: (1024 * 1024) * 2,
+    rawLeaves: true,
+    cidVersion: 1,
+    maxChildrenPerNode: 174
+  },
+  '3MB block size': {
+    chunkSize: (1024 * 1024) * 3,
+    rawLeaves: true,
+    cidVersion: 1,
+    maxChildrenPerNode: 174
+  },
+  'Max block size': {
+    chunkSize: 4193648,
+    rawLeaves: true,
+    cidVersion: 1,
+    maxChildrenPerNode: 174
+  }
+  // Kubo will not sent bitswap messages larger than this
+*/
+}
+
+const tests: Record<string, File[]> = {}
+
+for (const [name, options] of Object.entries(opts)) {
+  tests[name] = []
+
+  for (let i = 100; i < 1100; i += 100) {
+    tests[name].push({
+      name: `${i}`,
+      options,
+      size: ONE_MEG * i
+    })
+
+    console.info(prettyBytes(ONE_MEG * i))
+  }
+}
+
+const impls: Array<{ name: string, create: () => Promise<TransferBenchmark> }> = [{
+  name: 'helia',
+  create: async () => await createHeliaBenchmark()
+}, {
+  name: 'kubo',
+  create: async () => await createKuboBenchmark()
+}]
+
+async function main (): Promise<void> {
+  for (const [name, files] of Object.entries(tests)) {
+    for (const implA of impls) {
+      for (const implB of impls) {
+        console.info(`${implA.name} -> ${implB.name} ${name}`)
+
+        for (const file of files) {
+          const subjectA = await implA.create()
+          const subjectB = await implB.create()
+
+          const addr = await subjectB.addr()
+          await subjectA.dial(addr)
+
+          const cid = await subjectA.add(bufferStream(file.size), file.options)
+
+          const start = Date.now()
+
+          // b pulls from a
+          await subjectB.get(cid)
+
+          console.info(`${Date.now() - start}`)
+
+          await subjectA.teardown()
+          await subjectB.teardown()
+        }
+      }
+    }
+  }
+}
+
+main().catch(err => {
+  console.error(err) // eslint-disable-line no-console
+  process.exit(1)
+})