From 77ed64bb8800f31d534a113474279b2dd038c37a Mon Sep 17 00:00:00 2001 From: Johnathan Corgan Date: Wed, 29 Jul 2026 01:45:12 +0000 Subject: [PATCH] Compute every peer's outgoing bloom filter in one sweep Each peer must be sent the union of all other peers' inbound filters, excluding its own contribution. Building that per recipient rebuilt the whole peer-filter map and re-ORed it once for every peer, so a tick that announced to R peers did R kilobyte-scale map builds and R by T merges. At 240 peers this was 20.6 ms per tick, roughly half the tick body, and it was steady work rather than a tail: the per-interval maximum had a median of 34.5 ms. Replace it with a prefix and suffix union sweep that produces every target's filter in one pass, T merges instead of R by T. The packet path gets the same treatment, since marking changed peers had the identical shape once per inbound announce. The result is exactly equal, not approximately. Merging is a bytewise OR, so regrouping the unions cannot change the outcome, and a filter whose size does not match is rejected before any byte is touched, at every merge site in both the old and new arrangement. Every accumulator here is default-sized, so an odd-sized peer filter is skipped in the new code exactly where it was skipped in the old. Cadence, the debounce, the sequence rule and the fill-ratio cap are untouched. A sequence number is still drawn after the debounce re-check and before encoding, so a suppressed peer still consumes none and an encode failure still burns one. Known trade-off, measured rather than assumed: the sweep does its full O(T) work regardless of how many peers are ready, so a tick that announces to only one or two peers now costs about twice what it did. Break-even is around three ready peers, and the saving above that grows without bound. The marking path is a pure win, since it always targets every peer. --- src/bloom/state.rs | 72 +++++++++++++++++++++++++++++++++++++++++----- src/bloom/tests.rs | 67 ++++++++++++++++++++++++++++++++++++++++++ src/node/bloom.rs | 40 +++++++++++++++----------- 3 files changed, 156 insertions(+), 23 deletions(-) diff --git a/src/bloom/state.rs b/src/bloom/state.rs index 5a0f5c8..e47422d 100644 --- a/src/bloom/state.rs +++ b/src/bloom/state.rs @@ -172,21 +172,79 @@ impl BloomState { peer_addrs: &[NodeAddr], peer_filters: &HashMap, ) { - for peer_addr in peer_addrs { - if peer_addr == exclude_from { - continue; - } - let new_filter = self.compute_outgoing_filter(peer_addr, peer_filters); - let changed = match self.last_sent_filters.get(peer_addr) { + let targets: Vec = peer_addrs + .iter() + .filter(|addr| *addr != exclude_from) + .copied() + .collect(); + + for (peer_addr, new_filter) in self.compute_outgoing_filters(&targets, peer_filters) { + let changed = match self.last_sent_filters.get(&peer_addr) { Some(last) => *last != new_filter, None => true, // never sent → must send }; if changed { - self.pending_updates.insert(*peer_addr); + self.pending_updates.insert(peer_addr); } } } + /// Compute the outgoing filter for many peers in one pass. + /// + /// Equivalent to calling [`compute_outgoing_filter`](Self::compute_outgoing_filter) + /// once per target, but linear in the number of contributing peer + /// filters instead of quadratic. The per-peer call rebuilds the whole + /// union from scratch, so computing it for every peer costs + /// O(targets × filters) 1 KB merges; announce fan-out on a + /// large node does exactly that, once per tick and again on every + /// inbound announce. + /// + /// The split-horizon exclusion is the only thing that differs between + /// targets, so the union of "everything except peer i" is assembled + /// from a running prefix union and a precomputed suffix union. Merging + /// is a bytewise OR, which is commutative and associative, so the + /// result is bit-identical to the per-peer computation. + pub fn compute_outgoing_filters( + &self, + targets: &[NodeAddr], + peer_filters: &HashMap, + ) -> HashMap { + let base = self.base_filter(); + let keys: Vec = peer_filters.keys().copied().collect(); + let n = keys.len(); + + // suffix[i] = union of peer_filters[keys[i..]]; suffix[n] is empty. + let mut suffix = vec![BloomFilter::new(); n + 1]; + for i in (0..n).rev() { + let mut acc = suffix[i + 1].clone(); + // Size mismatches are skipped, exactly as in the per-peer path. + let _ = acc.merge(&peer_filters[&keys[i]]); + suffix[i] = acc; + } + + // Filter for a target that contributes nothing: everything merged. + let mut all = base.clone(); + let _ = all.merge(&suffix[0]); + + let mut per_key: HashMap = HashMap::with_capacity(n); + let mut prefix = BloomFilter::new(); + for i in 0..n { + let mut outgoing = base.clone(); + let _ = outgoing.merge(&prefix); + let _ = outgoing.merge(&suffix[i + 1]); + per_key.insert(keys[i], outgoing); + let _ = prefix.merge(&peer_filters[&keys[i]]); + } + + targets + .iter() + .map(|target| { + let filter = per_key.get(target).cloned().unwrap_or_else(|| all.clone()); + (*target, filter) + }) + .collect() + } + /// Compute the outgoing filter for a specific peer. /// /// The filter includes: diff --git a/src/bloom/tests.rs b/src/bloom/tests.rs index 37a7e70..4a8ba83 100644 --- a/src/bloom/tests.rs +++ b/src/bloom/tests.rs @@ -684,3 +684,70 @@ fn test_bloom_state_mark_changed_peers_excludes_source() { assert!(!state.needs_update(&peer1)); } + +#[test] +fn test_compute_outgoing_filters_matches_per_peer() { + let node = make_node_addr(0); + let mut state = BloomState::new(node); + state.add_leaf_dependent(make_node_addr(200)); + state.add_leaf_dependent(make_node_addr(201)); + + // Six contributing peers with overlapping content, plus one whose + // filter is a different size and must be skipped by both paths. + let mut peer_filters = HashMap::new(); + for i in 1u8..=6 { + let mut filter = BloomFilter::new(); + for j in 0..5u8 { + filter.insert(&make_node_addr(i.wrapping_mul(7).wrapping_add(j))); + } + peer_filters.insert(make_node_addr(i), filter); + } + let odd_peer = make_node_addr(7); + let mut odd = BloomFilter::with_params(4096, DEFAULT_HASH_COUNT).unwrap(); + odd.insert(&make_node_addr(99)); + peer_filters.insert(odd_peer, odd); + + // Targets: every contributing peer, the odd-sized one, and two peers + // that contribute nothing (non-tree peers get announces too). + let mut targets: Vec = (1u8..=7).map(make_node_addr).collect(); + targets.push(make_node_addr(120)); + targets.push(make_node_addr(121)); + + let batch = state.compute_outgoing_filters(&targets, &peer_filters); + + assert_eq!(batch.len(), targets.len()); + for target in &targets { + let expected = state.compute_outgoing_filter(target, &peer_filters); + assert_eq!( + batch.get(target), + Some(&expected), + "batch filter for {:?} differs from per-peer computation", + target + ); + } + + // Split horizon is real, not vacuous: a contributing peer's own + // entries must be absent from its own outgoing filter, and present + // in another peer's. + let peer3 = make_node_addr(3); + let own_entry = make_node_addr(3u8.wrapping_mul(7)); + assert!(!batch[&peer3].contains(&own_entry)); + assert!(batch[&make_node_addr(1)].contains(&own_entry)); +} + +#[test] +fn test_compute_outgoing_filters_empty_inputs() { + let node = make_node_addr(0); + let state = BloomState::new(node); + let peer_filters = HashMap::new(); + + assert!( + state + .compute_outgoing_filters(&[], &peer_filters) + .is_empty() + ); + + let target = make_node_addr(1); + let batch = state.compute_outgoing_filters(&[target], &peer_filters); + assert_eq!(batch[&target], state.base_filter()); +} diff --git a/src/node/bloom.rs b/src/node/bloom.rs index f16c243..cde819f 100644 --- a/src/node/bloom.rs +++ b/src/node/bloom.rs @@ -29,27 +29,19 @@ impl Node { filters } - /// Build a FilterAnnounce for a specific peer. - /// - /// The outgoing filter excludes the destination peer's own filter - /// to prevent routing loops (don't tell a peer about destinations - /// reachable only through them). - fn build_filter_announce(&mut self, exclude_peer: &NodeAddr) -> FilterAnnounce { - let peer_filters = self.peer_inbound_filters(); - let filter = self - .bloom_state - .compute_outgoing_filter(exclude_peer, &peer_filters); - let sequence = self.bloom_state.next_sequence(); - FilterAnnounce::new(filter, sequence) - } - /// Send a FilterAnnounce to a specific peer, respecting debounce. /// + /// `filter` is the outgoing filter for this peer, already computed + /// with the destination peer's own contribution excluded to prevent + /// routing loops (don't tell a peer about destinations reachable + /// only through them). + /// /// If the peer is rate-limited, the update stays pending for /// delivery on the next tick cycle. pub(super) async fn send_filter_announce_to_peer( &mut self, peer_addr: &NodeAddr, + filter: BloomFilter, ) -> Result<(), NodeError> { let now_ms = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -64,7 +56,7 @@ impl Node { } // Build and encode - let announce = self.build_filter_announce(peer_addr); + let announce = FilterAnnounce::new(filter, self.bloom_state.next_sequence()); let sent_filter = announce.filter.clone(); let encoded = announce.encode().map_err(|e| NodeError::SendFailed { node_addr: *peer_addr, @@ -142,8 +134,24 @@ impl Node { .copied() .collect(); + if ready.is_empty() { + return; + } + + // One snapshot and one union pass for the whole ready set. The + // send path never mutates peer inbound filters or the tree state, + // and the rx loop holds `&mut self` across the awaits, so the + // snapshot cannot go stale mid-loop. + let peer_filters = self.peer_inbound_filters(); + let mut outgoing = self + .bloom_state + .compute_outgoing_filters(&ready, &peer_filters); + for peer_addr in ready { - if let Err(e) = self.send_filter_announce_to_peer(&peer_addr).await { + let Some(filter) = outgoing.remove(&peer_addr) else { + continue; + }; + if let Err(e) = self.send_filter_announce_to_peer(&peer_addr, filter).await { debug!( peer = %self.peer_display_name(&peer_addr), error = %e,