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

Starting and using the forwarding channel

Browse Source
This commit is contained in:
Jedrzej Stuczynski
2020-02-27 12:20:00 +00:00
parent 8f48836fd4
commit c6cceb8d4d
3 changed files with 67 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mixnode/src/node/listener.rs
+33 -19
View File
@@ -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<u8>)>,
) {
// 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<u8>)>,
) {
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<u8>)>,
) -> JoinHandle<io::Result<()>> {
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;
});
}
})
mixnode/src/node/mod.rs
+33 -3
View File
@@ -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<u8>)>,
) {
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<u8>)> {
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!(