test: create add-dir benchmark (#167)

* test: create add-dir benchmark

* chore(bench/add-dir): remove unused taskResult prop

* test(bench/add-dir): add kubo-direct test

* docs(bench/add-dir): ipfs-core node_modules output

* chore(bench/add-dir): kubo-direct doesnt pin files when adding

* chore(bench/add-dir): split helia benchmark by blockstore type (fs/mem)

* fix: ⚡ Double Digit Percentage Improvements. (#169)

* fix: ⚡ 2x improvement

Signed-off-by: Nishant Arora <1895906+whizzzkid@users.noreply.github.com>

* fix: dag generation with parallelization at each level

---------

Signed-off-by: Nishant Arora <1895906+whizzzkid@users.noreply.github.com>
Co-authored-by: Russell Dempsey <1173416+SgtPooki@users.noreply.github.com>

* test: use globSource in add-dir benchmark (#171)

* fix: use glob source for importing directories

- Use `globSource` from ipfs-utils for import
- Make all impls return the same CID
- Remove the dir size - if the CID is the same that's good enough

* chore: fix linting

* chore: remove benchmark from workspaces

* test: align error and success output headers

---------

Signed-off-by: Nishant Arora <1895906+whizzzkid@users.noreply.github.com>
Co-authored-by: Nishant Arora <1895906+whizzzkid@users.noreply.github.com>
Co-authored-by: Alex Potsides <alex@achingbrain.net>

benchmarks/add-dir/package.json

@@ -0,0 +1,37 @@
+{
+  "name": "benchmarks-add-dir",
+  "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": "^11.0.0",
+    "@chainsafe/libp2p-yamux": "^3.0.5",
+    "@helia/unixfs": "^1.4.0",
+    "@ipld/dag-pb": "^4.0.2",
+    "@libp2p/websockets": "^5.0.3",
+    "aegir": "^39.0.4",
+    "blockstore-fs": "^1.0.1",
+    "datastore-level": "^10.0.1",
+    "execa": "^7.0.0",
+    "go-ipfs": "^0.19.0",
+    "helia": "^1.0.0",
+    "ipfs-core": "^0.18.0",
+    "ipfs-unixfs-importer": "^15.1.5",
+    "ipfsd-ctl": "^13.0.0",
+    "it-all": "^2.0.0",
+    "it-drain": "^2.0.0",
+    "it-map": "^2.0.1",
+    "kubo-rpc-client": "^3.0.1",
+    "libp2p": "^0.43.0",
+    "multiformats": "^11.0.1",
+    "tinybench": "^2.4.0"
+  }
+}

benchmarks/add-dir/src/helia.ts

@@ -0,0 +1,78 @@
+import fs, { promises as fsPromises } from 'node:fs'
+import os from 'node:os'
+import nodePath from 'node:path'
+import { type AddOptions, unixfs, globSource } from '@helia/unixfs'
+import * as dagPb from '@ipld/dag-pb'
+import { MemoryBlockstore } from 'blockstore-core'
+import { FsBlockstore } from 'blockstore-fs'
+import { MemoryDatastore } from 'datastore-core'
+import { LevelDatastore } from 'datastore-level'
+import { createHelia, type DAGWalker } from 'helia'
+import { fixedSize } from 'ipfs-unixfs-importer/chunker'
+import { balanced } from 'ipfs-unixfs-importer/layout'
+import last from 'it-last'
+import type { AddDirBenchmark } from './index.js'
+import type { CID } from 'multiformats/cid'
+
+const dagPbWalker: DAGWalker = {
+  codec: dagPb.code,
+  async * walk (block) {
+    const node = dagPb.decode(block)
+
+    yield * node.Links.map(l => l.Hash)
+  }
+}
+
+const unixFsAddOptions: Partial<AddOptions> = {
+  // default kubo options
+  cidVersion: 0,
+  rawLeaves: false,
+  layout: balanced({
+    maxChildrenPerNode: 174
+  }),
+  chunker: fixedSize({
+    chunkSize: 262144
+  })
+}
+interface HeliaBenchmarkOptions {
+  blockstoreType?: 'fs' | 'mem'
+  datastoreType?: 'fs' | 'mem'
+}
+
+export async function createHeliaBenchmark ({ blockstoreType = 'fs', datastoreType = 'fs' }: HeliaBenchmarkOptions = {}): Promise<AddDirBenchmark> {
+  const repoPath = nodePath.join(os.tmpdir(), `helia-${Math.random()}`)
+
+  const helia = await createHelia({
+    blockstore: blockstoreType === 'fs' ? new FsBlockstore(`${repoPath}/blocks`) : new MemoryBlockstore(),
+    datastore: datastoreType === 'fs' ? new LevelDatastore(`${repoPath}/data`) : new MemoryDatastore(),
+    dagWalkers: [
+      dagPbWalker
+    ],
+    start: false
+  })
+  const unixFs = unixfs(helia)
+
+  const addFile = async (path: string): Promise<CID> => unixFs.addFile({
+    path: nodePath.relative(process.cwd(), path),
+    content: fs.createReadStream(path)
+  }, unixFsAddOptions)
+
+  const addDir = async function (dir: string): Promise<CID> {
+    const res = await last(unixFs.addAll(globSource(nodePath.dirname(dir), `${nodePath.basename(dir)}/**/*`), unixFsAddOptions))
+
+    if (res == null) {
+      throw new Error('Import failed')
+    }
+
+    return res.cid
+  }
+
+  return {
+    async teardown () {
+      await helia.stop()
+      await fsPromises.rm(repoPath, { recursive: true, force: true })
+    },
+    addFile,
+    addDir
+  }
+}

benchmarks/add-dir/src/index.ts

@@ -0,0 +1,182 @@
+/* eslint-disable no-console,no-loop-func */
+
+import nodePath from 'node:path'
+import debug from 'debug'
+import { CID } from 'multiformats/cid'
+import { Bench } from 'tinybench'
+import { createHeliaBenchmark } from './helia.js'
+import { createIpfsBenchmark } from './ipfs.js'
+import { createKuboDirectBenchmark } from './kubo-direct.js'
+import { createKuboBenchmark } from './kubo.js'
+
+const log = debug('bench:add-dir')
+const ITERATIONS = parseInt(process.env.ITERATIONS ?? '5')
+const MIN_TIME = parseInt(process.env.MIN_TIME ?? '1')
+const TEST_PATH = process.env.TEST_PATH
+const RESULT_PRECISION = 2
+
+export interface AddDirBenchmark {
+  teardown: () => Promise<void>
+  addFile?: (path: string) => Promise<CID>
+  addDir: (path: string) => Promise<CID>
+  getSize?: (cid: CID) => Promise<bigint>
+}
+
+interface BenchmarkTaskResult {
+  timing: number[]
+  cids: Map<string, Set<string>>
+  sizes: Map<string, Set<string>>
+}
+
+const getDefaultResults = (): BenchmarkTaskResult => ({
+  timing: [],
+  cids: new Map<string, Set<string>>(),
+  sizes: new Map<string, Set<string>>()
+})
+
+const impls: Array<{ name: string, create: () => Promise<AddDirBenchmark>, results: BenchmarkTaskResult }> = [
+  {
+    name: 'helia-fs',
+    create: async () => createHeliaBenchmark(),
+    results: getDefaultResults()
+  },
+  {
+    name: 'helia-mem',
+    create: async () => createHeliaBenchmark({ blockstoreType: 'mem', datastoreType: 'mem' }),
+    results: getDefaultResults()
+  },
+  {
+    name: 'ipfs',
+    create: async () => createIpfsBenchmark(),
+    results: getDefaultResults()
+  },
+  {
+    name: 'kubo',
+    create: async () => createKuboBenchmark(),
+    results: getDefaultResults()
+  },
+  {
+    name: 'kubo-direct',
+    create: async () => createKuboDirectBenchmark(),
+    results: getDefaultResults()
+  }
+]
+
+async function main (): Promise<void> {
+  let subject: AddDirBenchmark
+
+  const suite = new Bench({
+    iterations: ITERATIONS,
+    time: MIN_TIME,
+    setup: async (task) => {
+      log('Start: setup')
+      const impl = impls.find(({ name }) => task.name.includes(name))
+      if (impl != null) {
+        subject = await impl.create()
+      } else {
+        throw new Error(`No implementation with name '${task.name}'`)
+      }
+      log('End: setup')
+    },
+    teardown: async () => {
+      log('Start: teardown')
+      await subject.teardown()
+      log('End: teardown')
+    }
+  })
+
+  const testPaths = TEST_PATH != null
+    ? [TEST_PATH]
+    : [
+        nodePath.relative(process.cwd(), nodePath.join(process.cwd(), 'src')),
+        nodePath.relative(process.cwd(), nodePath.join(process.cwd(), 'dist')),
+        nodePath.relative(process.cwd(), nodePath.join(process.cwd(), '..', 'gc', 'src'))
+      ]
+
+  for (const impl of impls) {
+    for (const testPath of testPaths) {
+      const absPath = nodePath.join(process.cwd(), testPath)
+      suite.add(`${impl.name} - ${testPath}`, async function () {
+        const start = Date.now()
+        const cid = await subject.addDir(absPath)
+        impl.results.timing.push(Date.now() - start)
+        const cidSet = impl.results.cids.get(testPath) ?? new Set()
+        cidSet.add(cid.toString())
+        impl.results.cids.set(testPath, cidSet)
+      },
+      {
+        beforeEach: async () => {
+          log(`Start: test ${impl.name}`)
+        },
+        afterEach: async () => {
+          log(`End: test ${impl.name}`)
+          const cidSet = impl.results.cids.get(testPath)
+          if (cidSet != null) {
+            for (const cid of cidSet.values()) {
+              const size = await subject.getSize?.(CID.parse(cid))
+              if (size != null) {
+                const statsSet = impl.results.sizes.get(testPath) ?? new Set()
+                statsSet.add(size?.toString())
+                impl.results.sizes.set(testPath, statsSet)
+              }
+            }
+          }
+        }
+      }
+      )
+    }
+  }
+
+  await suite.run()
+
+  if (process.env.INCREMENT != null) {
+    if (process.env.ITERATION === '1') {
+      console.info('implementation, count, add dir (ms), cid')
+    }
+
+    for (const impl of impls) {
+      console.info(
+        `${impl.name},`,
+        `${process.env.INCREMENT},`,
+        `${(impl.results.timing.reduce((acc, curr) => acc + curr, 0) / impl.results.timing.length).toFixed(RESULT_PRECISION)},`
+      )
+    }
+  } else {
+    const implCids: Record<string, string> = {}
+    const implSizes: Record<string, string> = {}
+    for (const impl of impls) {
+      for (const [testPath, cids] of impl.results.cids.entries()) {
+        implCids[`${impl.name} - ${testPath}`] = Array.from(cids).join(', ')
+      }
+      for (const [testPath, sizes] of impl.results.sizes.entries()) {
+        implSizes[`${impl.name} - ${testPath}`] = Array.from(sizes).join(', ')
+      }
+    }
+    console.table(suite.tasks.map(({ name, result }) => {
+      if (result?.error != null) {
+        return {
+          Implementation: name,
+          'ops/s': 'error',
+          'ms/op': 'error',
+          runs: 'error',
+          p99: 'error',
+          CID: (result?.error as any)?.message
+        }
+      }
+      return {
+        Implementation: name,
+        'ops/s': result?.hz.toFixed(RESULT_PRECISION),
+        'ms/op': result?.period.toFixed(RESULT_PRECISION),
+        runs: result?.samples.length,
+        p99: result?.p99.toFixed(RESULT_PRECISION),
+        CID: implCids[name]
+      }
+    }))
+  }
+  process.exit(0) // sometimes the test hangs (need to debug)
+}
+
+main().catch(err => {
+  console.error(err) // eslint-disable-line no-console
+  process.exit(1)
+})

benchmarks/add-dir/src/ipfs.ts

@@ -0,0 +1,46 @@
+import fs, { promises as fsPromises } from 'node:fs'
+import os from 'node:os'
+import nodePath from 'node:path'
+import { create, globSource } from 'ipfs-core'
+import last from 'it-last'
+import type { AddDirBenchmark } from './index.js'
+import type { CID } from 'multiformats/cid'
+
+export async function createIpfsBenchmark (): Promise<AddDirBenchmark> {
+  const repoPath = nodePath.join(os.tmpdir(), `ipfs-${Math.random()}`)
+
+  const ipfs = await create({
+    config: {
+      Addresses: {
+        Swarm: []
+      }
+    },
+    repo: repoPath,
+    start: false,
+    init: {
+      emptyRepo: true
+    }
+  })
+
+  const addFile = async (path: string): Promise<CID> => (await ipfs.add({ path: nodePath.relative(process.cwd(), path), content: fs.createReadStream(path) }, { cidVersion: 1, pin: false })).cid
+
+  const addDir = async function (dir: string): Promise<CID> {
+    // @ts-expect-error types are messed up
+    const res = await last(ipfs.addAll(globSource(nodePath.dirname(dir), `${nodePath.basename(dir)}/**/*`)))
+
+    if (res == null) {
+      throw new Error('Import failed')
+    }
+
+    return res.cid
+  }
+
+  return {
+    async teardown () {
+      await ipfs.stop()
+      await fsPromises.rm(repoPath, { recursive: true, force: true })
+    },
+    addFile,
+    addDir
+  }
+}

benchmarks/add-dir/src/kubo-direct.ts

@@ -0,0 +1,30 @@
+import { promises as fsPromises } from 'node:fs'
+import os from 'node:os'
+import nodePath from 'node:path'
+import { execa } from 'execa'
+// @ts-expect-error no types
+import * as goIpfs from 'go-ipfs'
+import { CID } from 'multiformats/cid'
+import type { AddDirBenchmark } from './index.js'
+
+export async function createKuboDirectBenchmark (): Promise<AddDirBenchmark> {
+  const repoDir = nodePath.join(os.tmpdir(), 'kubo-direct')
+
+  await execa(goIpfs.path(), ['--repo-dir', repoDir, 'init'])
+
+  const addDir = async function (dir: string): Promise<CID> {
+    const { stdout } = await execa(goIpfs.path(), ['--repo-dir', repoDir, 'add', '-r', '--pin=false', dir])
+    const lines = stdout.split('\n')
+    const lastLine = lines.pop()
+    const cid = CID.parse(lastLine?.split(' ')[1] as string)
+
+    return cid
+  }
+
+  return {
+    async teardown () {
+      await fsPromises.rm(repoDir, { recursive: true, force: true })
+    },
+    addDir
+  }
+}