diff --git a/clients/client-core/src/client/real_messages_control/mod.rs b/clients/client-core/src/client/real_messages_control/mod.rs index 7d4f14df06..63ccf10940 100644 --- a/clients/client-core/src/client/real_messages_control/mod.rs +++ b/clients/client-core/src/client/real_messages_control/mod.rs @@ -8,13 +8,15 @@ use self::{ acknowledgement_control::AcknowledgementController, real_traffic_stream::OutQueueControl, }; -use crate::client::real_messages_control::acknowledgement_control::AcknowledgementControllerConnectors; -use crate::client::{ - inbound_messages::InputMessageReceiver, mix_traffic::BatchMixMessageSender, - topology_control::TopologyAccessor, +use crate::{ + client::{ + inbound_messages::InputMessageReceiver, mix_traffic::BatchMixMessageSender, + real_messages_control::acknowledgement_control::AcknowledgementControllerConnectors, + topology_control::TopologyAccessor, + }, + spawn_future, }; -use crate::spawn_future; -use client_connections::{ClosedConnectionReceiver, LaneQueueLength}; +use client_connections::{ClosedConnectionReceiver, LaneQueueLengths}; use futures::channel::mpsc; use gateway_client::AcknowledgementReceiver; use log::*; @@ -111,7 +113,7 @@ impl RealMessagesController { mix_sender: BatchMixMessageSender, topology_access: TopologyAccessor, #[cfg(feature = "reply-surb")] reply_key_storage: ReplyKeyStorage, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, closed_connection_rx: ClosedConnectionReceiver, ) -> Self { let rng = OsRng; @@ -162,7 +164,7 @@ impl RealMessagesController { rng, config.self_recipient, topology_access, - lane_queue_length, + lane_queue_lengths, closed_connection_rx, ); diff --git a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs index 537771a367..43be7524b3 100644 --- a/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs +++ b/clients/client-core/src/client/real_messages_control/real_traffic_stream.rs @@ -5,7 +5,7 @@ use crate::client::mix_traffic::BatchMixMessageSender; use crate::client::real_messages_control::acknowledgement_control::SentPacketNotificationSender; use crate::client::topology_control::TopologyAccessor; use client_connections::{ - ClosedConnectionReceiver, ConnectionId, LaneQueueLength, TransmissionLane, + ClosedConnectionReceiver, ConnectionId, LaneQueueLengths, TransmissionLane, }; use futures::channel::mpsc; use futures::task::{Context, Poll}; @@ -138,7 +138,7 @@ where closed_connection_rx: ClosedConnectionReceiver, /// Report queue lengths so that upstream can backoff sending data, and keep connections open. - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, } pub(crate) struct RealMessage { @@ -182,7 +182,7 @@ where rng: R, our_full_destination: Recipient, topology_access: TopologyAccessor, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, closed_connection_rx: ClosedConnectionReceiver, ) -> Self { OutQueueControl { @@ -198,7 +198,7 @@ where topology_access, transmission_buffer: Default::default(), closed_connection_rx, - lane_queue_length, + lane_queue_lengths, } } @@ -322,7 +322,7 @@ where // Update the published queue length let lane_length = self.transmission_buffer.lane_length(&lane); - self.lane_queue_length.set(&lane, lane_length); + self.lane_queue_lengths.set(&lane, lane_length); Some(real_next) } diff --git a/clients/native/src/client/mod.rs b/clients/native/src/client/mod.rs index 2dbff7a6ac..36cbf5e519 100644 --- a/clients/native/src/client/mod.rs +++ b/clients/native/src/client/mod.rs @@ -2,7 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use client_connections::{ - ClosedConnectionReceiver, ClosedConnectionSender, LaneQueueLength, TransmissionLane, + ClosedConnectionReceiver, ClosedConnectionSender, LaneQueueLengths, TransmissionLane, }; use client_core::client::cover_traffic_stream::LoopCoverTrafficStream; use client_core::client::inbound_messages::{ @@ -121,7 +121,7 @@ impl NymClient { ack_receiver: AcknowledgementReceiver, input_receiver: InputMessageReceiver, mix_sender: BatchMixMessageSender, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, closed_connection_rx: ClosedConnectionReceiver, shutdown: ShutdownListener, ) { @@ -152,7 +152,7 @@ impl NymClient { mix_sender, topology_accessor, reply_key_storage, - lane_queue_length, + lane_queue_lengths, closed_connection_rx, ) .start_with_shutdown(shutdown); @@ -423,7 +423,7 @@ impl NymClient { // Shared queue length data. Published by the `OutQueueController` in the client, and used // primarily to throttle incoming connections (e.g socks5 for attached network-requesters) - let shared_lane_queue_length = LaneQueueLength::new(); + let shared_lane_queue_length = LaneQueueLengths::new(); self.start_real_traffic_controller( shared_topology_accessor.clone(), diff --git a/clients/socks5/src/client/mod.rs b/clients/socks5/src/client/mod.rs index 32eb3c9acd..2d2bc4ce35 100644 --- a/clients/socks5/src/client/mod.rs +++ b/clients/socks5/src/client/mod.rs @@ -9,7 +9,7 @@ use crate::socks::{ authentication::{AuthenticationMethods, Authenticator, User}, server::SphinxSocksServer, }; -use client_connections::{ClosedConnectionReceiver, ClosedConnectionSender, LaneQueueLength}; +use client_connections::{ClosedConnectionReceiver, ClosedConnectionSender, LaneQueueLengths}; use client_core::client::cover_traffic_stream::LoopCoverTrafficStream; use client_core::client::inbound_messages::{ InputMessage, InputMessageReceiver, InputMessageSender, @@ -120,7 +120,7 @@ impl NymClient { input_receiver: InputMessageReceiver, mix_sender: BatchMixMessageSender, closed_connection_rx: ClosedConnectionReceiver, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, shutdown: ShutdownListener, ) { let mut controller_config = client_core::client::real_messages_control::Config::new( @@ -150,7 +150,7 @@ impl NymClient { mix_sender, topology_accessor, reply_key_storage, - lane_queue_length, + lane_queue_lengths, closed_connection_rx, ) .start_with_shutdown(shutdown); @@ -413,7 +413,7 @@ impl NymClient { // Shared queue length data. Published by the `OutQueueController` in the client, and used // primarily to throttle incoming connections - let shared_lane_queue_length = LaneQueueLength::new(); + let shared_lane_queue_length = LaneQueueLengths::new(); self.start_real_traffic_controller( shared_topology_accessor.clone(), diff --git a/common/client-connections/src/lib.rs b/common/client-connections/src/lib.rs index 67e7827830..265e21f9fd 100644 --- a/common/client-connections/src/lib.rs +++ b/common/client-connections/src/lib.rs @@ -25,12 +25,12 @@ pub type ClosedConnectionReceiver = mpsc::UnboundedReceiver; // The `OutQueueControl` publishes the backlog per lane, primarily so that upstream can slow down // if needed. #[derive(Clone)] -pub struct LaneQueueLength(std::sync::Arc>); +pub struct LaneQueueLengths(std::sync::Arc>); -impl LaneQueueLength { +impl LaneQueueLengths { pub fn new() -> Self { - LaneQueueLength(std::sync::Arc::new(std::sync::Mutex::new( - LaneQueueLengthInner { + LaneQueueLengths(std::sync::Arc::new(std::sync::Mutex::new( + LaneQueueLengthsInner { map: HashMap::new(), }, ))) @@ -54,20 +54,20 @@ impl LaneQueueLength { } } -impl Default for LaneQueueLength { +impl Default for LaneQueueLengths { fn default() -> Self { Self::new() } } -impl std::ops::Deref for LaneQueueLength { - type Target = std::sync::Arc>; +impl std::ops::Deref for LaneQueueLengths { + type Target = std::sync::Arc>; fn deref(&self) -> &Self::Target { &self.0 } } -pub struct LaneQueueLengthInner { +pub struct LaneQueueLengthsInner { map: HashMap, }