refactor: update benchmark tests and allow streaming to putMany

Author: achingbrain

benchmarks/index.js

@@ -2,109 +2,76 @@
 /* eslint-disable no-console */
 'use strict'
 
-const series = require('async/series')
-const parallel = require('async/parallel')
-const map = require('async/map')
-const mapSeries = require('async/mapSeries')
-const each = require('async/each')
-const _ = require('lodash')
-const Block = require('ipfs-block')
 const assert = require('assert')
-const crypto = require('crypto')
-const CID = require('cids')
-const multihashing = require('multihashing-async')
+const range = require('lodash.range')
 
-const utils = require('../test/utils')
+const makeBlock = require('../test/utils/make-block')
+const genBitswapNetwork = require('../test/utils/mocks').genBitswapNetwork
 
 const nodes = [2, 5, 10, 20]
 const blockFactors = [1, 10, 100]
 
-console.log('-- start')
-mapSeries(nodes, (n, cb) => {
-  mapSeries(blockFactors, (blockFactor, cb) => {
-    utils.genBitswapNetwork(n, (err, nodeArr) => {
-      if (err) {
-        return cb(err)
-      }
-
-      round(nodeArr, blockFactor, n, (err) => {
-        if (err) {
-          return cb(err)
-        }
-
-        shutdown(nodeArr, cb)
-      })
+;(async function () {
+  console.log('-- start')
+  await Promise.all(
+    nodes.map(async nodeCount => {
+      await Promise.all(
+        blockFactors.map(async blockFactor => {
+          const nodeArr = await genBitswapNetwork(nodeCount)
+          await round(nodeArr, blockFactor, nodeCount)
+          await shutdown(nodeArr)
+        })
+      )
     })
-  }, cb)
-}, (err) => {
-  if (err) {
-    throw err
-  }
+  )
+
   console.log('-- finished')
-})
+})()
 
-function shutdown (nodeArr, cb) {
-  each(nodeArr, (node, cb) => {
-    node.bitswap.stop()
-    node.libp2p.stop(cb)
-  }, cb)
+async function shutdown (nodeArr) {
+  await Promise.all(
+    nodeArr.map(async node => {
+      await node.bitswap.stop()
+      await node.libp2p.stop()
+    })
+  )
 }
 
-function round (nodeArr, blockFactor, n, cb) {
-  createBlocks(n, blockFactor, (err, blocks) => {
-    if (err) {
-      return cb(err)
-    }
-    const cids = blocks.map((b) => b.cid)
-    let d
-    series([
-      // put blockFactor amount of blocks per node
-      (cb) => parallel(_.map(nodeArr, (node, i) => (callback) => {
-        node.bitswap.start()
-
-        const data = _.map(_.range(blockFactor), (j) => {
+async function round (nodeArr, blockFactor, n) {
+  const blocks = await makeBlock(n * blockFactor)
+  const cids = blocks.map((b) => b.cid)
+
+  console.info('put blockFactor amount of blocks per node')
+
+  await Promise.all(
+    nodeArr.map(async (node, i) => {
+      await node.bitswap.start()
+
+      await Promise.all(
+        range(blockFactor).map(async j => {
           const index = i * blockFactor + j
-          return blocks[index]
-        })
-        each(
-          data,
-          (d, cb) => node.bitswap.put(d, cb),
-          callback
-        )
-      }), cb),
-      (cb) => {
-        d = (new Date()).getTime()
-        cb()
-      },
-      // fetch all blocks on every node
-      (cb) => parallel(_.map(nodeArr, (node, i) => (callback) => {
-        map(cids, (cid, cb) => node.bitswap.get(cid, cb), (err, res) => {
-          if (err) {
-            return callback(err)
-          }
-
-          assert(res.length === blocks.length)
-          callback()
+
+          await node.bitswap.put(blocks[index])
         })
-      }), cb)
-    ], (err) => {
-      if (err) {
-        return cb(err)
-      }
-      console.log('  %s nodes - %s blocks/node - %sms', n, blockFactor, (new Date()).getTime() - d)
-      cb()
+      )
     })
-  })
-}
+  )
+
+  console.info('fetch all blocks on every node')
 
-function createBlocks (n, blockFactor, callback) {
-  map(_.range(n * blockFactor), (i, cb) => {
-    const data = crypto.randomBytes(n * blockFactor)
-    multihashing(data, 'sha2-256', (err, hash) => {
-      if (err) {
-        return cb(err)
+  const d = Date.now()
+
+  await Promise.all(
+    nodeArr.map(async node => {
+      let count = 0
+
+      for await (const _ of node.bitswap.getMany(cids)) { // eslint-disable-line no-unused-vars
+        count++
       }
-      cb(null, new Block(data, new CID(hash)))
+
+      assert(count === blocks.length)
     })
-  }, callback)
+  )
+
+  console.log('  %s nodes - %s blocks/node - %sms', n, blockFactor, Date.now() - d)
 }

benchmarks/put-get.js

@@ -3,45 +3,34 @@
 'use strict'
 
 const Benchmark = require('benchmark')
-const _ = require('lodash')
-const Block = require('ipfs-block')
 const assert = require('assert')
-const series = require('async/series')
-const map = require('async/map')
-const crypto = require('crypto')
-const CID = require('cids')
-const multihashing = require('multihashing-async')
-
-const utils = require('../test/utils')
+const all = require('async-iterator-all')
+const makeBlock = require('../test/utils/make-block')
+const genBitswapNetwork = require('../test/utils/mocks').genBitswapNetwork
 
 const suite = new Benchmark.Suite('put-get')
 
 const blockCounts = [1, 10, 1000]
 const blockSizes = [10, 1024, 10 * 1024]
 
-utils.genBitswapNetwork(1, (err, nodes) => {
-  if (err) {
-    throw err
-  }
-  const node = nodes[0]
+;(async function () {
+  const [
+    node
+  ] = await genBitswapNetwork(1)
+
   const bitswap = node.bitswap
 
   blockCounts.forEach((n) => blockSizes.forEach((k) => {
-    suite.add(`put-get ${n} blocks of size ${k}`, (defer) => {
-      createBlocks(n, k, (err, blocks) => {
-        if (err) {
-          throw err
-        }
-        series([
-          (cb) => bitswap.putMany(blocks, cb),
-          (cb) => get(blocks, bitswap, cb)
-        ], (err) => {
-          if (err) {
-            throw err
-          }
-          defer.resolve()
-        })
-      })
+    suite.add(`put-get ${n} blocks of size ${k}`, async (defer) => {
+      const blocks = await makeBlock(n, k)
+
+      await bitswap.putMany(blocks)
+
+      const res = await all(bitswap.getMany(blocks.map(block => block.cid)))
+
+      assert(res.length === blocks.length)
+
+      defer.resolve()
     }, {
       defer: true
     })
@@ -57,29 +46,4 @@ utils.genBitswapNetwork(1, (err, nodes) => {
     .run({
       async: true
     })
-})
-
-function createBlocks (n, k, callback) {
-  map(_.range(n), (i, cb) => {
-    const data = crypto.randomBytes(k)
-    multihashing(data, 'sha2-256', (err, hash) => {
-      if (err) {
-        return cb(err)
-      }
-      cb(null, new Block(data, new CID(hash)))
-    })
-  }, callback)
-}
-
-function get (blocks, bs, callback) {
-  map(blocks, (b, cb) => {
-    bs.get(b.cid, cb)
-  }, (err, res) => {
-    if (err) {
-      return callback(err)
-    }
-
-    assert(res.length === blocks.length)
-    callback()
-  })
-}
+})()

package.json

@@ -44,12 +44,11 @@
   "devDependencies": {
     "@nodeutils/defaults-deep": "^1.1.0",
     "aegir": "^20.3.1",
-    "async": "^2.6.1",
     "async-iterator-all": "^1.0.0",
     "benchmark": "^2.1.4",
     "chai": "^4.2.0",
     "dirty-chai": "^2.0.1",
-    "ipfs-repo": "^0.27.1",
+    "ipfs-repo": "ipfs/js-ipfs-repo#blockstore-accept-async-iterator",
     "libp2p": "^0.26.1",
     "libp2p-kad-dht": "^0.16.0",
     "libp2p-mplex": "^0.8.0",
@@ -83,7 +82,6 @@
     "protons": "^1.0.1",
     "pull-length-prefixed": "^1.3.1",
     "pull-stream": "^3.6.9",
-    "typical": "^5.1.0",
     "varint-decoder": "~0.1.1"
   },
   "pre-push": [

src/index.js

@@ -264,41 +264,43 @@ class Bitswap {
    * @param {function(Error)} callback
    * @returns {void}
    */
-  async put (block) {
-    this._log('putting block')
-
-    const has = await this.blockstore.has(block.cid)
-
-    if (has) {
-      return
+  async put (block) { // eslint-disable-line require-await
+    if (!Array.isArray(block)) {
+      block = [
+        block
+      ]
     }
 
-    await this._putBlock(block)
+    return this.putMany(block)
   }
 
   /**
    * Put the given blocks to the underlying blockstore and
    * send it to nodes that have it them their wantlist.
    *
-   * @param {Array<Block>} blocks
+   * @param {AsyncIterable<Block>} blocks
    * @param {function(Error)} callback
    * @returns {void}
    */
-  async putMany (blocks) {
-    const newBlocks = await Promise.all(blocks.map(async (b) => {
-      return !(await this.blockstore.has(b.cid))
-    })).filter(Boolean)
-
-    await this.blockstore.putMany(newBlocks)
-
-    for (const block of newBlocks) {
-      this.notifications.hasBlock(block)
-      this.engine.receivedBlocks([block.cid])
-      // Note: Don't wait for provide to finish before returning
-      this.network.provide(block.cid).catch((err) => {
-        this._log.error('Failed to provide: %s', err.message)
-      })
-    }
+  async putMany (blocks) { // eslint-disable-line require-await
+    const self = this
+
+    return this.blockstore.putMany(async function * () {
+      for await (const block of blocks) {
+        if (await self.blockstore.has(block.cid)) {
+          continue
+        }
+
+        yield block
+
+        self.notifications.hasBlock(block)
+        self.engine.receivedBlocks([block.cid])
+        // Note: Don't wait for provide to finish before returning
+        self.network.provide(block.cid).catch((err) => {
+          self._log.error('Failed to provide: %s', err.message)
+        })
+      }
+    }())
   }
 
   /**

test/benchmarks/get-many.js

@@ -4,11 +4,14 @@
 
 const distributionTest = require('../utils/distribution-test')
 const print = require('./helpers/print-swarm-results')
+const EventEmitter = require('events')
 
-print('10 nodes, 10 blocks, 5 iterations', distributionTest(10, 10, 5, (err) => {
-  if (err) {
-    throw err
-  }
+;(async function () {
+  const emitter = new EventEmitter()
+
+  print('10 nodes, 10 blocks, 5 iterations', emitter)
+
+  await distributionTest(10, 10, 5, emitter)
 
   console.log('Finished. Can kill now...')
-}))
+})()