From 374da49bcd24ec10dd3e124cce6ebcf5979268fd Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 5 Mar 2020 12:54:17 +0000 Subject: [PATCH] Initial version of nym-client using persistent tcp connections Not yet fully refactored --- Cargo.lock | 1 + common/topology/src/lib.rs | 4 +- nym-client/Cargo.toml | 1 + nym-client/src/client/cover_traffic_stream.rs | 20 +- nym-client/src/client/mix_traffic.rs | 76 +++- nym-client/src/client/mod.rs | 388 +++++++++++------- nym-client/src/client/provider_poller.rs | 19 +- nym-client/src/client/real_traffic_stream.rs | 109 ++--- nym-client/src/client/received_buffer.rs | 4 +- nym-client/src/client/topology_control.rs | 160 ++++---- nym-client/src/commands/run.rs | 4 +- nym-client/src/sockets/tcp.rs | 64 ++- nym-client/src/sockets/ws.rs | 66 ++- 13 files changed, 552 insertions(+), 364 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 15f3faaa2e..2a6b9e79c6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1464,6 +1464,7 @@ dependencies = [ "healthcheck", "log", "mix-client", + "multi-tcp-client", "pem", "pemstore", "pretty_env_logger", diff --git a/common/topology/src/lib.rs b/common/topology/src/lib.rs index 5edba93d7d..eb81cdfd42 100644 --- a/common/topology/src/lib.rs +++ b/common/topology/src/lib.rs @@ -10,7 +10,9 @@ mod filter; pub mod mix; pub mod provider; -pub trait NymTopology: Sized + std::fmt::Debug + Send + Sync { +// TODO: Figure out why 'Clone' was required to have 'TopologyAccessor' working +// even though it only contains an Arc +pub trait NymTopology: Sized + std::fmt::Debug + Send + Sync + Clone { fn new(directory_server: String) -> Self; fn new_from_nodes( mix_nodes: Vec, diff --git a/nym-client/Cargo.toml b/nym-client/Cargo.toml index 87ba8de733..ac4bde4fc8 100644 --- a/nym-client/Cargo.toml +++ b/nym-client/Cargo.toml @@ -34,6 +34,7 @@ crypto = {path = "../common/crypto"} directory-client = { path = "../common/clients/directory-client" } healthcheck = { path = "../common/healthcheck" } mix-client = { path = "../common/clients/mix-client" } +multi-tcp-client = { path = "../common/clients/multi-tcp-client" } pemstore = {path = "../common/pemstore"} provider-client = { path = "../common/clients/provider-client" } sfw-provider-requests = { path = "../sfw-provider/sfw-provider-requests" } diff --git a/nym-client/src/client/cover_traffic_stream.rs b/nym-client/src/client/cover_traffic_stream.rs index 91faf16851..443f71e4c2 100644 --- a/nym-client/src/client/cover_traffic_stream.rs +++ b/nym-client/src/client/cover_traffic_stream.rs @@ -1,7 +1,7 @@ use crate::client::mix_traffic::{MixMessage, MixMessageSender}; use crate::client::topology_control::TopologyAccessor; use futures::task::{Context, Poll}; -use futures::{Stream, StreamExt}; +use futures::{Future, Stream, StreamExt}; use log::*; use sphinx::route::Destination; use std::pin::Pin; @@ -27,7 +27,7 @@ impl Stream for LoopCoverTrafficStream { // Perhaps this should be changed in the future. type Item = (); - fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { // it is not yet time to return a message if Pin::new(&mut self.next_delay).poll(cx).is_pending() { return Poll::Pending; @@ -36,7 +36,8 @@ impl Stream for LoopCoverTrafficStream { // we know it's time to send a message, so let's prepare delay for the next one // Get the `now` by looking at the current `delay` deadline let now = self.next_delay.deadline(); - let next_poisson_delay = mix_client::poisson::sample(self.average_message_sending_delay); + let next_poisson_delay = + mix_client::poisson::sample(self.average_cover_message_sending_delay); // The next interval value is `next_poisson_delay` after the one that just // yielded. @@ -47,11 +48,11 @@ impl Stream for LoopCoverTrafficStream { } } -impl LoopCoverTrafficStream { +impl LoopCoverTrafficStream { pub(crate) fn new( mix_tx: MixMessageSender, our_info: Destination, - mut topology_access: TopologyAccessor, + topology_access: TopologyAccessor, average_cover_message_sending_delay: time::Duration, average_packet_delay: time::Duration, ) -> Self { @@ -67,12 +68,7 @@ impl LoopCoverTrafficStream { async fn on_new_message(&mut self) { trace!("next cover message!"); - let topology = match self - .topology_access - .get_exclusive_topology_reference() - .await - .as_ref() - { + let topology = match self.topology_access.get_current_topology().await { None => { warn!("No valid topology detected - won't send any loop cover message this time"); return; @@ -83,7 +79,7 @@ impl LoopCoverTrafficStream { let cover_message = match mix_client::packet::loop_cover_message( self.our_info.address.clone(), self.our_info.identifier, - topology, + &topology, self.average_packet_delay, ) { Ok(message) => message, diff --git a/nym-client/src/client/mix_traffic.rs b/nym-client/src/client/mix_traffic.rs index bfcd3ddfc6..c71a7c3be6 100644 --- a/nym-client/src/client/mix_traffic.rs +++ b/nym-client/src/client/mix_traffic.rs @@ -1,10 +1,15 @@ use futures::channel::mpsc; use futures::StreamExt; -use log::{debug, error, info, trace}; +use log::*; use sphinx::SphinxPacket; use std::net::SocketAddr; +use std::time::Duration; +use tokio::runtime::Handle; +use tokio::task::JoinHandle; pub(crate) struct MixMessage(SocketAddr, SphinxPacket); +pub(crate) type MixMessageSender = mpsc::UnboundedSender; +pub(crate) type MixMessageReceiver = mpsc::UnboundedReceiver; impl MixMessage { pub(crate) fn new(address: SocketAddr, packet: SphinxPacket) -> Self { @@ -12,26 +17,57 @@ impl MixMessage { } } -pub(crate) struct MixTrafficController; +// TODO: put our TCP client here +pub(crate) struct MixTrafficController<'a> { + tcp_client: multi_tcp_client::Client<'a>, + mix_rx: MixMessageReceiver, +} -impl MixTrafficController { - pub(crate) async fn run(mut rx: mpsc::UnboundedReceiver) { - info!("Mix Traffic Controller started!"); - let mix_client = mix_client::MixClient::new(); - while let Some(mix_message) = rx.next().await { - debug!("Got a mix_message for {:?}", mix_message.0); - let send_res = mix_client.send(mix_message.1, mix_message.0).await; - match send_res { - Ok(_) => { - trace!("sent a mix message"); - } - // TODO: should there be some kind of threshold of failed messages - // that if reached, the application blows? - Err(e) => error!( - "We failed to send the message to {} :( - {:?}", - mix_message.0, e - ), - }; +impl MixTrafficController<'static> { + pub(crate) async fn new( + initial_endpoints: Vec, + initial_reconnection_backoff: Duration, + maximum_reconnection_backoff: Duration, + mix_rx: MixMessageReceiver, + ) -> Self { + let tcp_client_config = multi_tcp_client::Config::new( + initial_endpoints, + initial_reconnection_backoff, + maximum_reconnection_backoff, + ); + + MixTrafficController { + tcp_client: multi_tcp_client::Client::new(tcp_client_config).await, + mix_rx, } } + + async fn on_message(&mut self, mix_message: MixMessage) { + debug!("Got a mix_message for {:?}", mix_message.0); + match self + .tcp_client + .send(mix_message.0, &mix_message.1.to_bytes()) + .await + { + Ok(_) => trace!("sent a mix message"), + // TODO: should there be some kind of threshold of failed messages + // that if reached, the application blows? + Err(e) => error!( + "We failed to send the packet to {} - {:?}", + mix_message.0, e + ), + }; + } + + pub(crate) async fn run(&mut self) { + while let Some(mix_message) = self.mix_rx.next().await { + self.on_message(mix_message).await; + } + } + + pub(crate) fn start(mut self, handle: &Handle) -> JoinHandle<()> { + handle.spawn(async move { + self.run().await; + }) + } } diff --git a/nym-client/src/client/mod.rs b/nym-client/src/client/mod.rs index f2cc8b35d0..fcef9d8b64 100644 --- a/nym-client/src/client/mod.rs +++ b/nym-client/src/client/mod.rs @@ -1,14 +1,19 @@ -use crate::client::mix_traffic::MixTrafficController; -use crate::client::received_buffer::ReceivedMessagesBuffer; -use crate::client::topology_control::TopologyInnerRef; +use crate::client::cover_traffic_stream::LoopCoverTrafficStream; +use crate::client::mix_traffic::{MixMessageReceiver, MixMessageSender, MixTrafficController}; +use crate::client::provider_poller::{PolledMessagesReceiver, PolledMessagesSender}; +use crate::client::received_buffer::{ + ReceivedBufferRequestReceiver, ReceivedBufferRequestSender, ReceivedMessagesBufferController, +}; +use crate::client::topology_control::{ + TopologyAccessor, TopologyRefresher, TopologyRefresherConfig, +}; use crate::config::persistence::pathfinder::ClientPathfinder; use crate::config::{Config, SocketType}; use crate::sockets::tcp; use crate::sockets::ws; use crypto::identity::MixIdentityKeyPair; -use directory_client::presence::Topology; +use directory_client::presence; use futures::channel::mpsc; -use futures::join; use log::*; use pemstore::pemstore::PemStore; use sfw_provider_requests::AuthToken; @@ -24,14 +29,16 @@ mod real_traffic_stream; pub mod received_buffer; pub mod topology_control; +pub type InputMessageSender = mpsc::UnboundedSender; +pub type InputMessageReceiver = mpsc::UnboundedReceiver; + pub struct NymClient { + runtime: Runtime, config: Config, identity_keypair: MixIdentityKeyPair, // to be used by "send" function or socket, etc - pub input_tx: mpsc::UnboundedSender, - // the other end of the above channel - input_rx: mpsc::UnboundedReceiver, + pub input_tx: Option, } #[derive(Debug)] @@ -51,32 +58,220 @@ impl NymClient { pub fn new(config: Config) -> Self { let identity_keypair = Self::load_identity_keys(&config); - let (input_tx, input_rx) = mpsc::unbounded::(); NymClient { + runtime: Runtime::new().unwrap(), config, identity_keypair, - input_tx, - input_rx, + input_tx: None, } } + pub fn as_mix_destination(&self) -> Destination { + Destination::new( + self.identity_keypair.public_key().derive_address(), + // TODO: what with SURBs? + Default::default(), + ) + } + async fn get_provider_socket_address( - &self, - topology_ctrl_ref: TopologyInnerRef, + provider_id: String, + mut topology_accessor: TopologyAccessor, ) -> SocketAddr { - // this is temporary and assumes there exists only a single provider. - topology_ctrl_ref.read().await.topology.as_ref().unwrap() + topology_accessor.get_current_topology().await.as_ref().expect("The current network topoloy is empty - are you using correct directory server?") .providers() - .first() - .expect("Could not get a compatible provider from the initial network topology, are you using the right directory server?") + .iter() + .find(|provider| provider.pub_key == provider_id) + .expect(format!("Could not find provider with id {:?} - are you sure it is still online? Perhaps try to run `nym-client init` again to obtain a new provider", provider_id).as_ref()) .client_listener } - pub fn start(self) -> Result<(), Box> { - info!("Starting nym client"); - let mut rt = Runtime::new()?; + // future constantly pumping loop cover traffic at some specified average rate + // the pumped traffic goes to the MixTrafficController + fn start_cover_traffic_stream( + &self, + topology_accessor: TopologyAccessor, + mix_tx: MixMessageSender, + ) { + info!("Starting loop cover traffic stream..."); + // we need to explicitly enter runtime due to "next_delay: time::delay_for(Default::default())" + // set in the constructor which HAS TO be called within context of a tokio runtime + self.runtime + .enter(|| { + LoopCoverTrafficStream::new( + mix_tx, + self.as_mix_destination(), + topology_accessor, + self.config.get_loop_cover_traffic_average_delay(), + self.config.get_average_packet_delay(), + ) + }) + .start(self.runtime.handle()); + } + fn start_real_traffic_stream( + &self, + topology_accessor: TopologyAccessor, + mix_tx: MixMessageSender, + input_rx: InputMessageReceiver, + ) { + info!("Starting real traffic stream..."); + // we need to explicitly enter runtime due to "next_delay: time::delay_for(Default::default())" + // set in the constructor which HAS TO be called within context of a tokio runtime + self.runtime + .enter(|| { + real_traffic_stream::OutQueueControl::new( + mix_tx, + input_rx, + self.as_mix_destination(), + topology_accessor, + self.config.get_average_packet_delay(), + self.config.get_message_sending_average_delay(), + ) + }) + .start(self.runtime.handle()); + } + + // buffer controlling all messages fetched from provider + // required so that other components would be able to use them (say the websocket) + fn start_received_messages_buffer_controller( + &self, + query_receiver: ReceivedBufferRequestReceiver, + poller_receiver: PolledMessagesReceiver, + ) { + info!("Starting 'received messages buffer controller'..."); + ReceivedMessagesBufferController::new(query_receiver, poller_receiver) + .start(self.runtime.handle()) + } + + // future constantly trying to fetch any received messages from the provider + // the received messages are sent to ReceivedMessagesBuffer to be available to rest of the system + fn start_provider_poller( + &mut self, + topology_accessor: TopologyAccessor, + poller_input_tx: PolledMessagesSender, + ) { + info!("Starting provider poller..."); + // we already have our provider written in the config + let provider_id = self.config.get_provider_id(); + + let provider_client_listener_address = self.runtime.block_on( + Self::get_provider_socket_address(provider_id, topology_accessor), + ); + + let mut provider_poller = provider_poller::ProviderPoller::new( + poller_input_tx, + provider_client_listener_address, + self.identity_keypair.public_key().derive_address(), + self.config + .get_provider_auth_token() + .map(|str_token| AuthToken::try_from_base58_string(str_token).ok()) + .unwrap_or(None), + self.config.get_fetch_message_delay(), + ); + + if !provider_poller.is_registered() { + info!("Trying to perform initial provider registration..."); + self.runtime + .block_on(provider_poller.perform_initial_registration()) + .expect("Failed to perform initial provider registration"); + } + provider_poller.start(self.runtime.handle()); + } + + // future responsible for periodically polling directory server and updating + // the current global view of topology + fn start_topology_refresher( + &mut self, + topology_accessor: TopologyAccessor, + ) { + let healthcheck_keys = MixIdentityKeyPair::new(); + + let topology_refresher_config = TopologyRefresherConfig::new( + self.config.get_directory_server(), + self.config.get_topology_refresh_rate(), + healthcheck_keys, + self.config.get_topology_resolution_timeout(), + self.config.get_number_of_healthcheck_test_packets() as usize, + self.config.get_node_score_threshold(), + ); + let mut topology_refresher = + TopologyRefresher::new(topology_refresher_config, topology_accessor); + // before returning, block entire runtime to refresh the current network view so that any + // components depending on topology would see a non-empty view + info!("Obtaining initial network topology..."); + self.runtime.block_on(topology_refresher.refresh()); + info!("Starting topology refresher..."); + topology_refresher.start(self.runtime.handle()); + } + + // controller for sending sphinx packets to mixnet (either real traffic or cover traffic) + fn start_mix_traffic_controller(&mut self, mix_rx: MixMessageReceiver) { + info!("Starting mix trafic controller..."); + // TODO: possible optimisation: set the initial endpoints to all known mixes from layer 1 + let initial_mix_endpoints = Vec::new(); + self.runtime + .block_on(MixTrafficController::new( + initial_mix_endpoints, + self.config.get_packet_forwarding_initial_backoff(), + self.config.get_packet_forwarding_maximum_backoff(), + mix_rx, + )) + .start(self.runtime.handle()); + } + + fn start_socket_listener( + &self, + topology_accessor: TopologyAccessor, + received_messages_buffer_output_tx: ReceivedBufferRequestSender, + input_tx: InputMessageSender, + ) { + match self.config.get_socket_type() { + SocketType::WebSocket => { + ws::start_websocket( + self.runtime.handle(), + self.config.get_listening_port(), + input_tx, + received_messages_buffer_output_tx, + self.identity_keypair.public_key().derive_address(), + topology_accessor, + ); + } + SocketType::TCP => { + tcp::start_tcpsocket( + self.runtime.handle(), + self.config.get_listening_port(), + input_tx, + received_messages_buffer_output_tx, + self.identity_keypair.public_key().derive_address(), + topology_accessor, + ); + } + SocketType::None => (), + } + } + + /// EXPERIMENTAL DIRECT RUST API + pub fn send_message(&self) {} + + /// blocking version of `start` method. Will run forever (or until SIGINT is sent) + pub fn run_forever(&mut self) { + self.start(); + if let Err(e) = self.runtime.block_on(tokio::signal::ctrl_c()) { + error!( + "There was an error while capturing SIGINT - {:?}. We will terminate regardless", + e + ); + } + + println!( + "Received SIGINT - the mixnode will terminate now (threads are not YET nicely stopped)" + ); + } + + pub fn start(&mut self) { + info!("Starting nym client"); // channels for inter-component communication // mix_tx is the transmitter for any component generating sphinx packets that are to be sent to the mixnet @@ -94,142 +289,27 @@ impl NymClient { let (received_messages_buffer_output_tx, received_messages_buffer_output_rx) = mpsc::unbounded(); - let self_address = self.identity_keypair.public_key().derive_address(); + // channels responsible for controlling real messages + let (input_tx, input_rx) = mpsc::unbounded::(); - // generate same type of keys we have as our identity - let healthcheck_keys = MixIdentityKeyPair::new(); - - info!("Trying to obtain initial compatible network topology before starting up rest of modules"); - // TODO: when we switch to our graph topology, we need to remember to change 'Topology' type - let topology_controller_config = topology_control::TopologyControlConfig::::new( - self.config.get_directory_server(), - self.config.get_topology_refresh_rate(), - healthcheck_keys, - self.config.get_topology_resolution_timeout(), - self.config.get_number_of_healthcheck_test_packets() as usize, + // TODO: when we switch to our graph topology, we need to remember to change 'presence::Topology' type + let shared_topology_accessor = TopologyAccessor::::new(); + // the components are started in very specific order. Unless you know what you are doing, + // do not change that. + self.start_topology_refresher(shared_topology_accessor.clone()); + self.start_received_messages_buffer_controller( + received_messages_buffer_output_rx, + poller_input_rx, ); - - let topology_controller = rt.block_on(topology_control::TopologyControl::::new( - topology_controller_config, - )); - - let provider_client_listener_address = - rt.block_on(self.get_provider_socket_address(topology_controller.get_inner_ref())); - - let mut provider_poller = provider_poller::ProviderPoller::new( - poller_input_tx, - provider_client_listener_address, - self_address.clone(), - self.config - .get_provider_auth_token() - .map(|str_token| AuthToken::try_from_base58_string(str_token).ok()) - .unwrap_or(None), - self.config.get_fetch_message_delay(), + self.start_provider_poller(shared_topology_accessor.clone(), poller_input_tx); + self.start_mix_traffic_controller(mix_rx); + self.start_cover_traffic_stream(shared_topology_accessor.clone(), mix_tx.clone()); + self.start_real_traffic_stream(shared_topology_accessor.clone(), mix_tx, input_rx); + self.start_socket_listener( + shared_topology_accessor, + received_messages_buffer_output_tx, + input_tx.clone(), ); - - // registration - if let Err(err) = rt.block_on(provider_poller.perform_initial_registration()) { - panic!("Failed to perform initial provider registration: {:?}", err); - }; - - // setup all of futures for the components running on the client - - // buffer controlling all messages fetched from provider - // required so that other components would be able to use them (say the websocket) - let received_messages_buffer_controllers_future = rt.spawn( - ReceivedMessagesBuffer::new() - .start_controllers(poller_input_rx, received_messages_buffer_output_rx), - ); - - // controller for sending sphinx packets to mixnet (either real traffic or cover traffic) - let mix_traffic_future = rt.spawn(MixTrafficController::run(mix_rx)); - - // future constantly pumping loop cover traffic at some specified average rate - // the pumped traffic goes to the MixTrafficController - let loop_cover_traffic_future = - rt.spawn(cover_traffic_stream::start_loop_cover_traffic_stream( - mix_tx.clone(), - Destination::new(self_address.clone(), Default::default()), - topology_controller.get_inner_ref(), - self.config.get_loop_cover_traffic_average_delay(), - self.config.get_average_packet_delay(), - )); - - // cloning arguments required by OutQueueControl; required due to move - let input_rx = self.input_rx; - let topology_ref = topology_controller.get_inner_ref(); - let average_packet_delay = self.config.get_average_packet_delay(); - let message_sending_average_delay = self.config.get_message_sending_average_delay(); - let self_address_clone = self_address.clone(); - // future constantly pumping traffic at some specified average rate - // if a real message is available on 'input_rx' that might have been received from say - // the websocket, the real message is used, otherwise a loop cover message is generated - // the pumped traffic goes to the MixTrafficController - let out_queue_control_future = rt.spawn(async move { - real_traffic_stream::OutQueueControl::new( - mix_tx, - input_rx, - Destination::new(self_address_clone, Default::default()), - topology_ref, - average_packet_delay, - message_sending_average_delay, - ) - .run_out_queue_control() - .await - }); - - // future constantly trying to fetch any received messages from the provider - // the received messages are sent to ReceivedMessagesBuffer to be available to rest of the system - let provider_polling_future = rt.spawn(provider_poller.start_provider_polling()); - - match self.config.get_socket_type() { - SocketType::WebSocket => { - rt.spawn(ws::start_websocket( - self.config.get_listening_port(), - self.input_tx, - received_messages_buffer_output_tx, - self_address, - topology_controller.get_inner_ref(), - )); - } - SocketType::TCP => { - rt.spawn(tcp::start_tcpsocket( - self.config.get_listening_port(), - self.input_tx, - received_messages_buffer_output_tx, - self_address, - topology_controller.get_inner_ref(), - )); - } - SocketType::None => (), - } - - // future responsible for periodically polling directory server and updating - // the current global view of topology - let topology_refresher_future = rt.spawn(topology_controller.run_refresher()); - - rt.block_on(async { - let future_results = join!( - received_messages_buffer_controllers_future, - mix_traffic_future, - loop_cover_traffic_future, - out_queue_control_future, - provider_polling_future, - topology_refresher_future, - ); - - assert!( - future_results.0.is_ok() - && future_results.1.is_ok() - && future_results.2.is_ok() - && future_results.3.is_ok() - && future_results.4.is_ok() - && future_results.5.is_ok() - ); - }); - - // this line in theory should never be reached as the runtime should be permanently blocked on traffic senders - error!("The client went kaput..."); - Ok(()) + self.input_tx = Some(input_tx); } } diff --git a/nym-client/src/client/provider_poller.rs b/nym-client/src/client/provider_poller.rs index 526acb9e4b..24d329c9ae 100644 --- a/nym-client/src/client/provider_poller.rs +++ b/nym-client/src/client/provider_poller.rs @@ -1,10 +1,15 @@ use futures::channel::mpsc; -use log::{debug, error, info, trace, warn}; +use log::*; use provider_client::ProviderClientError; use sfw_provider_requests::AuthToken; use sphinx::route::DestinationAddressBytes; use std::net::SocketAddr; use std::time; +use tokio::runtime::Handle; +use tokio::task::JoinHandle; + +pub(crate) type PolledMessagesSender = mpsc::UnboundedSender>>; +pub(crate) type PolledMessagesReceiver = mpsc::UnboundedReceiver>>; pub(crate) struct ProviderPoller { polling_rate: time::Duration, @@ -31,11 +36,15 @@ impl ProviderPoller { } } + pub(crate) fn is_registered(&self) -> bool { + self.provider_client.is_registered() + } + // This method is only temporary until registration is moved to `client init` pub(crate) async fn perform_initial_registration(&mut self) -> Result<(), ProviderClientError> { debug!("performing initial provider registration"); - if !self.provider_client.is_registered() { + if !self.is_registered() { let auth_token = match self.provider_client.register().await { // in this particular case we can ignore this error Err(ProviderClientError::ClientAlreadyRegisteredError) => return Ok(()), @@ -52,8 +61,6 @@ impl ProviderPoller { } pub(crate) async fn start_provider_polling(self) { - info!("Starting provider poller"); - let loop_message = &mix_client::packet::LOOP_COVER_MESSAGE_PAYLOAD.to_vec(); let dummy_message = &sfw_provider_requests::DUMMY_MESSAGE_CONTENT.to_vec(); @@ -86,4 +93,8 @@ impl ProviderPoller { tokio::time::delay_for(self.polling_rate).await; } } + + pub(crate) fn start(self, handle: &Handle) -> JoinHandle<()> { + handle.spawn(async move { self.start_provider_polling().await }) + } } diff --git a/nym-client/src/client/real_traffic_stream.rs b/nym-client/src/client/real_traffic_stream.rs index 0fb9a198fe..43af644874 100644 --- a/nym-client/src/client/real_traffic_stream.rs +++ b/nym-client/src/client/real_traffic_stream.rs @@ -1,5 +1,5 @@ use crate::client::mix_traffic::MixMessage; -use crate::client::topology_control::TopologyInnerRef; +use crate::client::topology_control::TopologyAccessor; use crate::client::InputMessage; use futures::channel::mpsc; use futures::task::{Context, Poll}; @@ -8,6 +8,8 @@ use log::{error, info, trace, warn}; use sphinx::route::Destination; use std::pin::Pin; use std::time::Duration; +use tokio::runtime::Handle; +use tokio::task::JoinHandle; use tokio::time; use topology::NymTopology; @@ -18,7 +20,7 @@ pub(crate) struct OutQueueControl { mix_tx: mpsc::UnboundedSender, input_rx: mpsc::UnboundedReceiver, our_info: Destination, - topology_ctrl_ref: TopologyInnerRef, + topology_access: TopologyAccessor, } pub(crate) enum StreamMessage { @@ -60,12 +62,12 @@ impl Stream for OutQueueControl { } } -impl OutQueueControl { +impl OutQueueControl { pub(crate) fn new( mix_tx: mpsc::UnboundedSender, input_rx: mpsc::UnboundedReceiver, our_info: Destination, - topology: TopologyInnerRef, + topology_access: TopologyAccessor, average_packet_delay: Duration, average_message_sending_delay: Duration, ) -> Self { @@ -76,10 +78,57 @@ impl OutQueueControl { mix_tx, input_rx, our_info, - topology_ctrl_ref: topology, + topology_access, } } + async fn on_message(&mut self, next_message: StreamMessage) { + trace!("created new message"); + let topology = match self.topology_access.get_current_topology().await { + None => { + warn!("No valid topology detected - won't send any real or loop message this time"); + // TODO: this creates a potential problem: we can lose real messages if we were + // unable to get topology, perhaps we should store them in some buffer? + return; + } + Some(topology) => topology, + }; + + let next_packet = match next_message { + StreamMessage::Cover => mix_client::packet::loop_cover_message( + self.our_info.address.clone(), + self.our_info.identifier, + &topology, + self.average_packet_delay, + ), + StreamMessage::Real(real_message) => mix_client::packet::encapsulate_message( + real_message.0, + real_message.1, + &topology, + self.average_packet_delay, + ), + }; + + let next_packet = match next_packet { + Ok(message) => message, + Err(err) => { + error!( + "Somehow we managed to create an invalid traffic message - {:?}", + err + ); + return; + } + }; + + // if this one fails, there's no retrying because it means that either: + // - we run out of memory + // - the receiver channel is closed + // in either case there's no recovery and we can only panic + self.mix_tx + .unbounded_send(MixMessage::new(next_packet.0, next_packet.1)) + .unwrap(); + } + pub(crate) async fn run_out_queue_control(mut self) { // we should set initial delay only when we actually start the stream self.next_delay = time::delay_for(mix_client::poisson::sample( @@ -88,51 +137,11 @@ impl OutQueueControl { info!("starting out queue controller"); while let Some(next_message) = self.next().await { - trace!("created new message"); - let read_lock = self.topology_ctrl_ref.read().await; - let topology = match read_lock.topology.as_ref() { - None => { - warn!( - "No valid topology detected - won't send any loop cover message this time" - ); - continue; - } - Some(topology) => topology, - }; - - let next_packet = match next_message { - StreamMessage::Cover => mix_client::packet::loop_cover_message( - self.our_info.address.clone(), - self.our_info.identifier, - topology, - self.average_packet_delay, - ), - StreamMessage::Real(real_message) => mix_client::packet::encapsulate_message( - real_message.0, - real_message.1, - topology, - self.average_packet_delay, - ), - }; - - let next_packet = match next_packet { - Ok(message) => message, - Err(err) => { - error!( - "Somehow we managed to create an invalid traffic message - {:?}", - err - ); - continue; - } - }; - - // if this one fails, there's no retrying because it means that either: - // - we run out of memory - // - the receiver channel is closed - // in either case there's no recovery and we can only panic - self.mix_tx - .unbounded_send(MixMessage::new(next_packet.0, next_packet.1)) - .unwrap(); + self.on_message(next_message).await; } } + + pub(crate) fn start(self, handle: &Handle) -> JoinHandle<()> { + handle.spawn(async move { self.run_out_queue_control().await }) + } } diff --git a/nym-client/src/client/received_buffer.rs b/nym-client/src/client/received_buffer.rs index d776088151..123fc92882 100644 --- a/nym-client/src/client/received_buffer.rs +++ b/nym-client/src/client/received_buffer.rs @@ -38,7 +38,7 @@ impl ReceivedMessagesBuffer { async fn acquire_and_empty(&mut self) -> Vec> { trace!("Emptying the buffer and returning all messages"); - let mutex_guard = self.inner.lock().await; + let mut mutex_guard = self.inner.lock().await; std::mem::replace(&mut mutex_guard.messages, Vec::new()) } } @@ -116,7 +116,7 @@ impl ReceivedMessagesBufferController { } } - pub(crate) fn start(mut self, handle: &Handle) { + pub(crate) fn start(self, handle: &Handle) { // TODO: should we do anything with JoinHandle(s) returned by start methods? self.messsage_receiver.start(handle); self.request_receiver.start(handle); diff --git a/nym-client/src/client/topology_control.rs b/nym-client/src/client/topology_control.rs index c7403503c3..5bedcaddc8 100644 --- a/nym-client/src/client/topology_control.rs +++ b/nym-client/src/client/topology_control.rs @@ -1,23 +1,51 @@ use crate::built_info; use crypto::identity::MixIdentityKeyPair; +use futures::lock::Mutex; use healthcheck::HealthChecker; -use log::{error, info, trace, warn}; -use std::marker::PhantomData; +use log::*; use std::sync::Arc; use std::time; -use tokio::sync::RwLock as FRwLock; +use std::time::Duration; +use tokio::runtime::Handle; +// use tokio::sync::RwLock; +use tokio::task::JoinHandle; use topology::NymTopology; -const NODE_HEALTH_THRESHOLD: f64 = 0.0; +struct TopologyAccessorInner(Option); -// auxiliary type for ease of use -pub type TopologyInnerRef = Arc>>; +impl TopologyAccessorInner { + fn new() -> Self { + TopologyAccessorInner(None) + } -pub(crate) struct TopologyControl { - directory_server: String, - inner: Arc>>, - health_checker: HealthChecker, - refresh_rate: time::Duration, + fn update(&mut self, new: Option) { + self.0 = new; + } +} + +#[derive(Clone, Debug)] +pub(crate) struct TopologyAccessor { + // TODO: this requires some actual benchmarking to determine if obtaining mutex is not going + // to cause some bottlenecking and whether perhaps RwLock would be better + inner: Arc>>, +} + +impl TopologyAccessor { + pub(crate) fn new() -> Self { + TopologyAccessor { + inner: Arc::new(Mutex::new(TopologyAccessorInner::new())), + } + } + + async fn update_global_topology(&mut self, new_topology: Option) { + self.inner.lock().await.update(new_topology); + } + + pub(crate) async fn get_current_topology(&mut self) -> Option { + // TODO: considering topology is gotten quite frequently, the clone call might be rather + // expensive in the grand scheme of things... + self.inner.lock().await.0.clone() + } } #[derive(Debug)] @@ -26,40 +54,48 @@ enum TopologyError { NoValidPathsError, } -pub(crate) struct TopologyControlConfig { +pub(crate) struct TopologyRefresherConfig { directory_server: String, refresh_rate: time::Duration, identity_keypair: MixIdentityKeyPair, resolution_timeout: time::Duration, number_test_packets: usize, - - // the only reason I put phantom data here is so that we would we able to infer type - // of TopologyControl directly from the provided config rather than having to - // specify it during TopologyControl::::new() call - _topology_type_phantom: PhantomData<*const T>, + node_score_threshold: f64, } -impl TopologyControlConfig { +impl TopologyRefresherConfig { pub(crate) fn new( directory_server: String, refresh_rate: time::Duration, identity_keypair: MixIdentityKeyPair, resolution_timeout: time::Duration, number_test_packets: usize, + node_score_threshold: f64, ) -> Self { - TopologyControlConfig { + TopologyRefresherConfig { directory_server, refresh_rate, identity_keypair, resolution_timeout, number_test_packets, - _topology_type_phantom: PhantomData, + node_score_threshold, } } } -impl TopologyControl { - pub(crate) async fn new(cfg: TopologyControlConfig) -> Self { +pub(crate) struct TopologyRefresher { + directory_server: String, + topology_accessor: TopologyAccessor, + health_checker: HealthChecker, + refresh_rate: Duration, + node_score_threshold: f64, +} + +impl TopologyRefresher { + pub(crate) fn new( + cfg: TopologyRefresherConfig, + topology_accessor: TopologyAccessor, + ) -> Self { // this is a temporary solution as the healthcheck will eventually be moved to validators let health_checker = healthcheck::HealthChecker::new( cfg.resolution_timeout, @@ -67,26 +103,13 @@ impl TopologyControl { cfg.identity_keypair, ); - let mut topology_control = TopologyControl { + TopologyRefresher { directory_server: cfg.directory_server, - refresh_rate: cfg.refresh_rate, - inner: Arc::new(FRwLock::new(Inner::new(None))), + topology_accessor, health_checker, - }; - - // best effort approach to try to get a valid topology after call to 'new' - let initial_topology = match topology_control.get_current_compatible_topology().await { - Ok(topology) => Some(topology), - Err(err) => { - error!("Initial topology is invalid - {:?}. Right now it will be impossible to send any packets through the mixnet!", err); - None - } - }; - topology_control - .update_global_topology(initial_topology) - .await; - - topology_control + refresh_rate: cfg.refresh_rate, + node_score_threshold: cfg.node_score_threshold, + } } async fn get_current_compatible_topology(&self) -> Result { @@ -110,7 +133,7 @@ impl TopologyControl { }; let healthy_topology = healthcheck_scores - .filter_topology_by_score(&version_filtered_topology, NODE_HEALTH_THRESHOLD); + .filter_topology_by_score(&version_filtered_topology, self.node_score_threshold); // make sure you can still send a packet through the network: if !healthy_topology.can_construct_path_through() { @@ -120,42 +143,27 @@ impl TopologyControl { Ok(healthy_topology) } - pub(crate) fn get_inner_ref(&self) -> Arc>> { - self.inner.clone() + pub(crate) async fn refresh(&mut self) { + trace!("Refreshing the topology"); + let new_topology = match self.get_current_compatible_topology().await { + Ok(topology) => Some(topology), + Err(err) => { + warn!("the obtained topology seems to be invalid - {:?}, it will be impossible to send packets through", err); + None + } + }; + + self.topology_accessor + .update_global_topology(new_topology) + .await; } - async fn update_global_topology(&mut self, new_topology: Option) { - // acquire write lock - let mut write_lock = self.inner.write().await; - write_lock.topology = new_topology; - } - - pub(crate) async fn run_refresher(mut self) { - info!("Starting topology refresher"); - loop { - trace!("Refreshing the topology"); - let new_topology_res = self.get_current_compatible_topology().await; - - let new_topology = match new_topology_res { - Ok(topology) => Some(topology), - Err(err) => { - warn!("the obtained topology seems to be invalid - {:?}, it will be impossible to send packets through", err); - None - } - }; - - self.update_global_topology(new_topology).await; - tokio::time::delay_for(self.refresh_rate).await; - } - } -} - -pub struct Inner { - pub topology: Option, -} - -impl Inner { - fn new(topology: Option) -> Self { - Inner { topology } + pub(crate) fn start(mut self, handle: &Handle) -> JoinHandle<()> { + handle.spawn(async move { + loop { + self.refresh().await; + tokio::time::delay_for(self.refresh_rate).await; + } + }) } } diff --git a/nym-client/src/commands/run.rs b/nym-client/src/commands/run.rs index 9d2918a204..9f59192bf5 100644 --- a/nym-client/src/commands/run.rs +++ b/nym-client/src/commands/run.rs @@ -50,7 +50,5 @@ pub fn execute(matches: &ArgMatches) { .expect("Failed to load config file"); config = override_config(config, matches); - - let client = NymClient::new(config); - client.start().unwrap(); + NymClient::new(config).run_forever(); } diff --git a/nym-client/src/sockets/tcp.rs b/nym-client/src/sockets/tcp.rs index c8b65c0df4..54ae1281a9 100644 --- a/nym-client/src/sockets/tcp.rs +++ b/nym-client/src/sockets/tcp.rs @@ -1,5 +1,5 @@ -use crate::client::received_buffer::BufferResponse; -use crate::client::topology_control::TopologyInnerRef; +use crate::client::received_buffer::ReceivedBufferResponse; +use crate::client::topology_control::TopologyAccessor; use crate::client::InputMessage; use futures::channel::{mpsc, oneshot}; use futures::future::FutureExt; @@ -11,6 +11,8 @@ use std::convert::TryFrom; use std::io; use std::net::SocketAddr; use tokio::prelude::*; +use tokio::runtime::Handle; +use tokio::task::JoinHandle; use topology::NymTopology; const SEND_REQUEST_PREFIX: u8 = 1; @@ -110,7 +112,9 @@ impl ClientRequest { ServerResponse::Send } - async fn handle_fetch(mut msg_query: mpsc::UnboundedSender) -> ServerResponse { + async fn handle_fetch( + mut msg_query: mpsc::UnboundedSender, + ) -> ServerResponse { trace!("handle_fetch called"); let (res_tx, res_rx) = oneshot::channel(); if msg_query.send(res_tx).await.is_err() { @@ -133,9 +137,10 @@ impl ClientRequest { ServerResponse::Fetch { messages } } - async fn handle_get_clients(topology: &TopologyInnerRef) -> ServerResponse { - let topology_data = &topology.read().await.topology; - match topology_data { + async fn handle_get_clients( + mut topology_accessor: TopologyAccessor, + ) -> ServerResponse { + match topology_accessor.get_current_topology().await { Some(topology) => { let clients = topology .providers() @@ -228,7 +233,7 @@ async fn handle_connection( } ClientRequest::Fetch => ClientRequest::handle_fetch(request_handling_data.msg_query).await, ClientRequest::GetClients => { - ClientRequest::handle_get_clients(&request_handling_data.topology).await + ClientRequest::handle_get_clients(request_handling_data.topology_accessor).await } ClientRequest::OwnDetails => { ClientRequest::handle_own_details(request_handling_data.self_address).await @@ -240,17 +245,17 @@ async fn handle_connection( struct RequestHandlingData { msg_input: mpsc::UnboundedSender, - msg_query: mpsc::UnboundedSender, + msg_query: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: TopologyInnerRef, + topology_accessor: TopologyAccessor, } async fn accept_connection( mut socket: tokio::net::TcpStream, msg_input: mpsc::UnboundedSender, - msg_query: mpsc::UnboundedSender, + msg_query: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: TopologyInnerRef, + topology_accessor: TopologyAccessor, ) { let address = socket .peer_addr() @@ -271,7 +276,7 @@ async fn accept_connection( } Ok(n) => { let request_handling_data = RequestHandlingData { - topology: topology.clone(), + topology_accessor: topology_accessor.clone(), msg_input: msg_input.clone(), msg_query: msg_query.clone(), self_address: self_address.clone(), @@ -295,16 +300,16 @@ async fn accept_connection( } } -pub async fn start_tcpsocket( +pub(crate) async fn run_tcpsocket( listening_port: u16, message_tx: mpsc::UnboundedSender, - received_messages_query_tx: mpsc::UnboundedSender, + received_messages_query_tx: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: TopologyInnerRef, -) -> Result<(), TCPSocketError> { + topology_accessor: TopologyAccessor, +) { let address = SocketAddr::new("127.0.0.1".parse().unwrap(), listening_port); info!("Starting tcp socket listener at {:?}", address); - let mut listener = tokio::net::TcpListener::bind(address).await?; + let mut listener = tokio::net::TcpListener::bind(address).await.unwrap(); while let Ok((stream, _)) = listener.accept().await { // it's fine to be cloning the channel on all new connection, because in principle @@ -314,10 +319,27 @@ pub async fn start_tcpsocket( message_tx.clone(), received_messages_query_tx.clone(), self_address.clone(), - topology.clone(), + topology_accessor.clone(), )); } - - error!("The tcpsocket went kaput..."); - Ok(()) +} + +pub(crate) fn start_tcpsocket( + handle: &Handle, + listening_port: u16, + message_tx: mpsc::UnboundedSender, + received_messages_query_tx: mpsc::UnboundedSender, + self_address: DestinationAddressBytes, + topology_accessor: TopologyAccessor, +) -> JoinHandle<()> { + handle.spawn(async move { + run_tcpsocket( + listening_port, + message_tx, + received_messages_query_tx, + self_address, + topology_accessor, + ) + .await; + }) } diff --git a/nym-client/src/sockets/ws.rs b/nym-client/src/sockets/ws.rs index 845c4971c9..07a27acac7 100644 --- a/nym-client/src/sockets/ws.rs +++ b/nym-client/src/sockets/ws.rs @@ -1,5 +1,5 @@ -use crate::client::received_buffer::BufferResponse; -use crate::client::topology_control::TopologyInnerRef; +use crate::client::received_buffer::ReceivedBufferResponse; +use crate::client::topology_control::TopologyAccessor; use crate::client::InputMessage; use futures::channel::mpsc::{UnboundedReceiver, UnboundedSender}; use futures::channel::{mpsc, oneshot}; @@ -12,6 +12,8 @@ use sphinx::route::{Destination, DestinationAddressBytes}; use std::convert::TryFrom; use std::io; use std::net::SocketAddr; +use tokio::runtime::Handle; +use tokio::task::JoinHandle; use tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode; use tokio_tungstenite::tungstenite::protocol::{CloseFrame, Message}; use topology::NymTopology; @@ -19,10 +21,10 @@ use topology::NymTopology; struct Connection { address: SocketAddr, msg_input: mpsc::UnboundedSender, - msg_query: mpsc::UnboundedSender, + msg_query: mpsc::UnboundedSender, rx: UnboundedReceiver, self_address: DestinationAddressBytes, - topology: TopologyInnerRef, + topology_accessor: TopologyAccessor, tx: UnboundedSender, } @@ -49,7 +51,9 @@ impl Connection { ClientRequest::handle_send(message, recipient_address, self.msg_input.clone()).await } ClientRequest::Fetch => ClientRequest::handle_fetch(self.msg_query.clone()).await, - ClientRequest::GetClients => ClientRequest::handle_get_clients(&self.topology).await, + ClientRequest::GetClients => { + ClientRequest::handle_get_clients(self.topology_accessor.clone()).await + } ClientRequest::OwnDetails => { ClientRequest::handle_own_details(self.self_address.clone()).await } @@ -229,7 +233,9 @@ impl ClientRequest { ServerResponse::Send } - async fn handle_fetch(mut msg_query: mpsc::UnboundedSender) -> ServerResponse { + async fn handle_fetch( + mut msg_query: mpsc::UnboundedSender, + ) -> ServerResponse { let (res_tx, res_rx) = oneshot::channel(); if msg_query.send(res_tx).await.is_err() { warn!("Failed to handle_fetch. msg_query.send() is an error."); @@ -265,9 +271,10 @@ impl ClientRequest { ServerResponse::Fetch { messages } } - async fn handle_get_clients(topology: &TopologyInnerRef) -> ServerResponse { - let topology_data = &topology.read().await.topology; - match topology_data { + async fn handle_get_clients( + mut topology_accessor: TopologyAccessor, + ) -> ServerResponse { + match topology_accessor.get_current_topology().await { Some(topology) => { let clients = topology .providers() @@ -313,9 +320,9 @@ impl Into for ServerResponse { async fn accept_connection( stream: tokio::net::TcpStream, msg_input: mpsc::UnboundedSender, - msg_query: mpsc::UnboundedSender, + msg_query: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: TopologyInnerRef, + topology_accessor: TopologyAccessor, ) { let address = stream .peer_addr() @@ -339,7 +346,7 @@ async fn accept_connection( address, rx: msg_rx, tx: response_tx, - topology, + topology_accessor, msg_input, msg_query, self_address, @@ -391,16 +398,16 @@ async fn accept_connection( } } -pub async fn start_websocket( +pub(crate) async fn run_websocket( listening_port: u16, message_tx: mpsc::UnboundedSender, - received_messages_query_tx: mpsc::UnboundedSender, + received_messages_query_tx: mpsc::UnboundedSender, self_address: DestinationAddressBytes, - topology: TopologyInnerRef, -) -> Result<(), WebSocketError> { + topology_accessor: TopologyAccessor, +) { let address = SocketAddr::new("127.0.0.1".parse().unwrap(), listening_port); info!("Starting websocket listener at {:?}", address); - let mut listener = tokio::net::TcpListener::bind(address).await?; + let mut listener = tokio::net::TcpListener::bind(address).await.unwrap(); while let Ok((stream, _)) = listener.accept().await { // it's fine to be cloning the channel on all new connection, because in principle @@ -410,10 +417,27 @@ pub async fn start_websocket( message_tx.clone(), received_messages_query_tx.clone(), self_address.clone(), - topology.clone(), + topology_accessor.clone(), )); } - - error!("The websocket went kaput..."); - Ok(()) +} + +pub(crate) fn start_websocket( + handle: &Handle, + listening_port: u16, + message_tx: mpsc::UnboundedSender, + received_messages_query_tx: mpsc::UnboundedSender, + self_address: DestinationAddressBytes, + topology_accessor: TopologyAccessor, +) -> JoinHandle<()> { + handle.spawn(async move { + run_websocket( + listening_port, + message_tx, + received_messages_query_tx, + self_address, + topology_accessor, + ) + .await; + }) }