Adds exponential backoff to re-spawing new streams for supposedly dead peers

.gitignore

@@ -1,3 +1,5 @@
 cover.out
 prof.out
 go-floodsub.test
+
+.idea/

backoff.go

@@ -0,0 +1,71 @@
+package pubsub
+
+import (
+	"sync"
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/peer"
+)
+
+const (
+	MinBackoffDelay   = 100 * time.Millisecond
+	MaxBackoffDelay   = 10 * time.Second
+	TimeToLive        = 10 * time.Minute
+	BackoffMultiplier = 2
+)
+
+type backoffHistory struct {
+	duration  time.Duration
+	lastTried time.Time
+}
+
+type backoff struct {
+	mu   sync.Mutex
+	info map[peer.ID]*backoffHistory
+	ct   int // size threshold that kicks off the cleaner
+}
+
+func newBackoff(sizeThreshold int) *backoff {
+	return &backoff{
+		mu:   sync.Mutex{},
+		ct:   sizeThreshold,
+		info: make(map[peer.ID]*backoffHistory),
+	}
+}
+
+func (b *backoff) updateAndGet(id peer.ID) time.Duration {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	h, ok := b.info[id]
+	if !ok || time.Since(h.lastTried) > TimeToLive {
+		// first request goes immediately.
+		h = &backoffHistory{
+			duration: time.Duration(0),
+		}
+	} else if h.duration < MinBackoffDelay {
+		h.duration = MinBackoffDelay
+	} else if h.duration < MaxBackoffDelay {
+		h.duration = time.Duration(BackoffMultiplier * h.duration)
+		if h.duration > MaxBackoffDelay || h.duration < 0 {
+			h.duration = MaxBackoffDelay
+		}
+	}
+
+	h.lastTried = time.Now()
+	b.info[id] = h
+
+	if len(b.info) > b.ct {
+		b.cleanup()
+	}
+
+	return h.duration
+}
+
+func (b *backoff) cleanup() {
+	for id, h := range b.info {
+		if time.Since(h.lastTried) > TimeToLive {
+			delete(b.info, id)
+		}
+	}
+}

backoff_test.go

@@ -0,0 +1,82 @@
+package pubsub
+
+import (
+	"fmt"
+	"math"
+	"testing"
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/peer"
+)
+
+func TestBackoff_Update(t *testing.T){
+	id1 := peer.ID("peer-1")
+	id2 := peer.ID("peer-2")
+	b := newBackoff(10)
+
+	if len(b.info) > 0 {
+		t.Fatal("non-empty info map for backoff")
+	}
+
+	if d := b.updateAndGet(id1); d != time.Duration(0) {
+		t.Fatalf("invalid initialization: %v", d)
+	}
+	if d := b.updateAndGet(id2); d != time.Duration(0) {
+		t.Fatalf("invalid initialization: %v", d)
+	}
+
+	for i := 0; i < 10; i++{
+		got := b.updateAndGet(id1)
+
+		expected := time.Duration(math.Pow(BackoffMultiplier, float64(i)) * float64(MinBackoffDelay))
+		if expected > MaxBackoffDelay {
+			expected = MaxBackoffDelay
+		}
+
+		if expected != got {
+			t.Fatalf("invalid backoff result, expected: %v, got: %v", expected, got)
+		}
+	}
+
+	got := b.updateAndGet(id2)
+	if got != MinBackoffDelay {
+		t.Fatalf("invalid backoff result, expected: %v, got: %v", MinBackoffDelay, got)
+	}
+
+	// sets last tried of id2 to long ago that it resets back upon next try.
+	b.info[id2].lastTried = time.Now().Add(-TimeToLive)
+	got = b.updateAndGet(id2)
+	if got != time.Duration(0) {
+		t.Fatalf("invalid ttl expiration, expected: %v, got: %v", time.Duration(0), got)
+	}
+
+	if len(b.info) != 2 {
+		t.Fatalf("info map size mismatch, expected: %d, got: %d", 2, len(b.info))
+	}
+}
+
+func TestBackoff_Clean(t *testing.T){
+	size := 10
+	b := newBackoff(size)
+
+	for i := 0; i < size; i++{
+		id := peer.ID(fmt.Sprintf("peer-%d", i))
+		b.updateAndGet(id)
+		b.info[id].lastTried = time.Now().Add(-TimeToLive) // enforces expiry
+	}
+
+	if len(b.info) != size {
+		t.Fatalf("info map size mismatch, expected: %d, got: %d", size, len(b.info))
+	}
+
+	// next update should trigger cleanup
+	got := b.updateAndGet(peer.ID("some-new-peer"))
+	if got != time.Duration(0) {
+		t.Fatalf("invalid backoff result, expected: %v, got: %v", time.Duration(0), got)
+	}
+
+	// except "some-new-peer" every other records must be cleaned up
+	if len(b.info) != 1 {
+		t.Fatalf("info map size mismatch, expected: %d, got: %d", 1, len(b.info))
+	}
+}

comm.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"context"
 	"io"
+	"time"
 
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
@@ -121,6 +122,16 @@ func (p *PubSub) handleNewPeer(ctx context.Context, pid peer.ID, outgoing <-chan
 	}
 }
 
+func (p *PubSub) handleNewPeerWithBackoff(ctx context.Context, pid peer.ID, outgoing <-chan *RPC) {
+	delay := p.deadPeerBackoff.updateAndGet(pid)
+	select {
+	case <-time.After(delay):
+		p.handleNewPeer(ctx, pid, outgoing)
+	case <-ctx.Done():
+		return
+	}
+}
+
 func (p *PubSub) handlePeerEOF(ctx context.Context, s network.Stream) {
 	pid := s.Conn().RemotePeer()
 	r := protoio.NewDelimitedReader(s, p.maxMessageSize)

pubsub.go

@@ -110,6 +110,8 @@ type PubSub struct {
 	peerDeadPrioLk sync.RWMutex
 	peerDeadMx     sync.Mutex
 	peerDeadPend   map[peer.ID]struct{}
+	// backoff for retrying new connections to dead peers
+	deadPeerBackoff *backoff
 
 	// The set of topics we are subscribed to
 	mySubs map[string]map[*Subscription]struct{}
@@ -252,6 +254,7 @@ func NewPubSub(ctx context.Context, h host.Host, rt PubSubRouter, opts ...Option
 		newPeerError:          make(chan peer.ID),
 		peerDead:              make(chan struct{}, 1),
 		peerDeadPend:          make(map[peer.ID]struct{}),
+		deadPeerBackoff:       newBackoff(1000),
 		cancelCh:              make(chan *Subscription),
 		getPeers:              make(chan *listPeerReq),
 		addSub:                make(chan *addSubReq),
@@ -686,7 +689,7 @@ func (p *PubSub) handleDeadPeers() {
 			log.Debugf("peer declared dead but still connected; respawning writer: %s", pid)
 			messages := make(chan *RPC, p.peerOutboundQueueSize)
 			messages <- p.getHelloPacket()
-			go p.handleNewPeer(p.ctx, pid, messages)
+			go p.handleNewPeerWithBackoff(p.ctx, pid, messages)
 			p.peers[pid] = messages
 			continue
 		}