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,48 @@
+package pubsub
+
+import (
+	"sync"
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/peer"
+)
+
+const (
+	MinBackoffDelay   = 100 * time.Millisecond
+	MaxBackoffDelay   = 10 * time.Second
+	BackoffMultiplier = 2
+)
+
+type backoff struct {
+	mu sync.Mutex
+	info map[peer.ID]time.Duration
+}
+
+func newBackoff() *backoff{
+	return &backoff{
+		mu: sync.Mutex{},
+		info: make(map[peer.ID]time.Duration),
+	}
+}
+
+func (b *backoff) updateAndGet(id peer.ID) time.Duration{
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	h, ok := b.info[id]
+	if !ok {
+		// first request goes immediately.
+		h = time.Duration(0)
+	} else if h < MinBackoffDelay {
+		h = MinBackoffDelay
+	} else if h < MaxBackoffDelay {
+		h = time.Duration(BackoffMultiplier * h)
+		if h > MaxBackoffDelay || h < 0 {
+			h = MaxBackoffDelay
+		}
+	}
+
+	b.info[id] = h
+
+	return h
+}

backoff_test.go

@@ -0,0 +1,48 @@
+package pubsub
+
+import (
+	"math"
+	"testing"
+	"time"
+
+	"github.com/libp2p/go-libp2p-core/peer"
+)
+
+func TestBackoff(t *testing.T){
+	id1 := peer.ID("peer-1")
+	id2 := peer.ID("peer-2")
+	b := newBackoff()
+
+	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)
+	}
+
+	if len(b.info) != 2 {
+		t.Fatalf("info map size mismatch, expected: %d, got: %d", 2, 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(),
 		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
 		}