Builder refactoring, trickle builder and balanced builder (#118)

* builder refactoring. trickle builder. balanced builder

* removed unused experimental builder

* documented importer options

* default builder strategy is now the balanced strategy

* removed unused test

* removed superfluous comment

* fixed trickle builder

* removed superfluous comment

* using options.chunkerOptions for chunker-specific options

* docs: corrected option name

* fix: error handling in trickle reducer

* using pull-pair instead of backpressure-less bespoke pair

* fixed trickle builder tests

* recursive streaming trickle builder

* missing dep

* some style corrections

* importing multiple roots yields an error

* reinstated testing importing using flat and balanced strategies

* asserting that root node is one and only one

* testing import and export using various builder strategies

* fixed error propagation into push streams

* simplified some iteration logic

* default for maximum children pre node is 174

* by default, only reduces one leaf to self if specific option is present

* test results reflect new default config

* testing against big files genearted from a pseudo random byte stream gen

* added missing dep

* removed unnecessary dev dependency

* go-ipfs parity: no root node with single leaf

* docs: corrected the default maximum number of children nodes

Author: pgte

README.md

@@ -150,6 +150,20 @@ been written into the [DAG Service][]'s storage mechanism.
 The input's file paths and directory structure will be preserved in the DAG
 Nodes.
 
+### Importer options
+
+In the second argument of the importer constructor you can specify the following options:
+
+* `chunker` (string, defaults to `"fixed"`): the chunking strategy. Now only supports `"fixed"`
+* `chunkerOptions` (object, optional): the options for the chunker. Defaults to an object with the following properties:
+  * `maxChunkSize` (positive integer, defaults to `262144`): the maximum chunk size for the `fixed` chunker.
+* `strategy` (string, defaults to `"balanced"`): the DAG builder strategy name. Supports:
+  * `flat`: flat list of chunks
+  * `balanced`: builds a balanced tree
+  * `trickle`: builds [a trickle tree](https://github.com/ipfs/specs/pull/57#issuecomment-265205384)
+* `maxChildrenPerNode` (positive integer, defaults to `174`): the maximum children per node for the `balanced` and `trickle` DAG builder strategies
+* `layerRepeat` (positive integer, defaults to 4): (only applicable to the `trickle` DAG builder strategy). The maximum repetition of parent nodes for each layer of the tree.
+* `reduceSingleLeafToSelf` (boolean, defaults to `false`): optimization for, when reducing a set of nodes with one node, reduce it to that node.
 
 ### Example Exporter
 

package.json

@@ -48,19 +48,24 @@
     "ipfs-repo": "^0.11.2",
     "ncp": "^2.0.0",
     "pre-commit": "^1.2.2",
+    "pull-generate": "^2.2.0",
     "pull-zip": "^2.0.1",
     "rimraf": "^2.5.4"
   },
   "dependencies": {
     "async": "^2.1.4",
     "cids": "^0.3.5",
+    "deep-extend": "^0.4.1",
     "ipfs-unixfs": "^0.1.9",
     "ipld-dag-pb": "^0.9.3",
     "ipld-resolver": "^0.4.1",
     "is-ipfs": "^0.2.1",
     "multihashes": "^0.3.1",
+    "pull-batch": "^1.0.0",
     "pull-block": "^1.0.2",
+    "pull-pair": "^1.1.0",
     "pull-paramap": "^1.2.1",
+    "pull-pause": "0.0.0",
     "pull-pushable": "^2.0.1",
     "pull-stream": "^3.5.0",
     "pull-traverse": "^1.0.3",
@@ -76,4 +81,4 @@
     "jbenet <[email protected]>",
     "nginnever <[email protected]>"
   ]
-}
+}

src/builder/balanced/balanced-reducer.js

