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

Initial version of nym-client using persistent tcp connections

Browse Source 
Not yet fully refactored
This commit is contained in:
Jedrzej Stuczynski
2020-03-05 12:54:17 +00:00
parent b343410840
commit 374da49bcd
13 changed files with 552 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -1464,6 +1464,7 @@ dependencies = [
"healthcheck",
"log",
"mix-client",
"multi-tcp-client",
"pem",
"pemstore",
"pretty_env_logger",
common/topology/src/lib.rs
+3 -1
View File
@@ -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<T>' 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<mix::Node>,
nym-client/Cargo.toml
+1
View File
@@ -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" }
nym-client/src/client/cover_traffic_stream.rs
+8 -12
View File
@@ -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<T: NymTopology> Stream for LoopCoverTrafficStream<T> {
// Perhaps this should be changed in the future.
type Item = ();
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
// 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<T: NymTopology> Stream for LoopCoverTrafficStream<T> {
// 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<T: NymTopology> Stream for LoopCoverTrafficStream<T> {
}
}
impl<T: NymTopology> LoopCoverTrafficStream<T> {
impl<T: 'static + NymTopology> LoopCoverTrafficStream<T> {
pub(crate) fn new(
mix_tx: MixMessageSender,
our_info: Destination,
mut topology_access: TopologyAccessor<T>,
topology_access: TopologyAccessor<T>,
average_cover_message_sending_delay: time::Duration,
average_packet_delay: time::Duration,
) -> Self {
@@ -67,12 +68,7 @@ impl<T: NymTopology> LoopCoverTrafficStream<T> {
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<T: NymTopology> LoopCoverTrafficStream<T> {
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,
nym-client/src/client/mix_traffic.rs
+56 -20
View File
@@ -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<MixMessage>;
pub(crate) type MixMessageReceiver = mpsc::UnboundedReceiver<MixMessage>;
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<MixMessage>) {
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<SocketAddr>,
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;
})
}
}
nym-client/src/client/mod.rs
+234 -154
View File
@@ -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<InputMessage>;
pub type InputMessageReceiver = mpsc::UnboundedReceiver<InputMessage>;
pub struct NymClient {
runtime: Runtime,
config: Config,
identity_keypair: MixIdentityKeyPair,
// to be used by "send" function or socket, etc
pub input_tx: mpsc::UnboundedSender<InputMessage>,
// the other end of the above channel
input_rx: mpsc::UnboundedReceiver<InputMessage>,
pub input_tx: Option<InputMessageSender>,
}
#[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::<InputMessage>();
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<T: NymTopology>(
&self,
topology_ctrl_ref: TopologyInnerRef<T>,
provider_id: String,
mut topology_accessor: TopologyAccessor<T>,
) -> 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<dyn std::error::Error>> {
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<T: 'static + NymTopology>(
&self,
topology_accessor: TopologyAccessor<T>,
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<T: 'static + NymTopology>(
&self,
topology_accessor: TopologyAccessor<T>,
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<T: NymTopology>(
&mut self,
topology_accessor: TopologyAccessor<T>,
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<T: 'static + NymTopology>(
&mut self,
topology_accessor: TopologyAccessor<T>,
) {
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<T: 'static + NymTopology>(
&self,
topology_accessor: TopologyAccessor<T>,
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::<InputMessage>();
// 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::<Topology>::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::<presence::Topology>::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::<Topology>::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);
}
}
nym-client/src/client/provider_poller.rs
+15 -4
View File
@@ -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<Vec<Vec<u8>>>;
pub(crate) type PolledMessagesReceiver = mpsc::UnboundedReceiver<Vec<Vec<u8>>>;
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 })
}
}
nym-client/src/client/real_traffic_stream.rs
+59 -50
View File
@@ -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<T: NymTopology> {
mix_tx: mpsc::UnboundedSender<MixMessage>,
input_rx: mpsc::UnboundedReceiver<InputMessage>,
our_info: Destination,
topology_ctrl_ref: TopologyInnerRef<T>,
topology_access: TopologyAccessor<T>,
}
pub(crate) enum StreamMessage {
@@ -60,12 +62,12 @@ impl<T: NymTopology> Stream for OutQueueControl<T> {
}
}
impl<T: NymTopology> OutQueueControl<T> {
impl<T: 'static + NymTopology> OutQueueControl<T> {
pub(crate) fn new(
mix_tx: mpsc::UnboundedSender<MixMessage>,
input_rx: mpsc::UnboundedReceiver<InputMessage>,
our_info: Destination,
topology: TopologyInnerRef<T>,
topology_access: TopologyAccessor<T>,
average_packet_delay: Duration,
average_message_sending_delay: Duration,
) -> Self {
@@ -76,10 +78,57 @@ impl<T: NymTopology> OutQueueControl<T> {
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<T: NymTopology> OutQueueControl<T> {
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 })
}
}
nym-client/src/client/received_buffer.rs
+2 -2
View File
@@ -38,7 +38,7 @@ impl ReceivedMessagesBuffer {
async fn acquire_and_empty(&mut self) -> Vec<Vec<u8>> {
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);
nym-client/src/client/topology_control.rs
+84 -76
View File
@@ -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<T: NymTopology>(Option<T>);
// auxiliary type for ease of use
pub type TopologyInnerRef<T> = Arc<FRwLock<Inner<T>>>;
impl<T: NymTopology> TopologyAccessorInner<T> {
fn new() -> Self {
TopologyAccessorInner(None)
}
pub(crate) struct TopologyControl<T: NymTopology> {
directory_server: String,
inner: Arc<FRwLock<Inner<T>>>,
health_checker: HealthChecker,
refresh_rate: time::Duration,
fn update(&mut self, new: Option<T>) {
self.0 = new;
}
}
#[derive(Clone, Debug)]
pub(crate) struct TopologyAccessor<T: NymTopology> {
// 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<Mutex<TopologyAccessorInner<T>>>,
}
impl<T: NymTopology> TopologyAccessor<T> {
pub(crate) fn new() -> Self {
TopologyAccessor {
inner: Arc::new(Mutex::new(TopologyAccessorInner::new())),
}
}
async fn update_global_topology(&mut self, new_topology: Option<T>) {
self.inner.lock().await.update(new_topology);
}
pub(crate) async fn get_current_topology(&mut self) -> Option<T> {
// 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<T: NymTopology> {
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::<type>::new() call
_topology_type_phantom: PhantomData<*const T>,
node_score_threshold: f64,
}
impl<T: NymTopology> TopologyControlConfig<T> {
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<T: NymTopology> TopologyControl<T> {
pub(crate) async fn new(cfg: TopologyControlConfig<T>) -> Self {
pub(crate) struct TopologyRefresher<T: NymTopology> {
directory_server: String,
topology_accessor: TopologyAccessor<T>,
health_checker: HealthChecker,
refresh_rate: Duration,
node_score_threshold: f64,
}
impl<T: 'static + NymTopology> TopologyRefresher<T> {
pub(crate) fn new(
cfg: TopologyRefresherConfig,
topology_accessor: TopologyAccessor<T>,
) -> 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<T: NymTopology> TopologyControl<T> {
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<T, TopologyError> {
@@ -110,7 +133,7 @@ impl<T: NymTopology> TopologyControl<T> {
};
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<T: NymTopology> TopologyControl<T> {
Ok(healthy_topology)
}
pub(crate) fn get_inner_ref(&self) -> Arc<FRwLock<Inner<T>>> {
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<T>) {
// 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<T: NymTopology> {
pub topology: Option<T>,
}
impl<T: NymTopology> Inner<T> {
fn new(topology: Option<T>) -> 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;
}
})
}
}
nym-client/src/commands/run.rs
+1 -3
View File
@@ -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();
}
nym-client/src/sockets/tcp.rs
+43 -21
View File
@@ -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<BufferResponse>) -> ServerResponse {
async fn handle_fetch(
mut msg_query: mpsc::UnboundedSender<ReceivedBufferResponse>,
) -> 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<T: NymTopology>(topology: &TopologyInnerRef<T>) -> ServerResponse {
let topology_data = &topology.read().await.topology;
match topology_data {
async fn handle_get_clients<T: NymTopology>(
mut topology_accessor: TopologyAccessor<T>,
) -> ServerResponse {
match topology_accessor.get_current_topology().await {
Some(topology) => {
let clients = topology
.providers()
@@ -228,7 +233,7 @@ async fn handle_connection<T: NymTopology>(
}
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<T: NymTopology>(
struct RequestHandlingData<T: NymTopology> {
msg_input: mpsc::UnboundedSender<InputMessage>,
msg_query: mpsc::UnboundedSender<BufferResponse>,
msg_query: mpsc::UnboundedSender<ReceivedBufferResponse>,
self_address: DestinationAddressBytes,
topology: TopologyInnerRef<T>,
topology_accessor: TopologyAccessor<T>,
}
async fn accept_connection<T: 'static + NymTopology>(
mut socket: tokio::net::TcpStream,
msg_input: mpsc::UnboundedSender<InputMessage>,
msg_query: mpsc::UnboundedSender<BufferResponse>,
msg_query: mpsc::UnboundedSender<ReceivedBufferResponse>,
self_address: DestinationAddressBytes,
topology: TopologyInnerRef<T>,
topology_accessor: TopologyAccessor<T>,
) {
let address = socket
.peer_addr()
@@ -271,7 +276,7 @@ async fn accept_connection<T: 'static + NymTopology>(
}
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<T: 'static + NymTopology>(
}
}
pub async fn start_tcpsocket<T: 'static + NymTopology>(
pub(crate) async fn run_tcpsocket<T: 'static + NymTopology>(
listening_port: u16,
message_tx: mpsc::UnboundedSender<InputMessage>,
received_messages_query_tx: mpsc::UnboundedSender<BufferResponse>,
received_messages_query_tx: mpsc::UnboundedSender<ReceivedBufferResponse>,
self_address: DestinationAddressBytes,
topology: TopologyInnerRef<T>,
) -> Result<(), TCPSocketError> {
topology_accessor: TopologyAccessor<T>,
) {
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<T: 'static + NymTopology>(
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<T: 'static + NymTopology>(
handle: &Handle,
listening_port: u16,
message_tx: mpsc::UnboundedSender<InputMessage>,
received_messages_query_tx: mpsc::UnboundedSender<ReceivedBufferResponse>,
self_address: DestinationAddressBytes,
topology_accessor: TopologyAccessor<T>,
) -> JoinHandle<()> {
handle.spawn(async move {
run_tcpsocket(
listening_port,
message_tx,
received_messages_query_tx,
self_address,
topology_accessor,
)
.await;
})
}
nym-client/src/sockets/ws.rs
+45 -21
View File
@@ -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<T: NymTopology> {
address: SocketAddr,
msg_input: mpsc::UnboundedSender<InputMessage>,
msg_query: mpsc::UnboundedSender<BufferResponse>,
msg_query: mpsc::UnboundedSender<ReceivedBufferResponse>,
rx: UnboundedReceiver<Message>,
self_address: DestinationAddressBytes,
topology: TopologyInnerRef<T>,
topology_accessor: TopologyAccessor<T>,
tx: UnboundedSender<Message>,
}
@@ -49,7 +51,9 @@ impl<T: NymTopology> Connection<T> {
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<BufferResponse>) -> ServerResponse {
async fn handle_fetch(
mut msg_query: mpsc::UnboundedSender<ReceivedBufferResponse>,
) -> 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<T: NymTopology>(topology: &TopologyInnerRef<T>) -> ServerResponse {
let topology_data = &topology.read().await.topology;
match topology_data {
async fn handle_get_clients<T: NymTopology>(
mut topology_accessor: TopologyAccessor<T>,
) -> ServerResponse {
match topology_accessor.get_current_topology().await {
Some(topology) => {
let clients = topology
.providers()
@@ -313,9 +320,9 @@ impl Into<Message> for ServerResponse {
async fn accept_connection<T: 'static + NymTopology>(
stream: tokio::net::TcpStream,
msg_input: mpsc::UnboundedSender<InputMessage>,
msg_query: mpsc::UnboundedSender<BufferResponse>,
msg_query: mpsc::UnboundedSender<ReceivedBufferResponse>,
self_address: DestinationAddressBytes,
topology: TopologyInnerRef<T>,
topology_accessor: TopologyAccessor<T>,
) {
let address = stream
.peer_addr()
@@ -339,7 +346,7 @@ async fn accept_connection<T: 'static + NymTopology>(
address,
rx: msg_rx,
tx: response_tx,
topology,
topology_accessor,
msg_input,
msg_query,
self_address,
@@ -391,16 +398,16 @@ async fn accept_connection<T: 'static + NymTopology>(
}
}
pub async fn start_websocket<T: 'static + NymTopology>(
pub(crate) async fn run_websocket<T: 'static + NymTopology>(
listening_port: u16,
message_tx: mpsc::UnboundedSender<InputMessage>,
received_messages_query_tx: mpsc::UnboundedSender<BufferResponse>,
received_messages_query_tx: mpsc::UnboundedSender<ReceivedBufferResponse>,
self_address: DestinationAddressBytes,
topology: TopologyInnerRef<T>,
) -> Result<(), WebSocketError> {
topology_accessor: TopologyAccessor<T>,
) {
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<T: 'static + NymTopology>(
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<T: 'static + NymTopology>(
handle: &Handle,
listening_port: u16,
message_tx: mpsc::UnboundedSender<InputMessage>,
received_messages_query_tx: mpsc::UnboundedSender<ReceivedBufferResponse>,
self_address: DestinationAddressBytes,
topology_accessor: TopologyAccessor<T>,
) -> JoinHandle<()> {
handle.spawn(async move {
run_websocket(
listening_port,
message_tx,
received_messages_query_tx,
self_address,
topology_accessor,
)
.await;
})
}