bitswap/client: broadcast reduction and metrics

CHANGELOG.md

@@ -18,6 +18,8 @@ The following emojis are used to highlight certain changes:
 
 ### Changed
 
+- `bitswap/client`: Reduce bitswap broadcast volume by limiting broadcasts to peers that have previously responded as having wanted blocks and peers on local network.
+
 ### Removed
 
 - `bitswap/server` do not allow override of peer ledger with `WithPeerLedger` [#938](https://github.com/ipfs/boxo/pull/938)

bitswap/client/bitswap_with_sessions_test.go

@@ -137,7 +137,9 @@ func TestCustomProviderQueryManager(t *testing.T) {
 	defer cancel()
 
 	bs := bitswap.New(ctx, a.Adapter, pqm, a.Blockstore,
-		bitswap.WithClientOption(client.WithDefaultProviderQueryManager(false)))
+		bitswap.WithClientOption(client.WithDefaultProviderQueryManager(false)),
+		bitswap.WithClientOption(client.WithBroadcastReduction(false)),
+	)
 	a.Exchange.Close() // close old to be sure.
 	a.Exchange = bs
 	// Connect instances only after bitswap exists.
@@ -586,7 +588,9 @@ func TestDontHaveTimeoutConfig(t *testing.T) {
 	defer cancel()
 
 	bs := bitswap.New(ctx, a.Adapter, pqm, a.Blockstore,
-		bitswap.WithClientOption(client.WithDontHaveTimeoutConfig(cfg)))
+		bitswap.WithClientOption(client.WithDontHaveTimeoutConfig(cfg)),
+		bitswap.WithClientOption(client.WithBroadcastReduction(false)),
+	)
 	a.Exchange.Close()
 	a.Exchange = bs
 

bitswap/client/client.go

@@ -137,6 +137,36 @@ func WithDefaultProviderQueryManager(defaultProviderQueryManager bool) Option {
 	}
 }
 
