allow bitswap to attempt to write blocks to disk multiple times
License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Jul 13, 2015 at 11:01 UTC
7ef1a470c70c35af09e5a3497cce225645109223
1 file changed
+41
-19
exchange/bitswap/bitswap.go
+41
-19
@@ -228,7 +228,9 @@ func (bs *Bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
228
default:
229
}
230
231
- if err := bs.blockstore.Put(blk); err != nil {
231
+ err := bs.tryPutBlock(blk, 4) // attempt to store block up to four times
232
+ if err != nil {
233
+ log.Errorf("Error writing block to datastore: %s", err)
234
return err
235
}
236
@@ -242,6 +244,18 @@ func (bs *Bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
244
return nil
245
}
246
247
+func (bs *Bitswap) tryPutBlock(blk *blocks.Block, attempts int) error {
248
+ var err error
249
+ for i := 0; i < attempts; i++ {
250
+ if err = bs.blockstore.Put(blk); err == nil {
251
+ break
252
+ }
253
+
254
+ time.Sleep(time.Millisecond * time.Duration(400*(i+1)))
255
+ }
256
+ return err
257
+}
258
+
259
func (bs *Bitswap) connectToProviders(ctx context.Context, entries []wantlist.Entry) {
260
261
ctx, cancel := context.WithCancel(ctx)
@@ -297,38 +311,46 @@ func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg
311
wg.Add(1)
312
go func(b *blocks.Block) {
313
defer wg.Done()
300
- bs.counterLk.Lock()
301
- bs.blocksRecvd++
302
- has, err := bs.blockstore.Has(b.Key())
303
- if err != nil {
304
- bs.counterLk.Unlock()
305
- log.Infof("blockstore.Has error: %s", err)
306
- return
307
- }
308
- if err == nil && has {
309
- bs.dupBlocksRecvd++
310
- }
311
- brecvd := bs.blocksRecvd
312
- bdup := bs.dupBlocksRecvd
313
- bs.counterLk.Unlock()
314
- if has {
315
- return
314
+
315
+ if err := bs.updateReceiveCounters(b.Key()); err != nil {
316
+ return // ignore error, is either logged previously, or ErrAlreadyHaveBlock
317
}
318
319
k := b.Key()
320
log.Event(ctx, "Bitswap.GetBlockRequest.End", &k)
321
321
- log.Debugf("got block %s from %s (%d,%d)", b, p, brecvd, bdup)
322
+ log.Debugf("got block %s from %s", b, p)
323
hasBlockCtx, cancel := context.WithTimeout(ctx, hasBlockTimeout)
324
+ defer cancel()
325
if err := bs.HasBlock(hasBlockCtx, b); err != nil {
326
log.Warningf("ReceiveMessage HasBlock error: %s", err)
327
}
326
- cancel()
328
}(block)
329
}
330
wg.Wait()
331
}
332
333
+var ErrAlreadyHaveBlock = errors.New("already have block")
334
+
335
+func (bs *Bitswap) updateReceiveCounters(k key.Key) error {
336
+ bs.counterLk.Lock()
337
+ defer bs.counterLk.Unlock()
338
+ bs.blocksRecvd++
339
+ has, err := bs.blockstore.Has(k)
340
+ if err != nil {
341
+ log.Infof("blockstore.Has error: %s", err)
342
+ return err
343
+ }
344
+ if err == nil && has {
345
+ bs.dupBlocksRecvd++
346
+ }
347
+
348
+ if has {
349
+ return ErrAlreadyHaveBlock
350
+ }
351
+ return nil
352
+}
353
+
354
// Connected/Disconnected warns bitswap about peer connections
355
func (bs *Bitswap) PeerConnected(p peer.ID) {
356
bs.wm.Connected(p)