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 a50e2426e9..aadd8a6f25 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 @@ -285,7 +285,7 @@ where fn on_close_connection(&mut self, connection_id: ConnectionId) { //log::debug!("Removing lane for connection: {connection_id}"); //self.transmission_buffer - //.remove(&TransmissionLane::ConnectionId(connection_id)); + //.remove(&TransmissionLane::ConnectionId(connection_id)); } fn current_average_message_sending_delay(&self) -> Duration { diff --git a/clients/native/src/client/mod.rs b/clients/native/src/client/mod.rs index 0329437992..3a112d2767 100644 --- a/clients/native/src/client/mod.rs +++ b/clients/native/src/client/mod.rs @@ -287,7 +287,7 @@ impl NymClient { &self, buffer_requester: ReceivedBufferRequestSender, msg_input: InputMessageSender, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, closed_connection_tx: ClosedConnectionSender, ) { info!("Starting websocket listener..."); @@ -297,7 +297,7 @@ impl NymClient { closed_connection_tx, buffer_requester, &self.as_mix_recipient(), - lane_queue_length, + lane_queue_lengths, ); websocket::Listener::new(self.config.get_listening_port()).start(websocket_handler); @@ -425,7 +425,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 = LaneQueueLengths::new(); + let shared_lane_queue_lengths = LaneQueueLengths::new(); self.start_real_traffic_controller( shared_topology_accessor.clone(), @@ -433,7 +433,7 @@ impl NymClient { ack_receiver, input_receiver, sphinx_message_sender.clone(), - shared_lane_queue_length.clone(), + shared_lane_queue_lengths.clone(), closed_connection_rx, shutdown.subscribe(), ); @@ -454,7 +454,7 @@ impl NymClient { SocketType::WebSocket => self.start_websocket_listener( received_buffer_request_sender, input_sender, - shared_lane_queue_length, + shared_lane_queue_lengths, closed_connection_tx, ), SocketType::None => { diff --git a/clients/native/src/websocket/handler.rs b/clients/native/src/websocket/handler.rs index d8ec7250b9..54e7941d94 100644 --- a/clients/native/src/websocket/handler.rs +++ b/clients/native/src/websocket/handler.rs @@ -1,7 +1,7 @@ // Copyright 2021 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use client_connections::{ClosedConnectionSender, LaneQueueLength, TransmissionLane}; +use client_connections::{ClosedConnectionSender, LaneQueueLengths, TransmissionLane}; use client_core::client::{ inbound_messages::{InputMessage, InputMessageSender}, received_buffer::{ @@ -40,7 +40,7 @@ pub(crate) struct Handler { self_full_address: Recipient, socket: Option>, received_response_type: ReceivedResponseType, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, } // clone is used to use handler on a new connection, which initially is `None` @@ -53,7 +53,7 @@ impl Clone for Handler { self_full_address: self.self_full_address, socket: None, received_response_type: Default::default(), - lane_queue_length: self.lane_queue_length.clone(), + lane_queue_lengths: self.lane_queue_lengths.clone(), } } } @@ -72,7 +72,7 @@ impl Handler { closed_connection_tx: ClosedConnectionSender, buffer_requester: ReceivedBufferRequestSender, self_full_address: &Recipient, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, ) -> Self { Handler { msg_input, @@ -81,7 +81,7 @@ impl Handler { self_full_address: *self_full_address, socket: None, received_response_type: Default::default(), - lane_queue_length, + lane_queue_lengths, } } @@ -99,7 +99,7 @@ impl Handler { // WIP(JON) let queue_lengh = self - .lane_queue_length + .lane_queue_lengths .lock() .unwrap() .get(&lane) diff --git a/clients/socks5/src/client/mod.rs b/clients/socks5/src/client/mod.rs index 3fa8583e51..78508cd556 100644 --- a/clients/socks5/src/client/mod.rs +++ b/clients/socks5/src/client/mod.rs @@ -286,7 +286,7 @@ impl NymClient { buffer_requester: ReceivedBufferRequestSender, msg_input: InputMessageSender, closed_connection_tx: ClosedConnectionSender, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, shutdown: ShutdownListener, ) { info!("Starting socks5 listener..."); @@ -299,7 +299,7 @@ impl NymClient { authenticator, self.config.get_provider_mix_address(), self.as_mix_recipient(), - lane_queue_length, + lane_queue_lengths, shutdown, ); tokio::spawn(async move { @@ -415,7 +415,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 = LaneQueueLengths::new(); + let shared_lane_queue_lengths = LaneQueueLengths::new(); self.start_real_traffic_controller( shared_topology_accessor.clone(), @@ -424,7 +424,7 @@ impl NymClient { input_receiver, sphinx_message_sender.clone(), closed_connection_rx, - shared_lane_queue_length.clone(), + shared_lane_queue_lengths.clone(), shutdown.subscribe(), ); @@ -444,7 +444,7 @@ impl NymClient { received_buffer_request_sender, input_sender, closed_connection_tx, - shared_lane_queue_length, + shared_lane_queue_lengths, shutdown.subscribe(), ); diff --git a/clients/socks5/src/socks/client.rs b/clients/socks5/src/socks/client.rs index b39c030ca6..f8e9b4cf9b 100644 --- a/clients/socks5/src/socks/client.rs +++ b/clients/socks5/src/socks/client.rs @@ -4,7 +4,7 @@ use super::authentication::{AuthenticationMethods, Authenticator, User}; use super::request::{SocksCommand, SocksRequest}; use super::types::{ResponseCode, SocksProxyError}; use super::{RESERVED, SOCKS_VERSION}; -use client_connections::{LaneQueueLength, TransmissionLane}; +use client_connections::{LaneQueueLengths, TransmissionLane}; use client_core::client::inbound_messages::{InputMessage, InputMessageSender}; use futures::channel::mpsc; use futures::task::{Context, Poll}; @@ -142,7 +142,7 @@ pub(crate) struct SocksClient { service_provider: Recipient, self_address: Recipient, started_proxy: bool, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, shutdown_listener: ShutdownListener, active_connections: Arc>, } @@ -168,7 +168,7 @@ impl SocksClient { service_provider: Recipient, controller_sender: ControllerSender, self_address: Recipient, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, shutdown_listener: ShutdownListener, active_connections: Arc>, ) -> Self { @@ -190,7 +190,7 @@ impl SocksClient { service_provider, self_address, started_proxy: false, - lane_queue_length, + lane_queue_lengths, shutdown_listener, active_connections, } @@ -273,7 +273,7 @@ impl SocksClient { conn_receiver, input_sender, connection_id, - self.lane_queue_length.clone(), + self.lane_queue_lengths.clone(), self.shutdown_listener.clone(), ) .run(move |conn_id, read_data, socket_closed| { @@ -300,9 +300,11 @@ impl SocksClient { let request = SocksRequest::from_stream(&mut self.stream).await?; let remote_address = request.to_string(); - if active_connections > 50 { - log::warn!("Refusing SOCKS5: too many connections: {}", active_connections); + log::warn!( + "Refusing SOCKS5: too many connections: {}", + active_connections + ); self.refuse_connection_socks5().await; return Ok(()); } diff --git a/clients/socks5/src/socks/server.rs b/clients/socks5/src/socks/server.rs index 46677ae8ac..e1551dc167 100644 --- a/clients/socks5/src/socks/server.rs +++ b/clients/socks5/src/socks/server.rs @@ -4,7 +4,7 @@ use super::{ mixnet_responses::MixnetResponseListener, types::{ResponseCode, SocksProxyError}, }; -use client_connections::{ClosedConnectionSender, LaneQueueLength}; +use client_connections::{ClosedConnectionSender, LaneQueueLengths}; use client_core::client::{ inbound_messages::InputMessageSender, received_buffer::ReceivedBufferRequestSender, }; @@ -22,7 +22,7 @@ pub struct SphinxSocksServer { listening_address: SocketAddr, service_provider: Recipient, self_address: Recipient, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, shutdown: ShutdownListener, } @@ -33,7 +33,7 @@ impl SphinxSocksServer { authenticator: Authenticator, service_provider: Recipient, self_address: Recipient, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, shutdown: ShutdownListener, ) -> Self { // hardcode ip as we (presumably) ONLY want to listen locally. If we change it, we can @@ -45,7 +45,7 @@ impl SphinxSocksServer { listening_address: format!("{}:{}", ip, port).parse().unwrap(), service_provider, self_address, - lane_queue_length, + lane_queue_lengths, shutdown, } } @@ -95,7 +95,7 @@ impl SphinxSocksServer { self.service_provider, controller_sender.clone(), self.self_address, - self.lane_queue_length.clone(), + self.lane_queue_lengths.clone(), self.shutdown.clone(), active_connections.clone(), ); diff --git a/common/client-connections/src/lib.rs b/common/client-connections/src/lib.rs index d0b13460ca..d1782e8d90 100644 --- a/common/client-connections/src/lib.rs +++ b/common/client-connections/src/lib.rs @@ -80,7 +80,7 @@ impl std::ops::Deref for LaneQueueLengths { #[derive(Debug)] pub struct LaneQueueLengthsInner { - map: HashMap, + pub map: HashMap, } impl LaneQueueLengthsInner { diff --git a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs index 69b6197bc3..8dfdf60c6f 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/inbound.rs @@ -5,7 +5,7 @@ use super::MixProxySender; use super::SHUTDOWN_TIMEOUT; use crate::available_reader::AvailableReader; use bytes::Bytes; -use client_connections::LaneQueueLength; +use client_connections::LaneQueueLengths; use client_connections::TransmissionLane; use futures::FutureExt; use futures::StreamExt; @@ -40,7 +40,7 @@ async fn deal_with_data( message_sender: &mut OrderedMessageSender, mix_sender: &MixProxySender, adapter_fn: F, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, ) -> bool where F: Fn(ConnectionId, Vec, bool) -> S, @@ -77,7 +77,7 @@ where //loop { { let (queue_length, est_busy_conn) = { - let mut guard = lane_queue_length.lock().unwrap(); + let mut guard = lane_queue_lengths.lock().unwrap(); //let queue_length = *guard.get(&lane).unwrap_or(&0); //let queue_length = *guard.entry(lane).or_insert(0); let queue_length = guard.get(&lane).unwrap_or(0); @@ -156,7 +156,7 @@ pub(super) async fn run_inbound( mix_sender: MixProxySender, adapter_fn: F, shutdown_notify: Arc, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, mut shutdown_listener: ShutdownListener, ) -> OwnedReadHalf where @@ -179,7 +179,7 @@ where &mut message_sender, &mix_sender, &adapter_fn, - lane_queue_length.clone() + lane_queue_lengths.clone() ).await { break } diff --git a/common/socks5/proxy-helpers/src/proxy_runner/mod.rs b/common/socks5/proxy-helpers/src/proxy_runner/mod.rs index 4f3c8dceab..ee3fd6ed3c 100644 --- a/common/socks5/proxy-helpers/src/proxy_runner/mod.rs +++ b/common/socks5/proxy-helpers/src/proxy_runner/mod.rs @@ -2,7 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 use crate::connection_controller::ConnectionReceiver; -use client_connections::LaneQueueLength; +use client_connections::LaneQueueLengths; use futures::channel::mpsc; use socks5_requests::ConnectionId; use std::{sync::Arc, time::Duration}; @@ -46,7 +46,7 @@ pub struct ProxyRunner { local_destination_address: String, remote_source_address: String, connection_id: ConnectionId, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, // Listens to shutdown commands from higher up shutdown_listener: ShutdownListener, @@ -63,7 +63,7 @@ where mix_receiver: ConnectionReceiver, mix_sender: MixProxySender, connection_id: ConnectionId, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, shutdown_listener: ShutdownListener, ) -> Self { ProxyRunner { @@ -73,7 +73,7 @@ where local_destination_address, remote_source_address, connection_id, - lane_queue_length, + lane_queue_lengths, shutdown_listener, } } @@ -96,7 +96,7 @@ where self.mix_sender.clone(), adapter_fn, Arc::clone(&shutdown_notify), - self.lane_queue_length.clone(), + self.lane_queue_lengths.clone(), self.shutdown_listener.clone(), ); diff --git a/service-providers/network-requester/src/connection.rs b/service-providers/network-requester/src/connection.rs index e3b210b98d..5031ec1964 100644 --- a/service-providers/network-requester/src/connection.rs +++ b/service-providers/network-requester/src/connection.rs @@ -1,7 +1,7 @@ // Copyright 2020 - Nym Technologies SA // SPDX-License-Identifier: Apache-2.0 -use client_connections::LaneQueueLength; +use client_connections::LaneQueueLengths; use futures::channel::mpsc; use nymsphinx::addressing::clients::Recipient; use proxy_helpers::connection_controller::ConnectionReceiver; @@ -42,7 +42,7 @@ impl Connection { &mut self, mix_receiver: ConnectionReceiver, mix_sender: mpsc::UnboundedSender<(Socks5Message, Recipient)>, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, shutdown: ShutdownListener, ) { let stream = self.conn.take().unwrap(); @@ -56,7 +56,7 @@ impl Connection { mix_receiver, mix_sender, connection_id, - lane_queue_length, + lane_queue_lengths, shutdown, ) .run(move |conn_id, read_data, socket_closed| { diff --git a/service-providers/network-requester/src/core.rs b/service-providers/network-requester/src/core.rs index 2f50bcb07c..a3ffec094d 100644 --- a/service-providers/network-requester/src/core.rs +++ b/service-providers/network-requester/src/core.rs @@ -7,7 +7,7 @@ use crate::error::NetworkRequesterError; use crate::statistics::ServiceStatisticsCollector; use crate::websocket; use crate::websocket::TSWebsocketStream; -use client_connections::{ClosedConnectionReceiver, LaneQueueLength, TransmissionLane}; +use client_connections::{ClosedConnectionReceiver, LaneQueueLengths, TransmissionLane}; use futures::channel::mpsc; use futures::stream::{SplitSink, SplitStream}; use futures::{SinkExt, StreamExt}; @@ -118,7 +118,7 @@ impl ServiceProvider { async fn read_websocket_message( websocket_reader: &mut SplitStream, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, ) -> Option { while let Some(msg) = websocket_reader.next().await { let data = msg @@ -144,7 +144,7 @@ impl ServiceProvider { "received LaneQueueLength lane: {lane}, queue_length: {queue_length}" ); let lane = TransmissionLane::ConnectionId(lane); - let mut guard = lane_queue_length.lock().unwrap(); + let mut guard = lane_queue_lengths.lock().unwrap(); guard.map.insert(lane, queue_length); continue; } @@ -164,7 +164,7 @@ impl ServiceProvider { return_address: Recipient, controller_sender: ControllerSender, mix_input_sender: mpsc::UnboundedSender<(Socks5Message, Recipient)>, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, shutdown: ShutdownListener, ) { let mut conn = match Connection::new(conn_id, remote_addr.clone(), return_address).await { @@ -202,7 +202,7 @@ impl ServiceProvider { ); // run the proxy on the connection - conn.run_proxy(mix_receiver, mix_input_sender, lane_queue_length, shutdown) + conn.run_proxy(mix_receiver, mix_input_sender, lane_queue_lengths, shutdown) .await; // proxy is done - remove the access channel from the controller @@ -226,7 +226,7 @@ impl ServiceProvider { conn_id: ConnectionId, remote_addr: String, return_address: Recipient, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, shutdown: ShutdownListener, ) { if !self.open_proxy && !self.outbound_request_filter.check(&remote_addr) { @@ -254,7 +254,7 @@ impl ServiceProvider { return_address, controller_sender_clone, mix_input_sender_clone, - lane_queue_length, + lane_queue_lengths, shutdown, ) .await @@ -279,7 +279,7 @@ impl ServiceProvider { controller_sender: &mut ControllerSender, mix_input_sender: &mpsc::UnboundedSender<(Socks5Message, Recipient)>, stats_collector: Option, - lane_queue_length: LaneQueueLength, + lane_queue_lengths: LaneQueueLengths, shutdown: ShutdownListener, ) { let deserialized_msg = match Socks5Message::try_from_bytes(raw_request) { @@ -305,7 +305,7 @@ impl ServiceProvider { req.conn_id, req.remote_addr, req.return_address, - lane_queue_length, + lane_queue_lengths, shutdown, ) } @@ -354,7 +354,7 @@ impl ServiceProvider { // 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_lengths = LaneQueueLengths::new(); // Controller for managing all active connections. // We provide it with a ShutdownListener since it requires it, even though for the network @@ -398,7 +398,7 @@ impl ServiceProvider { loop { let received = match Self::read_websocket_message( &mut websocket_reader, - shared_lane_queue_length.clone(), + shared_lane_queue_lengths.clone(), ) .await { @@ -417,7 +417,7 @@ impl ServiceProvider { &mut controller_sender, &mix_input_sender, stats_collector.clone(), - shared_lane_queue_length.clone(), + shared_lane_queue_lengths.clone(), shutdown.subscribe(), ) .await;