+func WithBroadcastReduction(enable bool) Option {
+	return func(bs *Client) {
+		bs.bcastReduction = enable
+	}
+}
+
+func WithBroadcastReduceLocal(enable bool) Option {
+	return func(bs *Client) {
+		bs.bcastReduceLocal = enable
+	}
+}
+
+func WithBroadcastSendSkipped(n int) Option {
+	return func(bs *Client) {
+		bs.bcastSendSkipped = n
+	}
+}
+
+func WithBroadcastLimitPeers(limit int) Option {
+	return func(bs *Client) {
+		bs.bcastLimitPeers = limit
+	}
+}
+
+func WithBroadcastSendWithPending(enable bool) Option {
+	return func(bs *Client) {
+		bs.bcastSendWithPending = enable
+	}
+}
+
 type BlockReceivedNotifier interface {
 	// ReceivedBlocks notifies the decision engine that a peer is well-behaving
 	// and gave us useful data, potentially increasing its score and making us
@@ -166,17 +196,33 @@ func New(parent context.Context, network bsnet.BitSwapNetwork, providerFinder ro
 		counters:                    new(counters),
 		dupMetric:                   bmetrics.DupHist(ctx),
 		allMetric:                   bmetrics.AllHist(ctx),
+		havesReceivedGauge:          bmetrics.HavesReceivedGauge(ctx),
+		uniqueBlocksReceivedGauge:   bmetrics.UniqueBlocksReceivedGauge(ctx),
 		provSearchDelay:             defaults.ProvSearchDelay,
 		rebroadcastDelay:            delay.Fixed(defaults.RebroadcastDelay),
 		simulateDontHavesOnTimeout:  true,
 		defaultProviderQueryManager: true,
+		bcastReduction:              true,
 	}
 
 	// apply functional options before starting and running bitswap
 	for _, option := range options {
 		option(bs)
 	}
 
+	var bcastConfig *bspm.BroadcastConfig
+	if bs.bcastReduction {
+		bcastConfig = &bspm.BroadcastConfig{
+			LimitPeers:      bs.bcastLimitPeers,
+			SendSkipped:     bs.bcastSendSkipped,
+			SendWithPending: bs.bcastSendWithPending,
+			SkipGauge:       bmetrics.BroadcastSkipGauge(ctx),
+		}
+		if !bs.bcastReduceLocal {
+			bcastConfig.LocalAlways = network.GetPeerstore()
+		}
+	}
+
 	// onDontHaveTimeout is called when a want-block is sent to a peer that
 	// has an old version of Bitswap that doesn't support DONT_HAVE messages,
 	// or when no response is received within a timeout.
@@ -201,7 +247,7 @@ func New(parent context.Context, network bsnet.BitSwapNetwork, providerFinder ro
 
 	sim := bssim.New()
 	bpm := bsbpm.New()
-	pm := bspm.New(ctx, peerQueueFactory)
+	pm := bspm.New(ctx, peerQueueFactory, bcastConfig)
 
 	if bs.providerFinder != nil && bs.defaultProviderQueryManager {
 		// network can do dialing.
@@ -237,7 +283,7 @@ func New(parent context.Context, network bsnet.BitSwapNetwork, providerFinder ro
 		} else if providerFinder != nil {
 			sessionProvFinder = providerFinder
 		}
-		return bssession.New(sessctx, sessmgr, id, spm, sessionProvFinder, sim, pm, bpm, notif, provSearchDelay, rebroadcastDelay, self)
+		return bssession.New(sessctx, sessmgr, id, spm, sessionProvFinder, sim, pm, bpm, notif, provSearchDelay, rebroadcastDelay, self, bs.havesReceivedGauge)
 	}
 	sessionPeerManagerFactory := func(ctx context.Context, id uint64) bssession.SessionPeerManager {
 		return bsspm.New(id, network)
@@ -285,6 +331,9 @@ type Client struct {
 	dupMetric metrics.Histogram
 	allMetric metrics.Histogram
 
+	havesReceivedGauge        bspm.Gauge
+	uniqueBlocksReceivedGauge bspm.Gauge
+
 	// External statistics interface
 	tracer tracer.Tracer
 
@@ -311,6 +360,12 @@ type Client struct {
 	skipDuplicatedBlocksStats bool
 
 	perPeerSendDelay time.Duration
+
+	bcastReduction       bool
+	bcastReduceLocal     bool
+	bcastLimitPeers      int
+	bcastSendSkipped     int
+	bcastSendWithPending bool
 }
 
 type counters struct {
@@ -384,6 +439,10 @@ func (bs *Client) receiveBlocksFrom(ctx context.Context, from peer.ID, blks []bl
 	default:
 	}
 
+	if len(blks) != 0 || len(haves) != 0 {
+		bs.pm.MarkBroadcastTarget(from)
+	}
+
 	wanted, notWanted := bs.sim.SplitWantedUnwanted(blks)
 	if log.Level().Enabled(zapcore.DebugLevel) {
 		for _, b := range notWanted {
@@ -483,6 +542,8 @@ func (bs *Client) updateReceiveCounters(blocks []blocks.Block) {
 		if has {
 			c.dupBlocksRecvd++
 			c.dupDataRecvd += uint64(blkLen)
+		} else {
+			bs.uniqueBlocksReceivedGauge.Inc()
 		}
 	}
 }

bitswap/client/internal/messagequeue/messagequeue.go

@@ -107,6 +107,8 @@ type MessageQueue struct {
 	events chan<- messageEvent
 
 	perPeerDelay time.Duration
+
+	BcastInc func()
 }
 
 // recallWantlist keeps a list of pending wants and a list of sent wants
@@ -424,6 +426,13 @@ func (mq *MessageQueue) AddCancels(cancelKs []cid.Cid) {
 	}
 }
 
+func (mq *MessageQueue) HasMessage() bool {
+	mq.wllock.Lock()
+	defer mq.wllock.Unlock()
+
+	return mq.bcstWants.pending.Len() != 0 || mq.peerWants.pending.Len() != 0 || mq.cancels.Len() != 0
+}
+
 // ResponseReceived is called when a message is received from the network.
 // ks is the set of blocks, HAVEs and DONT_HAVEs in the message
 // Note that this is just used to calculate latency.
@@ -862,6 +871,9 @@ FINISH:
 		for _, e := range bcstEntries[:sentBcstEntries] {
 			if e.Cid.Defined() { // Check if want was canceled in the interim
 				mq.bcstWants.setSentAt(e.Cid, now)
+				if mq.BcastInc != nil {
+					mq.BcastInc()
+				}
 			}
 		}
 

bitswap/client/internal/peermanager/peermanager.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"sync"
 
+	"github.com/ipfs/boxo/bitswap/client/internal/messagequeue"
 	cid "github.com/ipfs/go-cid"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/ipfs/go-metrics-interface"
@@ -18,6 +19,7 @@ type PeerQueue interface {
 	AddWants([]cid.Cid, []cid.Cid)
 	AddCancels([]cid.Cid)
 	ResponseReceived(ks []cid.Cid)
+	HasMessage() bool
 	Startup()
 	Shutdown()
 }
@@ -44,20 +46,25 @@ type PeerManager struct {
 	psLk         sync.RWMutex
 	sessions     map[uint64]Session
 	peerSessions map[peer.ID]map[uint64]struct{}
+
+	bcastGauge Gauge
 }
 
 // New creates a new PeerManager, given a context and a peerQueueFactory.
-func New(ctx context.Context, createPeerQueue PeerQueueFactory) *PeerManager {
+func New(ctx context.Context, createPeerQueue PeerQueueFactory, bcastConfig *BroadcastConfig) *PeerManager {
 	wantGauge := metrics.NewCtx(ctx, "wantlist_total", "Number of items in wantlist.").Gauge()
 	wantBlockGauge := metrics.NewCtx(ctx, "want_blocks_total", "Number of want-blocks in wantlist.").Gauge()
+
 	return &PeerManager{
 		peerQueues:      make(map[peer.ID]PeerQueue),
-		pwm:             newPeerWantManager(wantGauge, wantBlockGauge),
+		pwm:             newPeerWantManager(wantGauge, wantBlockGauge, bcastConfig),
 		createPeerQueue: createPeerQueue,
 		ctx:             ctx,
 
 		sessions:     make(map[uint64]Session),
 		peerSessions: make(map[peer.ID]map[uint64]struct{}),
+
+		bcastGauge: metrics.NewCtx(ctx, "wanthaves_broadcast", "Number of want-haves broadcast.").Gauge(),
 	}
 }
 
@@ -189,6 +196,10 @@ func (pm *PeerManager) getOrCreate(p peer.ID) PeerQueue {
 	pq, ok := pm.peerQueues[p]
 	if !ok {
 		pq = pm.createPeerQueue(pm.ctx, p)
+		mq, ok := pq.(*messagequeue.MessageQueue)
+		if ok {
+			mq.BcastInc = pm.bcastGauge.Inc
+		}
 		pq.Startup()
 		pm.peerQueues[p] = pq
 	}
@@ -227,6 +238,10 @@ func (pm *PeerManager) UnregisterSession(ses uint64) {
 	delete(pm.sessions, ses)
 }
 
+func (pm *PeerManager) MarkBroadcastTarget(from peer.ID) {
+	pm.pwm.markBroadcastTarget(from)
+}
+
 // signalAvailability is called when a peer's connectivity changes.
 // It informs interested sessions.
 func (pm *PeerManager) signalAvailability(p peer.ID, isConnected bool) {

bitswap/client/internal/peermanager/peermanager_test.go

@@ -39,6 +39,10 @@ func (fp *mockPeerQueue) AddCancels(cs []cid.Cid) {
 	fp.msgs <- msg{fp.p, nil, nil, cs}
 }
 
+func (fp *mockPeerQueue) HasMessage() bool {
+	return true
+}
+
 func (fp *mockPeerQueue) ResponseReceived(ks []cid.Cid) {
 }
 
@@ -86,7 +90,7 @@ func TestAddingAndRemovingPeers(t *testing.T) {
 
 	tp := random.Peers(6)
 	peer1, peer2, peer3, peer4, peer5 := tp[0], tp[1], tp[2], tp[3], tp[4]
-	peerManager := New(ctx, peerQueueFactory)
+	peerManager := New(ctx, peerQueueFactory, bcastAlways)
 
 	peerManager.Connected(peer1)
 	peerManager.Connected(peer2)
@@ -129,7 +133,7 @@ func TestBroadcastOnConnect(t *testing.T) {
 	peerQueueFactory := makePeerQueueFactory(msgs)
 	tp := random.Peers(2)
 	peer1 := tp[0]
-	peerManager := New(ctx, peerQueueFactory)
+	peerManager := New(ctx, peerQueueFactory, bcastAlways)
 
 	cids := random.Cids(2)
 	peerManager.BroadcastWantHaves(ctx, cids)
@@ -150,7 +154,7 @@ func TestBroadcastWantHaves(t *testing.T) {
 	peerQueueFactory := makePeerQueueFactory(msgs)
 	tp := random.Peers(3)
 	peer1, peer2 := tp[0], tp[1]
-	peerManager := New(ctx, peerQueueFactory)
+	peerManager := New(ctx, peerQueueFactory, bcastAlways)
 
 	cids := random.Cids(3)
 
@@ -191,7 +195,7 @@ func TestSendWants(t *testing.T) {
 	peerQueueFactory := makePeerQueueFactory(msgs)
 	tp := random.Peers(2)
 	peer1 := tp[0]
-	peerManager := New(ctx, peerQueueFactory)
+	peerManager := New(ctx, peerQueueFactory, bcastAlways)
 	cids := random.Cids(4)
 
 	peerManager.Connected(peer1)
@@ -225,7 +229,7 @@ func TestSendCancels(t *testing.T) {
 	peerQueueFactory := makePeerQueueFactory(msgs)
 	tp := random.Peers(3)
 	peer1, peer2 := tp[0], tp[1]
-	peerManager := New(ctx, peerQueueFactory)
+	peerManager := New(ctx, peerQueueFactory, bcastAlways)
 	cids := random.Cids(4)
 
 	// Connect to peer1 and peer2
@@ -286,7 +290,7 @@ func TestSessionRegistration(t *testing.T) {
 
 	tp := random.Peers(3)
 	p1, p2 := tp[0], tp[1]
-	peerManager := New(ctx, peerQueueFactory)
+	peerManager := New(ctx, peerQueueFactory, bcastAlways)
 
 	id := uint64(1)
 	s := newSess(id)
@@ -332,6 +336,7 @@ func (*benchPeerQueue) Shutdown() {}
 func (*benchPeerQueue) AddBroadcastWantHaves(whs []cid.Cid)   {}
 func (*benchPeerQueue) AddWants(wbs []cid.Cid, whs []cid.Cid) {}
 func (*benchPeerQueue) AddCancels(cs []cid.Cid)               {}
+func (*benchPeerQueue) HasMessage() bool                      { return true }
 func (*benchPeerQueue) ResponseReceived(ks []cid.Cid)         {}
 
 // Simplistic benchmark to allow us to stress test
@@ -345,7 +350,7 @@ func BenchmarkPeerManager(b *testing.B) {
 	}
 
 	peers := random.Peers(500)
-	peerManager := New(ctx, peerQueueFactory)
+	peerManager := New(ctx, peerQueueFactory, bcastAlways)
 
 	// Create a bunch of connections
 	connected := 0