Continuous hashing

bin/cmd.js

@@ -451,18 +451,22 @@ function drawTorrent (torrent) {
       }
       var bar = ''
       for (var j = 0; j < piece.blocks.length; j++) {
-        switch(piece.blocks[j]) {
-        case 0:
-          bar += '{red:█}';
-          break;
-        case 1:
-          bar += '{blue:█}';
-          break;
-        case 2:
+        if (j < piece.blocksHashed) {
           bar += '{green:█}';
-          break;
-        default:
-          throw 'Invalid block state: ' + piece.blocks[j]
+        } else {
+          switch(piece.blocks[j]) {
+          case 0:
+            bar += '{red:█}';
+            break;
+          case 1:
+            bar += '{yellow:█}';
+            break;
+          case 2:
+            bar += '{blue:█}';
+            break;
+          default:
+            throw 'Invalid block state: ' + piece.blocks[j]
+          }
         }
       }
       clivas.line('{4+cyan:' + i + '} ' + bar);

lib/sha1/index.js

@@ -0,0 +1,28 @@
+var through2 = require('through2')
+var crypto = require('crypto')
+
+
+// encapsulated a crypto stream in order to:
+// * lazily instantiate the underlying implementation
+// * move to webworkers later on
+module.exports = function SHA1 () {
+  var hash
+  function spawnOnDemand () {
+    if (!hash)
+      hash = crypto.createHash('sha1')
+  }
+
+  var self = through2(function (buffer, enc, callback) {
+    spawnOnDemand()    
+    hash.update(buffer)
+    callback()
+  }, function (callback) {
+    spawnOnDemand()
+    var digest = hash.digest('hex')
+    self.hexDigest = digest
+    this.push(digest)
+    this.push(null)
+    callback()
+  })
+  return self
+}

lib/storage.js

@@ -11,7 +11,7 @@ var FileStream = require('./file-stream')
 var inherits = require('inherits')
 var MultiStream = require('multistream')
 var once = require('once')
-var sha1 = require('simple-sha1')
+var sha1 = require('./sha1')
 
 var BLOCK_LENGTH = 16 * 1024
 
@@ -74,13 +74,11 @@ Piece.prototype.writeBlock = function (offset, buffer, cb) {
     return cb(null)
   }
 
+  debug('piece ' + self.index + ' writeBlock@' + offset + '+' + buffer.length)
   buffer.copy(self.buffer, offset)
   self.blocks[i] = BLOCK_WRITTEN
   self.blocksWritten += 1
-
-  if (self.blocksWritten === self.blocks.length) {
-    self.verify()
-  }
+  self._hashNext()
 
   cb(null)
 }
@@ -123,38 +121,57 @@ Piece.prototype._reset = function () {
   self.blocks = new Buffer(Math.ceil(self.length / BLOCK_LENGTH))
   self.blocks.fill(0)
   self.blocksWritten = 0
+  self.sha = sha1()
+  // pull any received blocks
+  self.sha.on('drain', self._hashNext.bind(self))
+  // triggered by _hashNext():
+  self.sha.on('end', self._onHashed.bind(self))
+  self.blocksHashed = 0
+  // put into flowing mode
+  // we don't provide a sink, just waiting for 'end'
+  self.sha.resume()
 }
 
-Piece.prototype.verify = function (buffer) {
-  var self = this
-  buffer = buffer || self.buffer
-  if (self.verified || !buffer) {
+// called by writeBlock()
+Piece.prototype._hashNext = function () {
+  if (this.verified || !this.buffer) {
     return
   }
 
-  if (self.noVerify) {
-    self.verified = true
-    onResult()
-    return
-  }
-
-  var expectedHash
-  sha1(buffer, function (_expectedHash) {
-    expectedHash = _expectedHash
-    self.verified = (expectedHash === self.hash)
-    onResult()
-  })
-
-  function onResult () {
-    if (self.verified) {
-      self.emit('done')
+  debug('piece ' + this.index + ' _hashNext: blocksHashed=' + this.blocksHashed + '/' + this.blocks.length)
+  if (this.blocks[this.blocksHashed] === BLOCK_WRITTEN) {
+    debug('piece ' + this.index + ' _hashNext: write')
+    var block = this.buffer.slice(this.blocksHashed * BLOCK_LENGTH, (this.blocksHashed + 1) * BLOCK_LENGTH)
+    this.blocksHashed += 1
+    if (this.sha.write(block)) {
+      debug('piece ' + this.index + ' _hashNext: more')
+      // recurse
+      process.nextTick(this._hashNext.bind(this))
     } else {
-      self.emit('warning', new Error('piece ' + self.index + ' failed verification; ' + expectedHash + ' expected ' + self.hash))
-      self._reset()
+      debug('piece ' + this.index + ' _hashNext: wait for sha readable')
     }
+  } else if (this.blocksHashed >= this.blocks.length) {
+    // will trigger 'done' event
+    debug('piece ' + this.index + ' _hashNext: end')
+    this.sha.end()
+  } else {
+    debug('piece ' + this.index + ' _hashNext: wait more data')
+  }
+}
+
+Piece.prototype._onHashed = function () {
+  debug('piece ' + this.index + ' verified: ' + this.sha.hexDigest + ' == ' + this.hash)
+  this.verified = (this.sha.hexDigest === this.hash)
+
+  if (this.verified) {
+    this.emit('done')
+  } else {
+    this.emit('warning', new Error('piece ' + this.index + ' failed verification; ' + this.sha.hexDigest + ' expected ' + this.hash))
+    this._reset()
   }
 }
 
+// validates a readBlock()/writeBlock() offset
 Piece.prototype._verifyOffset = function (offset) {
   var self = this
   if (offset % BLOCK_LENGTH === 0) {
@@ -165,6 +182,7 @@ Piece.prototype._verifyOffset = function (offset) {
   }
 }
 
+// validates a writeBlock() buffer length
 Piece.prototype._verifyBlock = function (offset, buffer) {
   var self = this
   if (buffer.length === BLOCK_LENGTH) {

package.json

@@ -19,7 +19,8 @@
     "bittorrent-swarm": "webtorrent-swarm",
     "load-ip-set": false,
     "simple-get": false,
-    "ut_pex": false
+    "ut_pex": false,
+    "crypto": "crypto-browserify"
   },
   "browserify": {
     "transform": [
@@ -38,6 +39,7 @@
     "browserify-versionify": "^1.0.2",
     "clivas": "^0.1.4",
     "create-torrent": "^3.4.0",
+    "crypto-browserify": "^3.9.9",
     "debug": "^2.1.0",
     "dezalgo": "^1.0.1",
     "end-of-stream": "^1.0.0",
@@ -61,7 +63,6 @@
     "rimraf": "^2.2.5",
     "run-parallel": "^1.0.0",
     "simple-get": "^1.0.0",
-    "simple-sha1": "^2.0.0",
     "speedometer": "^0.1.2",
     "thunky": "^0.1.0",
     "torrent-discovery": "^2.0.1",