feat: deterministic DAG structure

.gitignore

@@ -1,5 +1,5 @@
 node_modules
 coverage
-pail.car
+*.car
 dist
 .clinic

bench/put-x10_000-batcher.js

@@ -4,6 +4,7 @@ import * as Batch from '../src/batch/index.js'
 import { ShardBlock } from '../src/shard.js'
 import { MemoryBlockstore } from '../src/block.js'
 import { randomCID, randomString, randomInteger } from '../test/helpers.js'
+import { collectMetrics, writePail } from './util.js'
 
 const NUM = 10_000
 
@@ -23,24 +24,35 @@ async function main () {
     kvs.push([k, v])
   }
 
+  /** @type {API.ShardLink} */
+  let root = rootBlock.cid
   console.log('bench')
   console.time(`put x${NUM}`)
-  const batch = await Batch.create(blocks, rootBlock.cid)
-  for (let i = 0; i < kvs.length; i++) {
-    await batch.put(kvs[i][0], kvs[i][1])
-    if (i % 1000 === 0) {
-      process.stdout.write('.')
+  try {
+    const batch = await Batch.create(blocks, rootBlock.cid)
+    for (let i = 0; i < kvs.length; i++) {
+      await batch.put(kvs[i][0], kvs[i][1])
+      if (i % 1000 === 0) {
+        process.stdout.write('.')
+      }
     }
+    const result = await batch.commit()
+    for (const b of result.additions) {
+      blocks.putSync(b.cid, b.bytes)
+    }
+    for (const b of result.removals) {
+      blocks.deleteSync(b.cid)
+    }
+    root = result.root
+  } catch (err) {
+    console.log('')
+    console.error(err)
+  } finally {
+    console.log('')
+    console.timeEnd(`put x${NUM}`)
+    await writePail(blocks, root)
+    console.log(await collectMetrics(blocks, root))
   }
-  const result = await batch.commit()
-  for (const b of result.additions) {
-    blocks.putSync(b.cid, b.bytes)
-  }
-  for (const b of result.removals) {
-    blocks.deleteSync(b.cid)
-  }
-  console.log('')
-  console.timeEnd(`put x${NUM}`)
 }
 
 main()

bench/put-x10_000.js

@@ -4,6 +4,7 @@ import { put } from '../src/index.js'
 import { ShardBlock } from '../src/shard.js'
 import { MemoryBlockstore } from '../src/block.js'
 import { randomCID, randomString, randomInteger } from '../test/helpers.js'
+import { collectMetrics, writePail } from './util.js'
 
 const NUM = 10_000
 
@@ -16,32 +17,40 @@ async function main () {
 
   /** @type {Array<[string, API.UnknownLink]>} */
   const kvs = []
-
   for (let i = 0; i < NUM; i++) {
     const k = randomString(randomInteger(1, 64))
     const v = await randomCID(randomInteger(8, 128))
     kvs.push([k, v])
   }
 
-  console.log('bench')
-  console.time(`put x${NUM}`)
   /** @type {API.ShardLink} */
   let root = rootBlock.cid
-  for (let i = 0; i < kvs.length; i++) {
-    const result = await put(blocks, root, kvs[i][0], kvs[i][1])
-    for (const b of result.additions) {
-      blocks.putSync(b.cid, b.bytes)
-    }
-    for (const b of result.removals) {
-      blocks.deleteSync(b.cid)
-    }
-    root = result.root
-    if (i % 1000 === 0) {
-      process.stdout.write('.')
+  console.log('bench')
+  console.time(`put x${NUM}`)
+
+  try {
+    for (let i = 0; i < kvs.length; i++) {
+      const result = await put(blocks, root, kvs[i][0], kvs[i][1])
+      for (const b of result.additions) {
+        blocks.putSync(b.cid, b.bytes)
+      }
+      for (const b of result.removals) {
+        blocks.deleteSync(b.cid)
+      }
+      root = result.root
+      if (i % 1000 === 0) {
+        process.stdout.write('.')
+      }
     }
+  } catch (err) {
+    console.log('')
+    console.error(err)
+  } finally {
+    console.log('')
+    console.timeEnd(`put x${NUM}`)
+    await writePail(blocks, root)
+    console.log(await collectMetrics(blocks, root))
   }
-  console.log('')
-  console.timeEnd(`put x${NUM}`)
 }
 
 main()

bench/util.js

@@ -0,0 +1,77 @@
+import fs from 'fs'
+import { Readable } from 'stream'
+import { CarWriter } from '@ipld/car'
+// eslint-disable-next-line no-unused-vars
+import * as API from '../src/api.js'
+import { get, entries } from '../src/index.js'
+import { MemoryBlockstore } from '../src/block.js'
+import { ShardFetcher } from '../src/shard.js'
+
+/**
+ * @param {MemoryBlockstore} blocks
+ * @param {API.ShardLink} root
+ */
+export const writePail = async (blocks, root) => {
+  // @ts-expect-error
+  const { writer, out } = CarWriter.create(root)
+  const finishPromise = new Promise(resolve => {
+    Readable.from(out).pipe(fs.createWriteStream('./bench.car')).on('finish', resolve)
+  })
+
+  for (const b of blocks.entries()) {
+    // @ts-expect-error
+    await writer.put(b)
+  }
+  await writer.close()
+  await finishPromise
+}
+
+/**
+ * @param {MemoryBlockstore} blocks
+ * @param {API.ShardLink} root
+ */
+export const collectMetrics = async (blocks, root) => {
+  const shards = new ShardFetcher(blocks)
+  const rshard = await shards.get(root)
+
+  let maxDepth = 0
+  let totalDepth = 0
+  let totalEntries = 0
+
+  let totalShards = 1
+  let maxShardSize = rshard.bytes.length
+  let totalSize = rshard.bytes.length
+
+  /**
+   * @param {API.ShardEntry} entry
+   * @param {number} depth
+   */
+  const collectData = async ([, v], depth) => {
+    if (!Array.isArray(v)) {
+      totalEntries++
+      maxDepth = depth > maxDepth ? depth : maxDepth
+      totalDepth += depth 
+      return
+    }
+    if (v[1]) totalEntries++
+    const blk = await shards.get(v[0])
+    totalShards++
+    maxShardSize = blk.bytes.length > maxShardSize ? blk.bytes.length : maxShardSize
+    totalSize += blk.bytes.length
+    return Promise.all(blk.value.entries.map(e => collectData(e, depth + 1)))
+  }
+
+  for (const entry of rshard.value.entries) {
+    await collectData(entry, 1)
+  }
+
+  return {
+    maxDepth,
+    avgDepth: Math.round(totalDepth / totalEntries),
+    totalEntries,
+    totalShards,
+    maxShardSize,
+    avgShardSize: Math.round(totalSize / totalShards),
+    totalSize
+  }
+}