Improve performance

benchmarks/index.js

@@ -0,0 +1,108 @@
+/* eslint max-nested-callbacks: ["error", 8] */
+'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 pull = require('pull-stream')
+const assert = require('assert')
+const crypto = require('crypto')
+
+const utils = require('../test/utils')
+
+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)
+      })
+    })
+  }, 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)
+}
+
+function round (nodeArr, blockFactor, n, cb) {
+  const blocks = createBlocks(n, blockFactor)
+  map(blocks, (b, cb) => b.key(cb), (err, keys) => {
+    if (err) {
+      return cb(err)
+    }
+    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) => {
+          const index = i * blockFactor + j
+          return {
+            data: blocks[index].data,
+            key: keys[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) => {
+        pull(
+          node.bitswap.getStream(keys),
+          pull.collect((err, res) => {
+            if (err) {
+              return callback(err)
+            }
+
+            assert(res.length === blocks.length)
+            callback()
+          })
+        )
+      }), cb)
+    ], (err) => {
+      if (err) {
+        return cb(err)
+      }
+      console.log('  %s nodes - %s blocks/node - %sms', n, blockFactor, (new Date()).getTime() - d)
+      cb()
+    })
+  })
+}
+
+function createBlocks (n, blockFactor) {
+  return _.map(_.range(n * blockFactor), () => {
+    return new Block(crypto.randomBytes(n * blockFactor))
+  })
+}

benchmarks/put-get.js

@@ -0,0 +1,89 @@
+'use strict'
+
+const Benchmark = require('benchmark')
+const _ = require('lodash')
+const Block = require('ipfs-block')
+const assert = require('assert')
+const pull = require('pull-stream')
+const series = require('async/series')
+const crypto = require('crypto')
+const utils = require('../test/utils')
+
+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]
+  const bitswap = node.bitswap
+
+  blockCounts.forEach((n) => blockSizes.forEach((k) => {
+    suite.add(`put-get ${n} blocks of size ${k}`, (defer) => {
+      const blocks = createBlocks(n, k)
+      series([
+        (cb) => put(blocks, bitswap, cb),
+        (cb) => get(blocks, bitswap, cb)
+      ], (err) => {
+        if (err) {
+          throw err
+        }
+        defer.resolve()
+      })
+    }, {
+      defer: true
+    })
+  }))
+
+  suite
+    .on('cycle', (event) => {
+      console.log(String(event.target))
+    })
+    .on('complete', () => {
+      process.exit()
+    })
+    .run({
+      async: true
+    })
+})
+
+function createBlocks (n, k) {
+  return _.map(_.range(n), () => {
+    return new Block(crypto.randomBytes(k))
+  })
+}
+
+function put (blocks, bs, callback) {
+  pull(
+    pull.values(blocks),
+    pull.asyncMap((b, cb) => {
+      b.key((err, key) => {
+        if (err) {
+          return cb(err)
+        }
+        cb(null, {key: key, data: b.data})
+      })
+    }),
+    bs.putStream(),
+    pull.onEnd(callback)
+  )
+}
+
+function get (blocks, bs, callback) {
+  pull(
+    pull.values(blocks),
+    pull.asyncMap((b, cb) => b.key(cb)),
+    pull.map((k) => bs.getStream(k)),
+    pull.flatten(),
+    pull.collect((err, res) => {
+      if (err) {
+        return callback(err)
+      }
+      assert(res.length === blocks.length)
+      callback()
+    })
+  )
+}

package.json

@@ -14,6 +14,7 @@
     "release": "aegir-release",
     "release-minor": "aegir-release --type minor",
     "release-major": "aegir-release --type major",
+    "bench": "node benchmarks/index",
     "build": "aegir-build",
     "coverage": "aegir-coverage",
     "coverage-publish": "aegir-coverage publish"
@@ -36,6 +37,7 @@
   "homepage": "https://github.com/ipfs/js-ipfs-bitswap#readme",
   "devDependencies": {
     "aegir": "9.2.2",
+    "benchmark": "^2.1.2",
     "buffer-loader": "0.0.1",
     "chai": "^3.5.0",
     "fs-pull-blob-store": "^0.4.1",
@@ -59,8 +61,13 @@
     "heap": "^0.2.6",
     "ipfs-block": "^0.5.3",
     "lodash.debounce": "^4.0.8",
+    "lodash.find": "^4.6.0",
+    "lodash.groupby": "^4.6.0",
     "lodash.isequalwith": "^4.4.0",
     "lodash.isundefined": "^3.0.1",
+    "lodash.pullallwith": "^4.7.0",
+    "lodash.uniqwith": "^4.5.0",
+    "lodash.values": "^4.3.0",
     "multihashes": "^0.3.0",
     "protocol-buffers": "^3.2.1",
     "pull-defer": "^0.2.2",
@@ -77,4 +84,4 @@
     "greenkeeperio-bot <support@greenkeeper.io>",
     "npmcdn-to-unpkg-bot <npmcdn-to-unpkg-bot@users.noreply.github.com>"
   ]
-}
+}