From c6cceb8d4d039139cb3eae87f346d457fe5ee6e2 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 27 Feb 2020 12:20:00 +0000 Subject: [PATCH] Starting and using the forwarding channel --- Cargo.lock | 1 + mixnode/src/node/listener.rs | 52 +++++++++++++++++++++++------------- mixnode/src/node/mod.rs | 36 ++++++++++++++++++++++--- 3 files changed, 67 insertions(+), 22 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 71eeba7ec2..9168cb857f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1475,6 +1475,7 @@ dependencies = [ "dotenv", "futures 0.3.1", "log", + "multi-tcp-client", "pemstore", "pretty_env_logger", "serde", diff --git a/mixnode/src/node/listener.rs b/mixnode/src/node/listener.rs index d0d8be2082..d12e9f4b6d 100644 --- a/mixnode/src/node/listener.rs +++ b/mixnode/src/node/listener.rs @@ -1,4 +1,5 @@ -use crate::node::packet_processing::PacketProcessor; +use crate::node::packet_processing::{MixProcessingResult, PacketProcessor}; +use futures::channel::mpsc; use log::*; use std::io; use std::net::SocketAddr; @@ -9,28 +10,34 @@ use tokio::task::JoinHandle; async fn process_received_packet( packet_data: [u8; sphinx::PACKET_SIZE], processing_data: PacketProcessor, + forwarding_channel: mpsc::UnboundedSender<(SocketAddr, Vec)>, ) { - // if we fail then we fail - processing_data.process_sphinx_packet(packet_data); - - // let fwd_data = match PacketProcessor::process_sphinx_data_packet( - // &packet_data, - // processing_data, - // ) - // .await - // { - // Ok(fwd_data) => fwd_data, - // Err(e) => { - // warn!("failed to process sphinx packet: {:?}", e); - // return; - // } - // }; - // PacketProcessor::wait_and_forward(fwd_data).await; + // all processing incl. delay was done, the only thing left is to forward it + match processing_data.process_sphinx_packet(packet_data).await { + Err(e) => debug!("We failed to process received sphinx packet - {:?}", e), + Ok(res) => match res { + MixProcessingResult::ForwardHop(hop_address, hop_data) => { + // send our data to tcp client for forwarding. If forwarding fails, then it fails, + // it's not like we can do anything about it + // + // in unbounded_send() failed it means that the receiver channel was disconnected + // and hence something weird must have happened without a way of recovering + forwarding_channel + .unbounded_send((hop_address, hop_data)) + .unwrap(); + processing_data.report_sent(hop_address); + } + MixProcessingResult::LoopMessage => { + warn!("Somehow processed a loop cover message that we haven't implemented yet!") + } + }, + } } async fn process_socket_connection( mut socket: tokio::net::TcpStream, packet_processor: PacketProcessor, + forwarding_channel: mpsc::UnboundedSender<(SocketAddr, Vec)>, ) { let mut buf = [0u8; sphinx::PACKET_SIZE]; loop { @@ -52,8 +59,8 @@ async fn process_socket_connection( // note: processing_data is relatively cheap (and safe) to clone - // it contains arc to private key and metrics reporter (which is just // a single mpsc unbounded sender) - // TODO: channel to tcp client packet_processor.clone(), + forwarding_channel.clone(), )) } Err(e) => { @@ -71,6 +78,7 @@ pub(crate) fn run_socket_listener( handle: &Handle, addr: SocketAddr, packet_processor: PacketProcessor, + forwarding_channel: mpsc::UnboundedSender<(SocketAddr, Vec)>, ) -> JoinHandle> { let handle_clone = handle.clone(); handle.spawn(async move { @@ -79,8 +87,14 @@ pub(crate) fn run_socket_listener( let (socket, _) = listener.accept().await?; let thread_packet_processor = packet_processor.clone(); + let forwarding_channel_clone = forwarding_channel.clone(); handle_clone.spawn(async move { - process_socket_connection(socket, thread_packet_processor).await; + process_socket_connection( + socket, + thread_packet_processor, + forwarding_channel_clone, + ) + .await; }); } }) diff --git a/mixnode/src/node/mod.rs b/mixnode/src/node/mod.rs index 93aac6fb96..74fc8f4d04 100644 --- a/mixnode/src/node/mod.rs +++ b/mixnode/src/node/mod.rs @@ -2,12 +2,15 @@ use crate::config::persistance::pathfinder::MixNodePathfinder; use crate::config::Config; use crate::node::packet_processing::PacketProcessor; use crypto::encryption; +use futures::channel::mpsc; use log::*; use pemstore::pemstore::PemStore; +use std::net::SocketAddr; use tokio::runtime::Runtime; mod listener; mod metrics; +mod packet_forwarding; pub(crate) mod packet_processing; mod presence; @@ -41,6 +44,7 @@ impl MixNode { } pub fn start_presence_notifier(&self) { + info!("Starting presence notifier..."); let notifier_config = presence::NotifierConfig::new( self.config.get_presence_directory_server(), self.config.get_announce_address(), @@ -52,6 +56,7 @@ impl MixNode { } pub fn start_metrics_reporter(&self) -> metrics::MetricsReporter { + info!("Starting metrics reporter..."); metrics::MetricsController::new( self.config.get_metrics_directory_server(), self.sphinx_keypair.public_key().to_base58_string(), @@ -60,7 +65,12 @@ impl MixNode { .start(self.runtime.handle()) } - pub fn start_socket_listener(&self, metrics_reporter: metrics::MetricsReporter) { + pub fn start_socket_listener( + &self, + metrics_reporter: metrics::MetricsReporter, + forwarding_channel: mpsc::UnboundedSender<(SocketAddr, Vec)>, + ) { + info!("Starting socket listener..."); // this is the only location where our private key is going to be copied // it will be held in memory owned by `MixNode` and inside an Arc of `PacketProcessor` let packet_processor = @@ -70,13 +80,33 @@ impl MixNode { self.runtime.handle(), self.config.get_listening_address(), packet_processor, + forwarding_channel, ); } + pub fn start_packet_forwarder(&mut self) -> mpsc::UnboundedSender<(SocketAddr, Vec)> { + info!("Starting packet forwarder..."); + // only temporary for test sake, will be replaced with dynamic list extremely soon + let initial_addresses = vec![ + "127.0.0.1:4000".parse().unwrap(), + "127.0.0.1:5000".parse().unwrap(), + "127.0.0.1:5001".parse().unwrap(), + "127.0.0.1:5002".parse().unwrap(), + ]; + self.runtime + .block_on(packet_forwarding::PacketForwarder::new( + initial_addresses, + self.config.get_packet_forwarding_initial_backoff(), + self.config.get_packet_forwarding_maximum_backoff(), + )) + .start(self.runtime.handle()) + } + pub fn run(&mut self) { - self.start_presence_notifier(); + let forwarding_channel = self.start_packet_forwarder(); let metrics_reporter = self.start_metrics_reporter(); - self.start_socket_listener(metrics_reporter); + self.start_socket_listener(metrics_reporter, forwarding_channel); + self.start_presence_notifier(); if let Err(e) = self.runtime.block_on(tokio::signal::ctrl_c()) { error!(