mpmc ingest workers using flume

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",

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"

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 }

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)
     }

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

nym-node/src/config/old_configs/old_config_v12.rs

@@ -614,7 +614,8 @@ pub async fn try_upgrade_config_v12<P: AsRef<Path>>(
                 },
                 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

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
     }

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<IngressNymPacket>);
+pub struct MixIngestSender(flume::Sender<IngressNymPacket>);
 
 impl MixIngestSender {
     pub fn ingest_packet(&self, packet: impl Into<IngressNymPacket>) -> 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<IngressNymPacket>;
+pub type MixIngestReceiver = flume::Receiver<IngressNymPacket>;