feat: gossip improvements - #5520
Conversation
Discard buffered gossip on shutdown instead of flushing, restore quit checks in broadcastNow, document the async BroadcastPeers contract, use fixed 100ms jitter, and clean up tests and metrics. Co-authored-by: Cursor <cursoragent@cursor.com>
| type pendingGossip struct { | ||
| addressee swarm.Address | ||
| peers map[string]swarm.Address // peer bytestring -> address (set semantics) | ||
| deadline time.Time |
There was a problem hiding this comment.
i wonder whether there's a benefit of keeping a deadline per peer. usually, write coalescing is simple enough:
- have an interval fire at a constant rate
- if new records arrived by interval fire
- when new entries arrived, optionally, postpone the sending the sending until the next interval firing (and so also you could extend up to a set upper bound, so that entries don't keep collecting forever but also guarantee that information goes out still relatively quickly)
- send all the pending sends
also: usually, when a peer arrives - we gossip that peer to all peers (full nodes) and gossip to that peer all of our connected peers.
this in turn means that sending is almost always involving all connected peers. which in turn also means that the timestamps on the individual pendingGossip entries would be almost identical (making the field even more so redundant)
There was a problem hiding this comment.
Nice work.
One idea: instead of deciding what to do based on how many peers are passed in, it might be cleaner to add a separate method like GossipPeer(addressee, peer) on discovery.Driver just for the buffered case. Make BroadcastPeers plain => send now and return an error method, and use GossipPeer as async (fire and forget) in kademlia.go:1082-1090 and flush in startGossipCoalescer worker. This allows you to drop coalesceThreshold. Also, the "buffer is full" logic, you can also move to the background worker with a wakeup channel so buffering never blocks or sends directly.
|
|
||
| key := addressee.ByteString() | ||
| peerSet, ok := b.pending[key] | ||
| if !ok { |
There was a problem hiding this comment.
since maps leak memory by design, it would be better to:
- check whether the b.pending key exists
- merge its contents with
peersif it does before writing the key to the map - check the max batch size and if the entry really exceeds max batch - return early without writing to the map
- finally if we are within the bounds of the max batch - write to map
| if err != nil { | ||
| s.logger.Debug("coalesced gossip flush failed", "addressee", addressee, "reason", reason, "batch_size", len(peers), "error", err) | ||
| } | ||
| cancel() |
There was a problem hiding this comment.
ideally this should be a defer call just after it gets created.
| select { | ||
| case <-ticker.C: | ||
| for _, batch := range s.gossipBuf.takeAll() { | ||
| s.flushGossipBatch(batch.addressee, batch.peers, coalesceFlushReasonTimer) |
There was a problem hiding this comment.
nit - this is a blocking call that makes slower peers to block other peers from getting the information. i would tend to turn this into go s.flushGossipBatch. iirc the latest go compilers make sure the values get copied correctly such that when the iterator changes batch values it doesn't change the underlying value for the goroutines already dispatched with that same variable name. but maybe also putting this into a closure won't hurt too much.
| ) | ||
|
|
||
| const ( | ||
| defaultGossipCoalesceInterval = time.Second |
There was a problem hiding this comment.
nit: i think this can be higher (like 5 sec)? the same for the coalesce threshold - we want to have bigger messages and less often. the timer fires every 5 seconds anyway.
| peers: slices.Collect(maps.Values(peerSet)), | ||
| }) | ||
| } | ||
| b.pending = make(map[string]map[string]swarm.Address) |
Nice idea, but looks like more changes than we need (?) |
not sure... also having a buffered vs unbuffered kinda defeats the purpose of batching the writes together. not sure if i see the case of non-buffered broadcast as needed. the thresholds should take care of that already - when the group is big enough - it is urgent enough (and that would always be the case for a peer that connects and gets a bunch of peers via gossip). so we can use the implicit behavior in this case instead of expanding the interfaces. my 2 cents |
| s.metrics.BroadcastPeersPeers.Add(float64(len(peers))) | ||
|
|
||
| // Already-batched messages go out immediately; single-peer gossips are coalesced. | ||
| if len(peers) >= coalesceThreshold { |
There was a problem hiding this comment.
if the addressee already has some peers which are queued for sending through the buffer - they are silently skipped here. not a big issue, can be handled later too. flagging this nevertheless.
also, you might want to gossip, but the rate limiter won't allow you to send the whole batch together because you can't get enough tokens from the bucket. this puts things as a best effort. i'm not sure we should handle all those edge cases right away but they are definitely worth documenting at least inline and perhaps as a follow up issue.
3 participants
Checklist
Description
Adds write coalescing for hive outbound gossip. Single-peer BroadcastPeers calls are buffered per addressee and flushed as one batched message after ~1s (configurable via GossipCoalesceInterval), or immediately when the buffer reaches maxBatchSize (30). Calls with 2+ peers are sent without coalescing
Open API Spec Version Changes (if applicable)
Motivation and Context (Optional)
Related Issue (Optional)
#5490
Screenshots (if appropriate):
AI Disclosure