Builder refactoring, trickle builder and balanced builder

README.md

@@ -150,6 +150,19 @@ 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"`
+* `chunkSize` (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 `172`): 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.
+
 
 ### Example Exporter
 

package.json

@@ -42,6 +42,7 @@
     "aegir": "^9.3.0",
     "buffer-loader": "0.0.1",
     "chai": "^3.5.0",
+    "deep-extend": "^0.4.1",
     "fs-pull-blob-store": "^0.4.1",
     "idb-pull-blob-store": "^0.5.1",
     "ipfs-block-service": "^0.8.0",
@@ -59,6 +60,7 @@
     "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-paramap": "^1.2.1",
     "pull-pushable": "^2.0.1",
@@ -76,4 +78,4 @@
     "jbenet <juan@benet.ai>",
     "nginnever <ginneversource@gmail.com>"
   ]
-}
+}

src/builder/balanced/balanced-reducer.js

@@ -0,0 +1,64 @@
+'use strict'
+
+const assert = require('assert')
+const pull = require('pull-stream')
+const pullWrite = require('pull-write')
+const pushable = require('pull-pushable')
+const batch = require('pull-batch')
+
+module.exports = function balancedReduceToRoot (reduce, options) {
+  const source = pushable()
+
+  const sink = pullWrite(
+    function (item, cb) {
+      source.push(item)
+      cb()
+    },
+    null,
+    1,
+    function (end) {
+      source.end(end)
+    }
+  )
+
+  const result = pushable()
+
+  reduceToParents(source, function (err, roots) {
+    if (err) {
+      result.emit('error', err)
+      return // early
+    }
+    assert.equal(roots.length, 1, 'no 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: sink,
+    source: result
+  }
+}

src/builder/balanced/index.js

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

src/builder/builder.js

@@ -0,0 +1,126 @@
+'use strict'
+
+const assert = require('assert')
+const UnixFS = require('ipfs-unixfs')
+const pull = require('pull-stream')
+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 = {
+  chunkSize: 262144
+  // chunkSize: 26214
+}
+
+module.exports = function (Chunker, ipldResolver, Reducer, _options) {
+  const options = Object.assign({}, 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 = Reducer(reduce(file, ipldResolver), options)
+
+    pull(
+      file.content,
+      Chunker(options),
+      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: ''
+        }
+      }),
+      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,30 @@
+'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) {
+        return source.end(err)
+      }
+
+      flushTree(files, ipldResolver, source, () => {
+        source.end()
+      })
+    }
+  )
+
+  return {
+    source: source,
+    sink: sink
+  }
+}

src/builder/flat/index.js

@@ -0,0 +1,46 @@
+'use strict'
+
+const pull = require('pull-stream')
+const pushable = require('pull-pushable')
+const pullWrite = require('pull-write')
+const batch = require('pull-batch')
+
+module.exports = function (reduce, options) {
+  const source = pushable()
+  const sink = pullWrite(
+    function (d, cb) {
+      source.push(d)
+      cb()
+    },
+    null,
+    1,
+    function (err) {
+      if (err) {
+        source.emit('error', err)
+      } else {
+        source.end()
+      }
+    }
+  )
+
+  const result = pushable()
+
+  pull(
+    source,
+    batch(Infinity),
+    pull.asyncMap(reduce),
+    pull.collect(function (err, roots) {
+      if (err) {
+        result.emit('error', err)
+        return // early
+      }
+      result.push(roots[0])
+      result.end()
+    })
+  )
+
+  return {
+    sink: sink,
+    source: result
+  }
+}

src/builder/index.js

@@ -0,0 +1,32 @@
+'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
+}
+
+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)
+}