libnetwork/networkdb: prioritize local broadcasts

corhere · corhere · commit 6ec6e0991a80 · 2025-06-13T17:08:18.000-04:00
A network node is responsible for both broadcasting table events for
entries it owns and for rebroadcasting table events from other nodes it
has received. Table events to be broadcast are added to a single queue
per network, including events for rebroadcasting. As the memberlist
TransmitLimitedQueue is (to a first approximation) LIFO, a flood of
events from other nodes could delay the broadcasting of
locally-generated events indefinitely. Prioritize broadcasting local
events by splitting up the queues and only pulling from the rebroadcast
queue if there is free space in the gossip packet after draining the
local-broadcast queue.

Signed-off-by: Cory Snider &lt;csnider@mirantis.com&gt;
diff --git a/libnetwork/networkdb/broadcast.go b/libnetwork/networkdb/broadcast.go
@@ -159,3 +159,18 @@ func (nDB *NetworkDB) sendTableEvent(event TableEvent_Type, nid string, tname st
 	})
 	return nil
 }
+
+func getBroadcasts(overhead, limit int, queues ...*memberlist.TransmitLimitedQueue) [][]byte {
+	var msgs [][]byte
+	for _, q := range queues {
+		b := q.GetBroadcasts(overhead, limit)
+		for _, m := range b {
+			limit -= overhead + len(m)
+		}
+		msgs = append(msgs, b...)
+		if limit <= 0 {
+			break
+		}
+	}
+	return msgs
+}
diff --git a/libnetwork/networkdb/cluster.go b/libnetwork/networkdb/cluster.go
@@ -469,14 +469,15 @@ func (nDB *NetworkDB) gossip() {
 			continue
 		}
 
-		msgs := network.tableBroadcasts.GetBroadcasts(compoundOverhead, bytesAvail)
+		msgs := getBroadcasts(compoundOverhead, bytesAvail, network.tableBroadcasts, network.tableRebroadcasts)
 		// Collect stats and print the queue info, note this code is here also to have a view of the queues empty
 		network.qMessagesSent.Add(int64(len(msgs)))
 		if printStats {
 			msent := network.qMessagesSent.Swap(0)
-			log.G(context.TODO()).Infof("NetworkDB stats %v(%v) - netID:%s leaving:%t netPeers:%d entries:%d Queue qLen:%d netMsg/s:%d",
+			log.G(context.TODO()).Infof("NetworkDB stats %v(%v) - netID:%s leaving:%t netPeers:%d entries:%d Queue qLen:%d+%d netMsg/s:%d",
 				nDB.config.Hostname, nDB.config.NodeID,
-				nid, network.leaving, network.tableBroadcasts.NumNodes(), network.entriesNumber.Load(), network.tableBroadcasts.NumQueued(),
+				nid, network.leaving, network.tableBroadcasts.NumNodes(), network.entriesNumber.Load(),
+				network.tableBroadcasts.NumQueued(), network.tableRebroadcasts.NumQueued(),
 				msent/int64((nDB.config.StatsPrintPeriod/time.Second)))
 		}
 
diff --git a/libnetwork/networkdb/delegate.go b/libnetwork/networkdb/delegate.go
@@ -292,11 +292,11 @@ func (nDB *NetworkDB) handleTableMessage(buf []byte, isBulkSync bool) {
 		}
 
 		// if the queue is over the threshold, avoid distributing information coming from TCP sync
-		if isBulkSync && n.tableBroadcasts.NumQueued() > maxQueueLenBroadcastOnSync {
+		if isBulkSync && n.tableRebroadcasts.NumQueued() > maxQueueLenBroadcastOnSync {
 			return
 		}
 
-		n.tableBroadcasts.QueueBroadcast(&tableEventMessage{
+		n.tableRebroadcasts.QueueBroadcast(&tableEventMessage{
 			msg:   buf,
 			id:    tEvent.NetworkID,
 			tname: tEvent.TableName,
@@ -419,14 +419,7 @@ func (d *delegate) NotifyMsg(buf []byte) {
 }
 
 func (d *delegate) GetBroadcasts(overhead, limit int) [][]byte {
-	msgs := d.nDB.networkBroadcasts.GetBroadcasts(overhead, limit)
-	for _, m := range msgs {
-		limit -= overhead + len(m)
-	}
-	if limit > 0 {
-		msgs = append(msgs, d.nDB.nodeBroadcasts.GetBroadcasts(overhead, limit)...)
-	}
-	return msgs
+	return getBroadcasts(overhead, limit, d.nDB.networkBroadcasts, d.nDB.nodeBroadcasts)
 }
 
 func (d *delegate) LocalState(join bool) []byte {
diff --git a/libnetwork/networkdb/networkdb.go b/libnetwork/networkdb/networkdb.go
@@ -149,10 +149,21 @@ type thisNodeNetwork struct {
 	// Gets set to true after the first bulk sync happens
 	inSync bool
 
-	// The broadcast queue for table event gossip. This is only
-	// initialized for this node's network attachment entries.
+	// The broadcast queue for this network's table event gossip
+	// for entries owned by this node.
 	tableBroadcasts *memberlist.TransmitLimitedQueue
 
+	// The broadcast queue for this network's table event gossip
+	// relayed from other nodes.
+	//
+	// Messages in this queue are broadcasted when there is space available
+	// in the gossip packet after filling it with tableBroadcast messages.
+	// Relayed messages are broadcasted at a lower priority than messages
+	// originating from this node to ensure that local messages are always
+	// broadcasted in a timely manner, irrespective of how many messages
+	// from other nodes are queued for rebroadcasting.
+	tableRebroadcasts *memberlist.TransmitLimitedQueue
+
 	// Number of gossip messages sent related to this network during the last stats collection period
 	qMessagesSent atomic.Int64
 
@@ -626,17 +637,22 @@ func (nDB *NetworkDB) JoinNetwork(nid string) error {
 		n.network = network{ltime: ltime}
 		n.inSync = false
 	} else {
+		numNodes := func() int {
+			// TODO fcrisciani this can be optimized maybe avoiding the lock?
+			// this call is done each GetBroadcasts call to evaluate the number of
+			// replicas for the message
+			nDB.RLock()
+			defer nDB.RUnlock()
+			return len(nDB.networkNodes[nid])
+		}
 		n = &thisNodeNetwork{
 			network: network{ltime: ltime},
 			tableBroadcasts: &memberlist.TransmitLimitedQueue{
-				NumNodes: func() int {
-					// TODO fcrisciani this can be optimized maybe avoiding the lock?
-					// this call is done each GetBroadcasts call to evaluate the number of
-					// replicas for the message
-					nDB.RLock()
-					defer nDB.RUnlock()
-					return len(nDB.networkNodes[nid])
-				},
+				NumNodes:       numNodes,
+				RetransmitMult: 4,
+			},
+			tableRebroadcasts: &memberlist.TransmitLimitedQueue{
+				NumNodes:       numNodes,
 				RetransmitMult: 4,
 			},
 		}
