Merge pull request #409 from jbenet/no-dupe-send

prevent sending of same block to a peer twice

exchange/bitswap/bitswap.go

@@ -307,6 +307,7 @@ func (bs *bitswap) ReceiveMessage(ctx context.Context, p peer.Peer, incoming bsm
 
 				blkmsg.AddBlock(block)
 				bs.send(ctx, p, blkmsg)
+				bs.strategy.BlockSentToPeer(block.Key(), p)
 			}
 		}
 	}

exchange/bitswap/strategy/interface.go

@@ -32,6 +32,8 @@ type Strategy interface {
 
 	NumBytesReceivedFrom(peer.Peer) uint64
 
+	BlockSentToPeer(u.Key, peer.Peer)
+
 	// Values determining bitswap behavioural patterns
 	GetBatchSize() int
 	GetRebroadcastDelay() time.Duration

exchange/bitswap/strategy/ledger.go

@@ -13,9 +13,10 @@ type keySet map[u.Key]struct{}
 
 func newLedger(p peer.Peer, strategy strategyFunc) *ledger {
 	return &ledger{
-		wantList: keySet{},
-		Strategy: strategy,
-		Partner:  p,
+		wantList:   keySet{},
+		Strategy:   strategy,
+		Partner:    p,
+		sentToPeer: make(map[u.Key]time.Time),
 	}
 }
 
@@ -40,6 +41,10 @@ type ledger struct {
 	// wantList is a (bounded, small) set of keys that Partner desires.
 	wantList keySet
 
+	// sentToPeer is a set of keys to ensure we dont send duplicate blocks
+	// to a given peer
+	sentToPeer map[u.Key]time.Time
+
 	Strategy strategyFunc
 }
 

exchange/bitswap/strategy/strategy.go

@@ -10,6 +10,8 @@ import (
 	u "github.com/jbenet/go-ipfs/util"
 )
 
+const resendTimeoutPeriod = time.Minute
+
 var log = u.Logger("strategy")
 
 // TODO niceness should be on a per-peer basis. Use-case: Certain peers are
@@ -65,9 +67,25 @@ func (s *strategist) ShouldSendBlockToPeer(k u.Key, p peer.Peer) bool {
 	defer s.lock.RUnlock()
 
 	ledger := s.ledger(p)
+
+	// Dont resend blocks within a certain time period
+	t, ok := ledger.sentToPeer[k]
+	if ok && t.Add(resendTimeoutPeriod).After(time.Now()) {
+		log.Error("Prevented block resend!")
+		return false
+	}
+
 	return ledger.ShouldSend()
 }
 
+func (s *strategist) BlockSentToPeer(k u.Key, p peer.Peer) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	ledger := s.ledger(p)
+	ledger.sentToPeer[k] = time.Now()
+}
+
 func (s *strategist) Seed(int64) {
 	s.lock.Lock()
 	defer s.lock.Unlock()