Merge pull request #860 from jbenet/fix/hasblock-explosion

implement a worker to consolidate HasBlock provide calls into one

Author: jbenet

exchange/bitswap/bitswap.go

@@ -89,6 +89,7 @@ func New(parent context.Context, p peer.ID, network bsnet.BitSwapNetwork,
 		batchRequests: make(chan *blockRequest, sizeBatchRequestChan),
 		process:       px,
 		newBlocks:     make(chan *blocks.Block, HasBlockBufferSize),
+		provideKeys:   make(chan u.Key),
 	}
 	network.SetDelegate(bs)
 
@@ -124,6 +125,8 @@ type Bitswap struct {
 	process process.Process
 
 	newBlocks chan *blocks.Block
+
+	provideKeys chan u.Key
 }
 
 type blockRequest struct {

exchange/bitswap/workers.go

@@ -6,6 +6,7 @@ import (
 	inflect "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/chuckpreslar/inflect"
 	process "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
 	context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
+	u "github.com/jbenet/go-ipfs/util"
 )
 
 func (bs *Bitswap) startWorkers(px process.Process, ctx context.Context) {
@@ -24,6 +25,10 @@ func (bs *Bitswap) startWorkers(px process.Process, ctx context.Context) {
 		bs.rebroadcastWorker(ctx)
 	})
 
+	px.Go(func(px process.Process) {
+		bs.provideCollector(ctx)
+	})
+
 	// Spawn up multiple workers to handle incoming blocks
 	// consider increasing number if providing blocks bottlenecks
 	// file transfers
@@ -58,13 +63,13 @@ func (bs *Bitswap) taskWorker(ctx context.Context) {
 func (bs *Bitswap) provideWorker(ctx context.Context) {
 	for {
 		select {
-		case blk, ok := <-bs.newBlocks:
+		case k, ok := <-bs.provideKeys:
 			if !ok {
-				log.Debug("newBlocks channel closed")
+				log.Debug("provideKeys channel closed")
 				return
 			}
 			ctx, _ := context.WithTimeout(ctx, provideTimeout)
-			err := bs.network.Provide(ctx, blk.Key())
+			err := bs.network.Provide(ctx, k)
 			if err != nil {
 				log.Error(err)
 			}
@@ -74,6 +79,38 @@ func (bs *Bitswap) provideWorker(ctx context.Context) {
 	}
 }
 
+func (bs *Bitswap) provideCollector(ctx context.Context) {
+	defer close(bs.provideKeys)
+	var toProvide []u.Key
+	var nextKey u.Key
+	var keysOut chan u.Key
+
+	for {
+		select {
+		case blk, ok := <-bs.newBlocks:
+			if !ok {
+				log.Debug("newBlocks channel closed")
+				return
+			}
+			if keysOut == nil {
+				nextKey = blk.Key()
+				keysOut = bs.provideKeys
+			} else {
+				toProvide = append(toProvide, blk.Key())
+			}
+		case keysOut <- nextKey:
+			if len(toProvide) > 0 {
+				nextKey = toProvide[0]
+				toProvide = toProvide[1:]
+			} else {
+				keysOut = nil
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
 // TODO ensure only one active request per key
 func (bs *Bitswap) clientWorker(parent context.Context) {
 	defer log.Info("bitswap client worker shutting down...")