@@ -0,0 +1,53 @@
+'use strict'
+
+const assert = require('assert')
+const pull = require('pull-stream')
+const pushable = require('pull-pushable')
+const pullPair = require('pull-pair')
+const batch = require('pull-batch')
+
+module.exports = function balancedReduceToRoot (reduce, options) {
+  const pair = pullPair()
+  const source = pair.source
+
+  const result = pushable()
+
+  reduceToParents(source, (err, roots) => {
+    if (err) {
+      result.end(err)
+      return // early
+    }
+    assert.equal(roots.length, 1, 'need one root')
+    result.push(roots[0])
+    result.end()
+  })
+
+  function reduceToParents (_chunks, callback) {
+    let chunks = _chunks
+    if (Array.isArray(chunks)) {
+      chunks = pull.values(chunks)
+    }
+
+    pull(
+      chunks,
+      batch(options.maxChildrenPerNode),
+      pull.asyncMap(reduce),
+      pull.collect(reduced)
+    )
+
+    function reduced (err, roots) {
+      if (err) {
+        callback(err)
+      } else if (roots.length > 1) {
+        reduceToParents(roots, callback)
+      } else {
+        callback(null, roots)
+      }
+    }
+  }
+
+  return {
+    sink: pair.sink,
+    source: result
+  }
+}

src/builder/balanced/index.js

@@ -0,0 +1,12 @@
+'use strict'
+
+const balancedReducer = require('./balanced-reducer')
+
+const defaultOptions = {
+  maxChildrenPerNode: 174
+}
+
+module.exports = function (reduce, _options) {
+  const options = Object.assign({}, defaultOptions, _options)
+  return balancedReducer(reduce, options)
+}

src/builder/builder.js

@@ -0,0 +1,150 @@
+'use strict'
+
+const extend = require('deep-extend')
+const assert = require('assert')
+const UnixFS = require('ipfs-unixfs')
+const pull = require('pull-stream')
+const through = require('pull-through')
+const parallel = require('async/parallel')
+const waterfall = require('async/waterfall')
+const dagPB = require('ipld-dag-pb')
+const CID = require('cids')
+
+const reduce = require('./reduce')
+
+const DAGNode = dagPB.DAGNode
+
+const defaultOptions = {
+  chunkerOptions: {
+    maxChunkSize: 262144
+  }
+}
+
+module.exports = function (createChunker, ipldResolver, createReducer, _options) {
+  const options = extend({}, defaultOptions, _options)
+
+  return function (source, files) {
+    return function (items, cb) {
+      parallel(items.map((item) => (cb) => {
+        if (!item.content) {
+          // item is a directory
+          return createAndStoreDir(item, (err, node) => {
+            if (err) {
+              return cb(err)
+            }
+            source.push(node)
+            files.push(node)
+            cb()
+          })
+        }
+
+        // item is a file
+        createAndStoreFile(item, (err, node) => {
+          if (err) {
+            return cb(err)
+          }
+          source.push(node)
+          files.push(node)
+          cb()
+        })
+      }), cb)
+    }
+  }
+
+  function createAndStoreDir (item, callback) {
+    // 1. create the empty dir dag node
+    // 2. write it to the dag store
+
+    const d = new UnixFS('directory')
+    waterfall([
+      (cb) => DAGNode.create(d.marshal(), cb),
+      (node, cb) => {
+        ipldResolver.put({
+          node: node,
+          cid: new CID(node.multihash)
+        }, (err) => cb(err, node))
+      }
+    ], (err, node) => {
+      if (err) {
+        return callback(err)
+      }
+      callback(null, {
+        path: item.path,
+        multihash: node.multihash,
+        size: node.size
+      })
+    })
+  }
+
+  function createAndStoreFile (file, callback) {
+    if (Buffer.isBuffer(file.content)) {
+      file.content = pull.values([file.content])
+    }
+
+    if (typeof file.content !== 'function') {
+      return callback(new Error('invalid content'))
+    }
+
+    const reducer = createReducer(reduce(file, ipldResolver, options), options)
+
+    let previous
+    let count = 0
+
+    pull(
+      file.content,
+      createChunker(options.chunkerOptions),
+      pull.map(chunk => new Buffer(chunk)),
+      pull.map(buffer => new UnixFS('file', buffer)),
+      pull.asyncMap((fileNode, callback) => {
+        DAGNode.create(fileNode.marshal(), (err, node) => {
+          callback(err, { DAGNode: node, fileNode: fileNode })
+        })
+      }),
+      pull.asyncMap((leaf, callback) => {
+        ipldResolver.put(
+          {
+            node: leaf.DAGNode,
+            cid: new CID(leaf.DAGNode.multihash)
+          },
+          err => callback(err, leaf)
+        )
+      }),
+      pull.map((leaf) => {
+        return {
+          path: file.path,
+          multihash: leaf.DAGNode.multihash,
+          size: leaf.DAGNode.size,
+          leafSize: leaf.fileNode.fileSize(),
+          name: ''
+        }
+      }),
+      through( // mark as single node if only one single node
+        function onData (data) {
+          count++
+          if (previous) {
+            this.queue(previous)
+          }
+          previous = data
+        },
+        function ended () {
+          if (previous) {
+            if (count === 1) {
+              previous.single = true
+            }
+            this.queue(previous)
+          }
+          this.queue(null)
+        }
+      ),
+      reducer,
+      pull.collect((err, roots) => {
+        if (err) {
+          callback(err)
+        } else {
+          assert.equal(roots.length, 1, 'should result in exactly one root')
+          callback(null, roots[0])
+        }
+      })
+    )
+  }
+}

