parallelize block processing
Jeromy committed
May 27, 2015 at 19:03 UTC
11de3643840752708d2473400954fa224150ea35
1 file changed
+28
-26
exchange/bitswap/bitswap.go
+28
-26
@@ -279,39 +279,41 @@ func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg
279
// quickly send out cancels, reduces chances of duplicate block receives
280
var keys []u.Key
281
for _, block := range iblocks {
282
- keys = append(keys, block.Key())
283
- }
284
- bs.wm.CancelWants(keys)
285
-
286
- for _, block := range iblocks {
287
- bs.counterLk.Lock()
288
- bs.blocksRecvd++
289
- has, err := bs.blockstore.Has(block.Key())
290
- if err == nil && has {
291
- bs.dupBlocksRecvd++
292
- }
293
- brecvd := bs.blocksRecvd
294
- bdup := bs.dupBlocksRecvd
295
- bs.counterLk.Unlock()
296
- if has {
297
- continue
298
- }
299
-
300
- // put this after the duplicate check as a block not on our wantlist may
301
- // have already been received.
282
if _, found := bs.wm.wl.Contains(block.Key()); !found {
283
log.Notice("received un-asked-for block: %s", block)
284
continue
285
}
286
+ keys = append(keys, block.Key())
287
+ }
288
+ bs.wm.CancelWants(keys)
289
307
- log.Infof("got block %s from %s (%d,%d)", block, p, brecvd, bdup)
290
+ wg := sync.WaitGroup{}
291
+ for _, block := range iblocks {
292
+ wg.Add(1)
293
+ go func(b *blocks.Block) {
294
+ defer wg.Done()
295
+ bs.counterLk.Lock()
296
+ bs.blocksRecvd++
297
+ has, err := bs.blockstore.Has(b.Key())
298
+ if err == nil && has {
299
+ bs.dupBlocksRecvd++
300
+ }
301
+ brecvd := bs.blocksRecvd
302
+ bdup := bs.dupBlocksRecvd
303
+ bs.counterLk.Unlock()
304
+ if has {
305
+ return
306
+ }
307
309
- hasBlockCtx, cancel := context.WithTimeout(ctx, hasBlockTimeout)
310
- if err := bs.HasBlock(hasBlockCtx, block); err != nil {
311
- log.Warningf("ReceiveMessage HasBlock error: %s", err)
312
- }
313
- cancel()
308
+ log.Debugf("got block %s from %s (%d,%d)", b, p, brecvd, bdup)
309
+ hasBlockCtx, cancel := context.WithTimeout(ctx, hasBlockTimeout)
310
+ if err := bs.HasBlock(hasBlockCtx, b); err != nil {
311
+ log.Warningf("ReceiveMessage HasBlock error: %s", err)
312
+ }
313
+ cancel()
314
+ }(block)
315
}
316
+ wg.Wait()
317
}
318
319
// Connected/Disconnected warns bitswap about peer connections