GRIN/nym
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update names

Browse Source
This commit is contained in:
Jon Häggblad
2022-11-19 11:06:57 +01:00
parent d4d739e267
commit d21cbc9746
11 changed files with 57 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/client-core/src/client/real_messages_control/real_traffic_stream.rs
+1 -1
View File
@@ -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 {
clients/native/src/client/mod.rs
+5 -5
View File
@@ -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 => {
clients/native/src/websocket/handler.rs
+6 -6
View File
@@ -1,7 +1,7 @@
// Copyright 2021 - Nym Technologies SA <contact@nymtech.net>
// 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<WebSocketStream<TcpStream>>,
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)
clients/socks5/src/client/mod.rs
+5 -5
View File
@@ -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(),
);
clients/socks5/src/socks/client.rs
+9 -7
View File
@@ -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<std::sync::Mutex<u64>>,
}
@@ -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<std::sync::Mutex<u64>>,
) -> 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(());
}
clients/socks5/src/socks/server.rs
+5 -5
View File
@@ -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(),
);
common/client-connections/src/lib.rs
+1 -1
View File
@@ -80,7 +80,7 @@ impl std::ops::Deref for LaneQueueLengths {
#[derive(Debug)]
pub struct LaneQueueLengthsInner {
map: HashMap<TransmissionLane, usize>,
pub map: HashMap<TransmissionLane, usize>,
}
impl LaneQueueLengthsInner {
common/socks5/proxy-helpers/src/proxy_runner/inbound.rs
+5 -5
View File
@@ -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<F, S>(
message_sender: &mut OrderedMessageSender,
mix_sender: &MixProxySender<S>,
adapter_fn: F,
lane_queue_length: LaneQueueLength,
lane_queue_lengths: LaneQueueLengths,
) -> bool
where
F: Fn(ConnectionId, Vec<u8>, 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<F, S>(
mix_sender: MixProxySender<S>,
adapter_fn: F,
shutdown_notify: Arc<Notify>,
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
}
common/socks5/proxy-helpers/src/proxy_runner/mod.rs
+5 -5
View File
@@ -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<S> {
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<S>,
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(),
);
service-providers/network-requester/src/connection.rs
+3 -3
View File
@@ -1,7 +1,7 @@
// Copyright 2020 - Nym Technologies SA <contact@nymtech.net>
// 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| {
service-providers/network-requester/src/core.rs
+12 -12
View File
@@ -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<TSWebsocketStream>,
lane_queue_length: LaneQueueLength,
lane_queue_lengths: LaneQueueLengths,
) -> Option<ReconstructedMessage> {
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<ServiceStatisticsCollector>,
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;