src/builder/create-build-stream.js

@@ -0,0 +1,29 @@
+'use strict'
+
+const pullPushable = require('pull-pushable')
+const pullWrite = require('pull-write')
+
+module.exports = function createBuildStream (createStrategy, ipldResolver, flushTree, options) {
+  const files = []
+
+  const source = pullPushable()
+
+  const sink = pullWrite(
+    createStrategy(source, files),
+    null,
+    options.highWaterMark,
+    (err) => {
+      if (err) {
+        source.end(err)
+        return // early
+      }
+
+      flushTree(files, ipldResolver, source, source.end)
+    }
+  )
+
+  return {
+    source: source,
+    sink: sink
+  }
+}

src/builder/flat/index.js

@@ -0,0 +1,33 @@
+'use strict'
+
+const assert = require('assert')
+const pull = require('pull-stream')
+const pushable = require('pull-pushable')
+const pullPair = require('pull-pair')
+const batch = require('pull-batch')
+
+module.exports = function (reduce, options) {
+  const pair = pullPair()
+  const source = pair.source
+  const result = pushable()
+
+  pull(
+    source,
+    batch(Infinity),
+    pull.asyncMap(reduce),
+    pull.collect((err, roots) => {
+      if (err) {
+        result.end(err)
+        return // early
+      }
+      assert.equal(roots.length, 1, 'need one root')
+      result.push(roots[0])
+      result.end()
+    })
+  )
+
+  return {
+    sink: pair.sink,
+    source: result
+  }
+}

src/builder/index.js

@@ -0,0 +1,33 @@
+'use strict'
+
+const assert = require('assert')
+const createBuildStream = require('./create-build-stream')
+const Builder = require('./builder')
+
+const reducers = {
+  flat: require('./flat'),
+  balanced: require('./balanced'),
+  trickle: require('./trickle')
+}
+
+const defaultOptions = {
+  strategy: 'balanced',
+  highWaterMark: 100,
+  reduceSingleLeafToSelf: false
+}
+
+module.exports = function (Chunker, ipldResolver, flushTree, _options) {
+  assert(Chunker, 'Missing chunker creator function')
+  assert(ipldResolver, 'Missing IPLD Resolver')
+  assert(flushTree, 'Missing flushTree argument')
+
+  const options = Object.assign({}, defaultOptions, _options)
+
+  const strategyName = options.strategy
+  const reducer = reducers[strategyName]
+  assert(reducer, 'Unknown importer build strategy name: ' + strategyName)
+
+  const createStrategy = Builder(Chunker, ipldResolver, reducer, options)
+
+  return createBuildStream(createStrategy, ipldResolver, flushTree, options)
+}