Merge pull request #67 from ipfs/fixes

fix: better accounting and less logs

Author: dignifiedquire

src/decision/engine.js

@@ -66,7 +66,7 @@ module.exports = class Engine {
         return Boolean(nextTask)
       },
       (next) => {
-        log('got task', nextTask)
+        log('got task')
 
         pull(
           this.blockstore.getStream(nextTask.entry.key),
@@ -121,18 +121,19 @@ module.exports = class Engine {
         log.error(`failed to process blocks: ${err.message}`)
       }
 
-      log('wantlist', Array.from(msg.wantlist.values()).map((e) => e.toString()))
+      const arrayWantlist = Array.from(msg.wantlist.values())
+      log('wantlist', arrayWantlist.map((e) => e.toString()))
+
+      if (arrayWantlist.length === 0) {
+        return cb()
+      }
 
       pull(
-        pull.values(Array.from(msg.wantlist.values())),
+        pull.values(arrayWantlist),
         pull.asyncMap((entry, cb) => {
           this._processWantlist(ledger, peerId, entry, cb)
         }),
-        pull.onEnd((err) => {
-          if (err) return cb(err)
-          this._outbox()
-          cb()
-        })
+        pull.onEnd(cb)
       )
     })
   }
@@ -146,7 +147,6 @@ module.exports = class Engine {
     // Check all connected peers if they want the block we received
     for (let l of this.ledgerMap.values()) {
       const entry = l.wantlistContains(key)
-
       if (entry) {
         this.peerRequestQueue.push(entry, l.partner)
       }
@@ -170,6 +170,7 @@ module.exports = class Engine {
         } else if (exists) {
           log('has want %s', mh.toB58String(entry.key))
           this.peerRequestQueue.push(entry.entry, peerId)
+          this._outbox()
         }
         cb()
       })

src/decision/peer-request-queue.js

@@ -98,14 +98,14 @@ module.exports = class PeerRequestQueue {
     log('taskMap.set', task.key, task.toString())
     this.taskMap.set(task.key, task)
     partner.requests ++
-    partner.taskQueue.update(task)
+    this.pQueue.update(partner)
   }
 
   // Get the task with the hightest priority from the queue
   pop () {
-    // log('pop, empty? %s', this.pQueue.isEmpty())
-    // log('partners', Array.from(this.partners.values()).map((val) => [val.requests, val.taskQueue.size()]))
-    if (this.pQueue.isEmpty()) return
+    if (this.pQueue.isEmpty()) {
+      return
+    }
 
     let partner = this.pQueue.pop()
     let out
@@ -122,7 +122,6 @@ module.exports = class PeerRequestQueue {
       partner.requests --
       break
     }
-    // log('pop, out', partner.taskQueue.isEmpty(), out)
     this.pQueue.push(partner)
     return out
   }
@@ -137,7 +136,9 @@ module.exports = class PeerRequestQueue {
       t.trash = true
 
       // having canceled a block, we now account for that in the given partner
-      this.partners.get(peerId.toB58String()).requests --
+      const p = this.partners.get(peerId.toB58String())
+      p.requests --
+      this.pQueue.update(p)
     }
 
     log('taskMap', Array.from(this.taskMap.values()).map((v) => {

src/index.js

@@ -162,7 +162,6 @@ module.exports = class Bitwap {
   }
 
   getStream (keys) {
-    log('getStream', keys.length)
     if (!Array.isArray(keys)) {
       return this._getStreamSingle(keys)
     }
@@ -180,7 +179,6 @@ module.exports = class Bitwap {
   }
 
   _getStreamSingle (key) {
-    log('getStreamSingle', mh.toB58String(key))
     const unwantListeners = {}
     const blockListeners = {}
     const unwantEvent = (key) => `unwant:${key}`

src/wantmanager/index.js

@@ -46,7 +46,9 @@ module.exports = class Wantmanager {
         }
       }),
       pull.collect((err, entries) => {
-        if (err) throw err
+        if (err) {
+          throw err
+        }
         // broadcast changes
         for (let p of this.peers.values()) {
           p.addEntries(entries, false)
@@ -108,7 +110,6 @@ module.exports = class Wantmanager {
 
   // cancel wanting all of the given keys
   cancelWants (keys) {
-    log('keys', keys)
     log('cancel wants: ', keys.map((k) => mh.toB58String(k)))
     this._addEntries(keys, true)
   }