feat: use multiple `mpsc` channels instead of `broadcast` for async interface (#4190)
### Description Follow up to https://github.com/hyperlane-xyz/hyperlane-monorepo/pull/4180 Lowers txid broadcast channel capacity from 1M to 30k and refactors it to use a vec of `mpsc` under the hood. ### Context The heap profiler showed that each new chain takes up 100MB of memory because of the txid broadcast channel. In addition to this, the tokio broadcast channel is a ring buffer where new items simply overwrite old items, making it hard to safely lower the capacity. As such, to be able to safely lower the capacity of this channel, usage of `tokio`'s `broadcast` channel is replaced with a `Vec` of `mpsc`s, which offer an async `send` interface for backpressure. So instead of overwriting old items, senders will block on sends, which makes sense in our case as we absolutely don't want to miss items. The submitter is slower than the indexer based on empirical observation, so blocking on channel size doesn't bottleneck throughput. I noticed the max number of channel items we can reach is 29k, by running the RC relayer after an 8 day break, during which around 160k messages were submitted. Lowering the capacity to 30k should cause us to only encounter backpressure in very rare cases, and the footprint of this channel would lower from 100MB to 3MB. ### Drive-by changes The delivery indexer in the scraper is no longer passed a txid receiver, since message deliveries are not expected to occur in the same tx as message dispatches. ### Related issues - Fixes https://github.com/hyperlane-xyz/issues/issues/1293 ### Testing E2E, by adding invariants to make sure txid indexing works --------- Co-authored-by: Trevor Porter <tkporter4@gmail.com>pull/4219/head
Daniel Savu
4 months ago
committed by
GitHub
parent
98eb680adf
commit
b643fba90a
@ -0,0 +1,52 @@ |
||||
use std::sync::Arc; |
||||
|
||||
use derive_new::new; |
||||
use eyre::Result; |
||||
use hyperlane_core::H512; |
||||
use tokio::sync::{ |
||||
mpsc::{Receiver as MpscReceiver, Sender as MpscSender}, |
||||
Mutex, |
||||
}; |
||||
|
||||
#[derive(Debug, Clone, new)] |
||||
/// Wrapper around a vec of mpsc senders that broadcasts messages to all of them.
|
||||
/// This is a workaround to get an async interface for `send`, so senders are blocked if any of the receiving channels is full,
|
||||
/// rather than overwriting old messages (as the `broadcast` channel ring buffer implementation does).
|
||||
pub struct BroadcastMpscSender<T> { |
||||
capacity: usize, |
||||
/// To make this safe to `Clone`, the sending end has to be in an arc-mutex.
|
||||
/// Otherwise it would be possible to call `get_receiver` and create new receiver-sender pairs, whose sender is later dropped
|
||||
/// because the other `BroadcastMpscSender`s have no reference to it. The receiver would then point to a closed
|
||||
/// channel. So all instances of `BroadcastMpscSender` have to point to the entire set of senders.
|
||||
#[new(default)] |
||||
sender: Arc<Mutex<Vec<MpscSender<T>>>>, |
||||
} |
||||
|
||||
impl BroadcastMpscSender<H512> { |
||||
/// Send a message to all the receiving channels.
|
||||
// This will block if at least one of the receiving channels is full
|
||||
pub async fn send(&self, txid: H512) -> Result<()> { |
||||
let senders = self.sender.lock().await; |
||||
for sender in &*senders { |
||||
sender.send(txid).await? |
||||
} |
||||
Ok(()) |
||||
} |
||||
|
||||
/// Get a receiver channel that will receive messages broadcasted by all the senders
|
||||
pub async fn get_receiver(&self) -> MpscReceiver<H512> { |
||||
let (sender, receiver) = tokio::sync::mpsc::channel(self.capacity); |
||||
|
||||
self.sender.lock().await.push(sender); |
||||
receiver |
||||
} |
||||
|
||||
/// Utility function map an option of `BroadcastMpscSender` to an option of `MpscReceiver`
|
||||
pub async fn map_get_receiver(maybe_self: Option<&Self>) -> Option<MpscReceiver<H512>> { |
||||
if let Some(s) = maybe_self { |
||||
Some(s.get_receiver().await) |
||||
} else { |
||||
None |
||||
} |
||||
} |
||||
} |
||||
Loading…
Reference in new issue