Merge pull request #84 from ipfs/sq

Now all together: Perf and CIDs
ipfs · Dec 23, 2016 · 0c112c6 · 0c112c6
2 parents 252b192 + c65e722
commit 0c112c6
Show file tree

Hide file tree

Showing 52 changed files with 2,803 additions and 2,574 deletions.
diff --git a/API.md b/API.md
diff --git a/README.md b/README.md
@@ -31,21 +31,21 @@
 ### npm
 
 ```sh
-> npm i ipfs-bitswap
+> npm install ipfs-bitswap --save
 ```
 
 ### Use in Node.js
 
 ```js
-const bitswap = require('ipfs-bitswap')
+const Bitswap = require('ipfs-bitswap')
 ```
 
 ### Use in a browser with browserify, webpack or any other bundler
 
 The code published to npm that gets loaded on require is in fact a ES5 transpiled version with the right shims added. This means that you can require it and use with your favourite bundler without having to adjust asset management process.
 
 ```js
-const bitswap = require('ipfs-bitswap')
+const Bitswap = require('ipfs-bitswap')
 ```
 
 ### Use in a browser using a script tag
@@ -62,6 +62,114 @@ Loading this module through a script tag will make the `IpfsBitswap` object avai
 
 For the documentation see [API.md](API.md).
 
+### API
+
+#### `new Bitswap(libp2p, blockstore)`
+
+- `libp2p: Libp2p`, instance of the local network stack.
+- `blockstore: Blockstore`, instance of the local database (`IpfsRepo.blockstore`)
+
+Create a new instance.
+
+#### `getStream(cid)`
+
+- `cid: CID|Array`
+
+Returns a source `pull-stream`. Values emitted are the received blocks.
+
+Example:
+
+```js
+// Single block
+pull(
+  bitswap.getStream(cid),
+  pull.collect((err, blocks) => {
+    // blocks === [block]
+  })
+)
+
+// Many blocks
+pull(
+  bitswap.getStream([cid1, cid2, cid3]),
+  pull.collect((err, blocks) => {
+    // blocks === [block1, block2, block3]
+  })
+)
+```
+
+> Note: This is safe guarded so that the network is not asked
+> for blocks that are in the local `datastore`.
+
+#### `unwant(cids)`
+
+- `cids: CID|[]CID`
+
+Cancel previously requested cids, forcefully. That means they are removed from the
+wantlist independent of how many other resources requested these cids. Callbacks
+attached to `getBlock` are errored with `Error('manual unwant: cid)`.
+
+#### `cancelWants(cids)`
+
+- `cid: CID|[]CID`
+
+Cancel previously requested cids.
+
+#### `putStream()`
+
+Returns a duplex `pull-stream` that emits an object `{cid: CID}` for every written block when it was stored.
+Objects passed into here should be of the form `{data: Buffer, cid: CID}`
+
+#### `put(blockAndCid, callback)`
+
+- `blockAndCid: {data: Buffer, cid: CID}`
+- `callback: Function`
+
+Announce that the current node now has the block containing `data`. This will store it
+in the local database and attempt to serve it to all peers that are known
+ to have requested it. The callback is called when we are sure that the block
+ is stored.
+
+#### `wantlistForPeer(peerId)`
+
+- `peerId: PeerId`
+
+Get the wantlist for a given peer.
+
+#### `stat()`
+
+Get stats about about the current state of the bitswap instance.
+
+## Development
+
+### Structure
+
+![](/img/architecture.png)
+
+```sh
+» tree src
+src
+├── components
+│   ├── decision
+│   │   ├── engine.js
+│   │   ├── index.js
+│   │   └── ledger.js
+│   ├── network             # Handles peerSet and open new conns
+│   │   └── index.js
+│   └── want-manager        # Keeps track of all blocks the peer wants (not the others which it is connected)
+│       ├── index.js
+│       └── msg-queue.js    # Messages to send queue, one per peer
+├── constants.js
+├── index.js
+└── types
+    ├── message             # (Type) message that is put in the wire
+    │   ├── entry.js
+    │   ├── index.js
+    │   └── message.proto.js
+    └── wantlist            # (Type) track wanted blocks
+        ├── entry.js
+        └── index.js
+```
+
 ## Contribute
 
 Feel free to join in. All welcome. Open an [issue](https://github.com/ipfs/js-ipfs-bitswap/issues)!

diff --git a/benchmarks/index.js b/benchmarks/index.js
@@ -0,0 +1,110 @@
+/* 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 CID = require('cids')
+
+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)
+    }
+    const cids = keys.map((k) => new CID(k))
+    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 {
+            block: blocks[index],
+            cid: cids[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(cids),
+          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))
+  })
+}