From 8cf2a29a6c57359d522c10ac097728bda15608fa Mon Sep 17 00:00:00 2001 From: jmwample Date: Mon, 9 Mar 2026 11:07:20 -0600 Subject: [PATCH] mpmc ingest workers using flume --- Cargo.lock | 18 ++++- Cargo.toml | 1 + nym-node/Cargo.toml | 1 + nym-node/nym-node-metrics/src/mixnet.rs | 1 - nym-node/src/config/mod.rs | 2 +- .../src/config/old_configs/old_config_v12.rs | 3 +- nym-node/src/node/mixnet/listener.rs | 10 ++- nym-node/src/node/mixnet/packet_ingest.rs | 77 +++++++++++++++---- 8 files changed, 90 insertions(+), 23 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 937a30240a..a0296a26b7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2772,6 +2772,9 @@ name = "fastrand" version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" +dependencies = [ + "getrandom 0.2.16", +] [[package]] name = "ff" @@ -2844,6 +2847,18 @@ dependencies = [ "spin", ] +[[package]] +name = "flume" +version = "0.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5e139bc46ca777eb5efaf62df0ab8cc5fd400866427e56c68b22e414e53bd3be" +dependencies = [ + "fastrand", + "futures-core", + "futures-sink", + "spin", +] + [[package]] name = "fnv" version = "1.0.7" @@ -7149,6 +7164,7 @@ dependencies = [ "csv", "cupid", "dashmap", + "flume 0.12.0", "futures", "hex", "hkdf", @@ -11069,7 +11085,7 @@ checksum = "c2d12fe70b2c1b4401038055f90f151b78208de1f9f89a7dbfd41587a10c3eea" dependencies = [ "atoi", "chrono", - "flume", + "flume 0.11.1", "futures-channel", "futures-core", "futures-executor", diff --git a/Cargo.toml b/Cargo.toml index 4f7ebfae70..e68309908f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -269,6 +269,7 @@ etherparse = "0.13.0" eyre = "0.6.9" fastrand = "2.1.1" flate2 = "1.1.1" +flume = "0.12.0" futures = "0.3.31" futures-util = "0.3" generic-array = "0.14.7" diff --git a/nym-node/Cargo.toml b/nym-node/Cargo.toml index b17f5272f5..bfb8d8b4fa 100644 --- a/nym-node/Cargo.toml +++ b/nym-node/Cargo.toml @@ -30,6 +30,7 @@ console-subscriber = { workspace = true, optional = true } csv = { workspace = true } clap = { workspace = true, features = ["cargo", "env"] } dashmap = { workspace = true } +flume= { workspace = true } futures = { workspace = true } hex = { workspace = true } humantime-serde = { workspace = true } diff --git a/nym-node/nym-node-metrics/src/mixnet.rs b/nym-node/nym-node-metrics/src/mixnet.rs index fd48831ffe..741397b6d7 100644 --- a/nym-node/nym-node-metrics/src/mixnet.rs +++ b/nym-node/nym-node-metrics/src/mixnet.rs @@ -293,7 +293,6 @@ impl IngressMixingStats { self.final_hop_packets_dropped.load(Ordering::Relaxed) } - pub fn overflow_packets_dropped(&self) -> usize { self.overflow_packets_dropped.load(Ordering::Relaxed) } diff --git a/nym-node/src/config/mod.rs b/nym-node/src/config/mod.rs index e60e133524..c747cbac27 100644 --- a/nym-node/src/config/mod.rs +++ b/nym-node/src/config/mod.rs @@ -659,7 +659,7 @@ pub struct ReplayProtectionDebug { /// Channel capacity for the mixnet ingress channel. This determines the maximum number of /// packets that can be queued waiting for ingest processing. Once the queue is full packets - /// will still be taken off the wire, but dropped as the node is too busy to handle them. + /// will still be taken off the wire, but dropped as the node is too busy to handle them. pub ingress_channel_maximum_capacity: usize, /// Probability of false positives, fraction between 0 and 1 or a number indicating 1-in-p diff --git a/nym-node/src/config/old_configs/old_config_v12.rs b/nym-node/src/config/old_configs/old_config_v12.rs index 69b2d96b29..e3296c05b8 100644 --- a/nym-node/src/config/old_configs/old_config_v12.rs +++ b/nym-node/src/config/old_configs/old_config_v12.rs @@ -614,7 +614,8 @@ pub async fn try_upgrade_config_v12>( }, debug: ReplayProtectionDebug { unsafe_disabled: old_cfg.mixnet.replay_protection.debug.unsafe_disabled, - ingress_channel_maximum_capacity: ReplayProtectionDebug::DEFAULT_INGRESS_CHANNEL_MAXIMUM_CAPACITY, + ingress_channel_maximum_capacity: + ReplayProtectionDebug::DEFAULT_INGRESS_CHANNEL_MAXIMUM_CAPACITY, false_positive_rate: old_cfg.mixnet.replay_protection.debug.false_positive_rate, initial_expected_packets_per_second: old_cfg .mixnet diff --git a/nym-node/src/node/mixnet/listener.rs b/nym-node/src/node/mixnet/listener.rs index dbd212c60f..07d60eddb6 100644 --- a/nym-node/src/node/mixnet/listener.rs +++ b/nym-node/src/node/mixnet/listener.rs @@ -82,9 +82,8 @@ impl Listener { remote_addr, self.ingest_sender.clone(), ); - let join_handle = tokio::spawn(async move { - conn_handler.handle_connection(socket).await - }); + let join_handle = + tokio::spawn(async move { conn_handler.handle_connection(socket).await }); self.shared_data.log_connected_clients(); Some(join_handle) } @@ -152,7 +151,10 @@ impl ConnectionHandler { "noise_handshake_ms", handshake_start.elapsed().as_millis() as u64, ); - debug!("Noise responder handshake completed for {:?}", self.remote_addr); + debug!( + "Noise responder handshake completed for {:?}", + self.remote_addr + ); self.handle_stream(Framed::new(noise_stream, NymCodec)) .await } diff --git a/nym-node/src/node/mixnet/packet_ingest.rs b/nym-node/src/node/mixnet/packet_ingest.rs index 635550c0f6..ec38fe89cf 100644 --- a/nym-node/src/node/mixnet/packet_ingest.rs +++ b/nym-node/src/node/mixnet/packet_ingest.rs @@ -19,9 +19,7 @@ use nym_sphinx_framing::processing::{ use nym_sphinx_params::SphinxKeyRotation; use nym_sphinx_types::Delay; use nym_task::ShutdownToken; -use tokio::sync::mpsc; -use tokio::sync::mpsc::error::TrySendError; -use tracing::{Span, debug, error, instrument, trace, warn}; +use tracing::{Span, debug, error, info, instrument, trace, warn}; pub(crate) struct IngressNymPacket { packet: FramedNymPacket, @@ -46,6 +44,7 @@ impl IngressNymPacket { pub struct MixPacketIngest { shared: SharedData, + // Keep a copy of the channel to prevent accidental exit packet_sender: MixIngestSender, packet_receiver: MixIngestReceiver, } @@ -76,6 +75,51 @@ impl MixPacketIngest { } pub async fn run(&mut self, shutdown_token: ShutdownToken) { + let num_threads = std::thread::available_parallelism() + .expect("unable to query available parallelism") + .get(); + + let mut workers = tokio::task::JoinSet::new(); + + for i in 0..num_threads { + let recv = self.packet_receiver.clone(); + let worker = MixPacketIngestWorker { + packet_receiver: recv, + shared: SharedData { + processing_config: self.shared.processing_config, + sphinx_keys: self.shared.sphinx_keys.clone(), + replay_protection_filter: self.shared.replay_protection_filter.clone(), + mixnet_forwarder: self.shared.mixnet_forwarder.clone(), + final_hop: self.shared.final_hop.clone(), + noise_config: self.shared.noise_config.clone(), + metrics: self.shared.metrics.clone(), + shutdown_token: self.shared.shutdown_token.child_token(), + }, + }; + + let worker_token = shutdown_token.child_token(); + workers.spawn(worker.run(worker_token)); + } + info!("launched {num_threads} mix ingest workers"); + + while let Some(res) = workers.join_next().await { + if let Err(err) = res { + warn!("ingest worker closed with error: {err}"); + } else if !shutdown_token.is_cancelled() { + warn!("ingest worker closed unexpectedly!"); + } + } + } +} + +struct MixPacketIngestWorker { + shared: SharedData, + + packet_receiver: MixIngestReceiver, +} + +impl MixPacketIngestWorker { + pub async fn run(mut self, shutdown_token: ShutdownToken) { trace!("starting PacketIngest"); loop { tokio::select! { @@ -84,10 +128,12 @@ impl MixPacketIngest { debug!("PacketIngest: Received shutdown"); break; } - new_packet = self.packet_receiver.recv() => { - let Some(new_packet) = new_packet else { - todo!("the ingest receiver closed somehow") - }; + new_packet = self.packet_receiver.recv_async() => { + // this one is impossible to ever panic - the parent struct maintains a sender + // and waits for this process to end. Therefore it can't happen that ALL senders + // are dropped + #[allow(clippy::unwrap_used)] + let new_packet = new_packet.expect("the ingest receiver closed somehow"); self.handle_ingest_packet(new_packet).await; } } @@ -465,25 +511,26 @@ fn convert_to_metrics_version(processed: MixPacketVersion) -> PacketKind { pub fn mix_ingest_channels( processing_config: &ProcessingConfig, ) -> (MixIngestSender, MixIngestReceiver) { - let (tx, rx) = mpsc::channel(processing_config.ingress_channel_maximum_capacity); + let (tx, rx) = flume::bounded(processing_config.ingress_channel_maximum_capacity); (MixIngestSender(tx), rx) } #[derive(Clone)] -pub struct MixIngestSender(mpsc::Sender); +pub struct MixIngestSender(flume::Sender); impl MixIngestSender { pub fn ingest_packet(&self, packet: impl Into) -> io::Result<()> { let sender = &self.0; - let channel_capacity = sender.max_capacity(); - let channel_available = sender.capacity(); - let channel_used = channel_capacity - channel_available; + // we are using a bounded channel so unwrap is safe + #[allow(clippy::unwrap_used)] + let channel_capacity = sender.capacity().unwrap(); + let channel_used = sender.len(); let sending_res = sender.try_send(packet.into()); sending_res.map_err(|err| match err { - TrySendError::Full(_) => { + flume::TrySendError::Full(_) => { warn!( event = "mixnode.ingress_try_send", result = "full_dropped", @@ -493,7 +540,7 @@ impl MixIngestSender { ); io::Error::new(io::ErrorKind::WouldBlock, "ingress queue is full") } - TrySendError::Closed(_) => { + flume::TrySendError::Disconnected(_) => { debug!( event = "mixnode.ingress_try_send", result = "closed", @@ -507,4 +554,4 @@ impl MixIngestSender { } } -pub type MixIngestReceiver = mpsc::Receiver; +pub type MixIngestReceiver = flume::Receiver;