hyper-overlay / bench / profile-putstream-stages.js

/**
 * Isolate putStream's RSS growth to a specific pipeline stage. Runs three
 * variants of the chunk-write loop and reports peak RSS Δ for each:
 *
 *   A: read + chunk + hash, no bee writes      — chunker baseline
 *   B: A + bee.put(type:'local' pointer)       — Hyperbee + JSON metadata
 *   C: A + bee.put(type:'cache', base64 data)  — current putStream behavior
 *
 * Run: bare bench/profile-putstream-stages.js [src]
 */

import fs from 'bare-fs'
import path from 'bare-path'
import os from 'bare-os'
import process from 'bare-process'
import Corestore from 'corestore'

import { HyperOverlay } from '../index.js'
import { chunkStream, selectTier, createStreamingHasher } from '../lib/chunker.js'

const SRC = Bare.argv[2] || '/tmp/stress-200mb.bin'
if (!fs.existsSync(SRC)) {
  console.error('source file not found: ' + SRC)
  Bare.exit(1)
}
const SIZE = fs.statSync(SRC).size
console.log('source: ' + SRC + '  (' + (SIZE / 1024 / 1024).toFixed(0) + ' MB)')
console.log()

async function variant (label, run) {
  const tmpDir = fs.mkdtempSync(path.join(os.tmpdir(), 'profile-'))
  const corestore = new Corestore(path.join(tmpDir, 'store'))
  const overlay = new HyperOverlay(corestore)
  await overlay.ready()

  const baseline = process.memoryUsage().rss
  let peak = baseline
  let stop = false
  const sampler = (async () => {
    while (!stop) {
      const r = process.memoryUsage().rss
      if (r > peak) peak = r
      await new Promise(res => setTimeout(res, 50))
    }
  })()

  const t0 = Date.now()
  await run(overlay)
  const elapsed = Date.now() - t0
  stop = true
  await sampler

  const delta = Math.round((peak - baseline) / 1024 / 1024)
  console.log(label.padEnd(48) + ' peak Δ: ' + String(delta).padStart(4) + ' MB    elapsed: ' + (elapsed / 1000).toFixed(2) + ' s')

  await overlay.close()
  await corestore.close()
  fs.rmSync(tmpDir, { recursive: true, force: true })
}

await variant('A: read+chunk+hash, no bee writes', async () => {
  const hasher = createStreamingHasher({ size: SIZE })
  const reader = fs.createReadStream(SRC)
  const tier = selectTier(SIZE)
  for await (const c of chunkStream(reader, { tier })) hasher.update(c.data)
  hasher.digest()
})

await variant('B: A + bee.put(type:local pointer)', async (overlay) => {
  const hasher = createStreamingHasher({ size: SIZE })
  const reader = fs.createReadStream(SRC)
  const tier = selectTier(SIZE)
  for await (const c of chunkStream(reader, { tier })) {
    hasher.update(c.data)
    await overlay._registry.putChunk(c.hash, {
      size: c.length, type: 'local', path: SRC, offset: c.offset
    })
  }
  hasher.digest()
})

await variant('C: A + bee.put(type:cache, base64 data)', async (overlay) => {
  const hasher = createStreamingHasher({ size: SIZE })
  const reader = fs.createReadStream(SRC)
  const tier = selectTier(SIZE)
  for await (const c of chunkStream(reader, { tier })) {
    hasher.update(c.data)
    await overlay._registry.putChunk(c.hash, {
      size: c.length, type: 'cache', data: c.data.toString('base64')
    })
  }
  hasher.digest()
})