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

Feature/implicit runtime (#973)

Browse Source 
* Made mixnode runtime implicit

* Ibid for native and socks5 clients
This commit is contained in:
Jędrzej Stuczyński
2022-01-04 11:00:48 +00:00
committed by GitHub
parent 4f109169af
commit d95df4b286
18 changed files with 231 additions and 292 deletions
Download Patch File Download Diff File Expand all files Collapse all files
clients/client-core/src/client/cover_traffic_stream.rs
+2 -3
View File
@@ -13,7 +13,6 @@ use nymsphinx::utils::sample_poisson_duration;
use rand::{rngs::OsRng, CryptoRng, Rng};
use std::pin::Pin;
use std::sync::Arc;
use tokio::runtime::Handle;
use tokio::task::JoinHandle;
use tokio::time;
@@ -165,8 +164,8 @@ impl LoopCoverTrafficStream<OsRng> {
}
}
pub fn start(mut self, handle: &Handle) -> JoinHandle<()> {
handle.spawn(async move {
pub fn start(mut self) -> JoinHandle<()> {
tokio::spawn(async move {
self.run().await;
})
}
clients/client-core/src/client/mix_traffic.rs
+2 -3
View File
@@ -6,7 +6,6 @@ use futures::StreamExt;
use gateway_client::GatewayClient;
use log::*;
use nymsphinx::forwarding::packet::MixPacket;
use tokio::runtime::Handle;
use tokio::task::JoinHandle;
pub type BatchMixMessageSender = mpsc::UnboundedSender<Vec<MixPacket>>;
@@ -72,8 +71,8 @@ impl MixTrafficController {
}
}
pub fn start(mut self, handle: &Handle) -> JoinHandle<()> {
handle.spawn(async move {
pub fn start(mut self) -> JoinHandle<()> {
tokio::spawn(async move {
self.run().await;
})
}
clients/client-core/src/client/real_messages_control/mod.rs
+2 -5
View File
@@ -22,7 +22,6 @@ use nymsphinx::addressing::clients::Recipient;
use rand::{rngs::OsRng, CryptoRng, Rng};
use std::sync::Arc;
use std::time::Duration;
use tokio::runtime::Handle;
use tokio::task::JoinHandle;
mod acknowledgement_control;
@@ -170,10 +169,8 @@ impl RealMessagesController<OsRng> {
self.ack_control = Some(ack_control_fut.await.unwrap());
}
// &Handle is only passed for consistency sake with other client modules, but I think
// when we get to refactoring, we should apply gateway approach and make it implicit
pub fn start(mut self, handle: &Handle) -> JoinHandle<Self> {
handle.spawn(async move {
pub fn start(mut self) -> JoinHandle<Self> {
tokio::spawn(async move {
self.run().await;
self
})
clients/client-core/src/client/received_buffer.rs
+7 -8
View File
@@ -15,7 +15,6 @@ use nymsphinx::params::{ReplySurbEncryptionAlgorithm, ReplySurbKeyDigestAlgorith
use nymsphinx::receiver::{MessageReceiver, MessageRecoveryError, ReconstructedMessage};
use std::collections::HashSet;
use std::sync::Arc;
use tokio::runtime::Handle;
use tokio::task::JoinHandle;
// Buffer Requests to say "hey, send any reconstructed messages to this channel"
@@ -291,8 +290,8 @@ impl RequestReceiver {
}
}
fn start(mut self, handle: &Handle) -> JoinHandle<()> {
handle.spawn(async move {
fn start(mut self) -> JoinHandle<()> {
tokio::spawn(async move {
while let Some(request) = self.query_receiver.next().await {
match request {
ReceivedBufferMessage::ReceiverAnnounce(sender) => {
@@ -322,8 +321,8 @@ impl FragmentedMessageReceiver {
mixnet_packet_receiver,
}
}
fn start(mut self, handle: &Handle) -> JoinHandle<()> {
handle.spawn(async move {
fn start(mut self) -> JoinHandle<()> {
tokio::spawn(async move {
while let Some(new_messages) = self.mixnet_packet_receiver.next().await {
self.received_buffer.handle_new_received(new_messages).await;
}
@@ -355,9 +354,9 @@ impl ReceivedMessagesBufferController {
}
}
pub fn start(self, handle: &Handle) {
pub fn start(self) {
// TODO: should we do anything with JoinHandle(s) returned by start methods?
self.fragmented_message_receiver.start(handle);
self.request_receiver.start(handle);
self.fragmented_message_receiver.start();
self.request_receiver.start();
}
}
clients/client-core/src/client/topology_control.rs
+2 -3
View File
@@ -10,7 +10,6 @@ use std::ops::Deref;
use std::sync::Arc;
use std::time;
use std::time::Duration;
use tokio::runtime::Handle;
use tokio::sync::{RwLock, RwLockReadGuard};
use tokio::task::JoinHandle;
use topology::{nym_topology_from_bonds, NymTopology};
@@ -304,8 +303,8 @@ impl TopologyRefresher {
self.topology_accessor.is_routable().await
}
pub fn start(mut self, handle: &Handle) -> JoinHandle<()> {
handle.spawn(async move {
pub fn start(mut self) -> JoinHandle<()> {
tokio::spawn(async move {
loop {
tokio::time::sleep(self.refresh_rate).await;
self.refresh().await;
clients/native/src/client/mod.rs
+49 -67
View File
@@ -32,7 +32,6 @@ use nymsphinx::addressing::clients::Recipient;
use nymsphinx::addressing::nodes::NodeIdentity;
use nymsphinx::anonymous_replies::ReplySurb;
use nymsphinx::receiver::ReconstructedMessage;
use tokio::runtime::Runtime;
use crate::client::config::{Config, SocketType};
use crate::websocket;
@@ -44,11 +43,6 @@ pub struct NymClient {
/// key filepaths, etc.
config: Config,
/// Tokio runtime used for futures execution.
// TODO: JS: Personally I think I prefer the implicit way of using it that we've done with the
// gateway.
runtime: Runtime,
/// KeyManager object containing smart pointers to all relevant keys used by the client.
key_manager: KeyManager,
@@ -68,7 +62,6 @@ impl NymClient {
let key_manager = KeyManager::load_keys(&pathfinder).expect("failed to load stored keys");
NymClient {
runtime: Runtime::new().unwrap(),
config,
key_manager,
input_tx: None,
@@ -94,9 +87,6 @@ impl NymClient {
mix_tx: BatchMixMessageSender,
) {
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
let _guard = self.runtime.enter();
LoopCoverTrafficStream::new(
self.key_manager.ack_key(),
@@ -109,7 +99,7 @@ impl NymClient {
self.as_mix_recipient(),
topology_accessor,
)
.start(self.runtime.handle());
.start();
}
fn start_real_traffic_controller(
@@ -131,10 +121,6 @@ impl NymClient {
);
info!("Starting real traffic stream...");
// we need to explicitly enter runtime due to "next_delay: time::delay_for(Default::default())"
// set in the constructor [of OutQueueControl] which HAS TO be called within context of a tokio runtime
// When refactoring this restriction should definitely be removed.
let _guard = self.runtime.enter();
RealMessagesController::new(
controller_config,
@@ -144,7 +130,7 @@ impl NymClient {
topology_accessor,
reply_key_storage,
)
.start(self.runtime.handle());
.start();
}
// buffer controlling all messages fetched from provider
@@ -162,10 +148,10 @@ impl NymClient {
mixnet_receiver,
reply_key_storage,
)
.start(self.runtime.handle())
.start()
}
fn start_gateway_client(
async fn start_gateway_client(
&mut self,
mixnet_message_sender: MixnetMessageSender,
ack_sender: AcknowledgementSender,
@@ -182,47 +168,44 @@ impl NymClient {
let gateway_identity = identity::PublicKey::from_base58_string(gateway_id)
.expect("provided gateway id is invalid!");
self.runtime.block_on(async {
#[cfg(feature = "coconut")]
let bandwidth_controller = BandwidthController::new(
self.config.get_base().get_validator_api_endpoints(),
*self.key_manager.identity_keypair().public_key(),
);
#[cfg(not(feature = "coconut"))]
let bandwidth_controller = BandwidthController::new(
self.config.get_base().get_eth_endpoint(),
self.config.get_base().get_eth_private_key(),
self.config.get_base().get_backup_bandwidth_token_keys_dir(),
)
.expect("Could not create bandwidth controller");
#[cfg(feature = "coconut")]
let bandwidth_controller = BandwidthController::new(
self.config.get_base().get_validator_api_endpoints(),
*self.key_manager.identity_keypair().public_key(),
);
#[cfg(not(feature = "coconut"))]
let bandwidth_controller = BandwidthController::new(
self.config.get_base().get_eth_endpoint(),
self.config.get_base().get_eth_private_key(),
self.config.get_base().get_backup_bandwidth_token_keys_dir(),
)
.expect("Could not create bandwidth controller");
let mut gateway_client = GatewayClient::new(
gateway_address,
self.key_manager.identity_keypair(),
gateway_identity,
Some(self.key_manager.gateway_shared_key()),
mixnet_message_sender,
ack_sender,
self.config.get_base().get_gateway_response_timeout(),
Some(bandwidth_controller),
);
let mut gateway_client = GatewayClient::new(
gateway_address,
self.key_manager.identity_keypair(),
gateway_identity,
Some(self.key_manager.gateway_shared_key()),
mixnet_message_sender,
ack_sender,
self.config.get_base().get_gateway_response_timeout(),
Some(bandwidth_controller),
);
if self.config.get_base().get_testnet_mode() {
gateway_client.set_testnet_mode(true)
}
if self.config.get_base().get_testnet_mode() {
gateway_client.set_testnet_mode(true)
}
gateway_client
.authenticate_and_start()
.await
.expect("could not authenticate and start up the gateway connection");
gateway_client
.authenticate_and_start()
.await
.expect("could not authenticate and start up the gateway connection");
gateway_client
})
gateway_client
}
// future responsible for periodically polling directory server and updating
// the current global view of topology
fn start_topology_refresher(&mut self, topology_accessor: TopologyAccessor) {
async fn start_topology_refresher(&mut self, topology_accessor: TopologyAccessor) {
let topology_refresher_config = TopologyRefresherConfig::new(
self.config.get_base().get_validator_api_endpoints(),
self.config.get_base().get_topology_refresh_rate(),
@@ -233,13 +216,10 @@ impl NymClient {
// 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());
topology_refresher.refresh().await;
// TODO: a slightly more graceful termination here
if !self
.runtime
.block_on(topology_refresher.is_topology_routable())
{
if !topology_refresher.is_topology_routable().await {
panic!(
"The current network topology seem to be insufficient to route any packets through\
- check if enough nodes and a gateway are online"
@@ -247,7 +227,7 @@ impl NymClient {
}
info!("Starting topology refresher...");
topology_refresher.start(self.runtime.handle());
topology_refresher.start();
}
// controller for sending sphinx packets to mixnet (either real traffic or cover traffic)
@@ -260,7 +240,7 @@ impl NymClient {
gateway_client: GatewayClient,
) {
info!("Starting mix traffic controller...");
MixTrafficController::new(mix_rx, gateway_client).start(self.runtime.handle());
MixTrafficController::new(mix_rx, gateway_client).start();
}
fn start_websocket_listener(
@@ -273,8 +253,7 @@ impl NymClient {
let websocket_handler =
websocket::Handler::new(msg_input, buffer_requester, self.as_mix_recipient());
websocket::Listener::new(self.config.get_listening_port())
.start(self.runtime.handle(), websocket_handler);
websocket::Listener::new(self.config.get_listening_port()).start(websocket_handler);
}
/// EXPERIMENTAL DIRECT RUST API
@@ -321,9 +300,9 @@ impl NymClient {
}
/// 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()) {
pub async fn run_forever(&mut self) {
self.start().await;
if let Err(e) = tokio::signal::ctrl_c().await {
error!(
"There was an error while capturing SIGINT - {:?}. We will terminate regardless",
e
@@ -335,7 +314,7 @@ impl NymClient {
);
}
pub fn start(&mut self) {
pub async fn start(&mut self) {
info!("Starting nym client");
// channels for inter-component communication
// TODO: make the channels be internally created by the relevant components
@@ -367,14 +346,17 @@ impl NymClient {
// 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_topology_refresher(shared_topology_accessor.clone())
.await;
self.start_received_messages_buffer_controller(
received_buffer_request_receiver,
mixnet_messages_receiver,
reply_key_storage.clone(),
);
let gateway_client = self.start_gateway_client(mixnet_messages_sender, ack_sender);
let gateway_client = self
.start_gateway_client(mixnet_messages_sender, ack_sender)
.await;
self.start_mix_traffic_controller(sphinx_message_receiver, gateway_client);
self.start_real_traffic_controller(
clients/native/src/commands/init.rs
+14 -20
View File
@@ -219,7 +219,7 @@ fn show_address(config: &Config) {
println!("\nThe address of this client is: {}", client_recipient);
}
pub fn execute(matches: &ArgMatches) {
pub async fn execute(matches: ArgMatches<'static>) {
println!("Initialising client...");
let id = matches.value_of("id").unwrap(); // required for now
@@ -237,7 +237,7 @@ pub fn execute(matches: &ArgMatches) {
// TODO: ideally that should be the last thing that's being done to config.
// However, we are later further overriding it with gateway id
config = override_config(config, matches);
config = override_config(config, &matches);
if matches.is_present("fastmode") {
config.get_base_mut().set_high_default_traffic_volume();
}
@@ -250,26 +250,20 @@ pub fn execute(matches: &ArgMatches) {
let chosen_gateway_id = matches.value_of("gateway");
let registration_fut = async {
let gate_details = gateway_details(
config.get_base().get_validator_api_endpoints(),
chosen_gateway_id,
)
.await;
config
.get_base_mut()
.with_gateway_id(gate_details.identity_key.to_base58_string());
let shared_keys =
register_with_gateway(&gate_details, key_manager.identity_keypair()).await;
(shared_keys, gate_details.clients_address())
};
// TODO: is there perhaps a way to make it work without having to spawn entire runtime?
let rt = tokio::runtime::Runtime::new().unwrap();
let (shared_keys, gateway_listener) = rt.block_on(registration_fut);
let gateway_details = gateway_details(
config.get_base().get_validator_api_endpoints(),
chosen_gateway_id,
)
.await;
config
.get_base_mut()
.with_gateway_listener(gateway_listener);
.with_gateway_id(gateway_details.identity_key.to_base58_string());
let shared_keys =
register_with_gateway(&gateway_details, key_manager.identity_keypair()).await;
config
.get_base_mut()
.with_gateway_listener(gateway_details.clients_address());
key_manager.insert_gateway_shared_key(shared_keys);
let pathfinder = ClientKeyPathfinder::new_from_config(config.get_base());
clients/native/src/commands/run.rs
+3 -3
View File
@@ -82,7 +82,7 @@ fn version_check(cfg: &Config) -> bool {
}
}
pub fn execute(matches: &ArgMatches) {
pub async fn execute(matches: ArgMatches<'static>) {
let id = matches.value_of("id").unwrap();
let mut config = match Config::load_from_file(Some(id)) {
@@ -93,12 +93,12 @@ pub fn execute(matches: &ArgMatches) {
}
};
config = override_config(config, matches);
config = override_config(config, &matches);
if !version_check(&config) {
error!("failed the local version check");
return;
}
NymClient::new(config).run_forever();
NymClient::new(config).run_forever().await;
}
clients/native/src/main.rs
+6 -5
View File
@@ -7,7 +7,8 @@ pub mod client;
pub mod commands;
pub mod websocket;
fn main() {
#[tokio::main]
async fn main() {
dotenv::dotenv().ok();
setup_logging();
println!("{}", banner());
@@ -22,13 +23,13 @@ fn main() {
.subcommand(commands::upgrade::command_args())
.get_matches();
execute(arg_matches);
execute(arg_matches).await;
}
fn execute(matches: ArgMatches) {
async fn execute(matches: ArgMatches<'static>) {
match matches.subcommand() {
("init", Some(m)) => commands::init::execute(m),
("run", Some(m)) => commands::run::execute(m),
("init", Some(m)) => commands::init::execute(m.clone()).await,
("run", Some(m)) => commands::run::execute(m.clone()).await,
("upgrade", Some(m)) => commands::upgrade::execute(m),
_ => println!("{}", usage()),
}
clients/native/src/websocket/listener.rs
+2 -3
View File
@@ -5,7 +5,6 @@ use super::handler::Handler;
use log::*;
use std::{net::SocketAddr, process, sync::Arc};
use tokio::io::AsyncWriteExt;
use tokio::runtime;
use tokio::{sync::Notify, task::JoinHandle};
enum State {
@@ -87,9 +86,9 @@ impl Listener {
}
}
pub(crate) fn start(mut self, rt_handle: &runtime::Handle, handler: Handler) -> JoinHandle<()> {
pub(crate) fn start(mut self, handler: Handler) -> JoinHandle<()> {
info!("Running websocket on {:?}", self.address.to_string());
rt_handle.spawn(async move { self.run(handler).await })
tokio::spawn(async move { self.run(handler).await })
}
}
clients/socks5/src/client/mod.rs
+49 -67
View File
@@ -28,7 +28,6 @@ use gateway_client::{
use log::*;
use nymsphinx::addressing::clients::Recipient;
use nymsphinx::addressing::nodes::NodeIdentity;
use tokio::runtime::Runtime;
use crate::client::config::Config;
use crate::socks::{
@@ -43,11 +42,6 @@ pub struct NymClient {
/// key filepaths, etc.
config: Config,
/// Tokio runtime used for futures execution.
// TODO: JS: Personally I think I prefer the implicit way of using it that we've done with the
// gateway.
runtime: Runtime,
/// KeyManager object containing smart pointers to all relevant keys used by the client.
key_manager: KeyManager,
}
@@ -58,7 +52,6 @@ impl NymClient {
let key_manager = KeyManager::load_keys(&pathfinder).expect("failed to load stored keys");
NymClient {
runtime: Runtime::new().unwrap(),
config,
key_manager,
}
@@ -82,9 +75,6 @@ impl NymClient {
mix_tx: BatchMixMessageSender,
) {
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
let _guard = self.runtime.enter();
LoopCoverTrafficStream::new(
self.key_manager.ack_key(),
@@ -97,7 +87,7 @@ impl NymClient {
self.as_mix_recipient(),
topology_accessor,
)
.start(self.runtime.handle());
.start();
}
fn start_real_traffic_controller(
@@ -119,10 +109,6 @@ impl NymClient {
);
info!("Starting real traffic stream...");
// we need to explicitly enter runtime due to "next_delay: time::delay_for(Default::default())"
// set in the constructor [of OutQueueControl] which HAS TO be called within context of a tokio runtime
// When refactoring this restriction should definitely be removed.
let _guard = self.runtime.enter();
RealMessagesController::new(
controller_config,
@@ -132,7 +118,7 @@ impl NymClient {
topology_accessor,
reply_key_storage,
)
.start(self.runtime.handle());
.start();
}
// buffer controlling all messages fetched from provider
@@ -150,10 +136,10 @@ impl NymClient {
mixnet_receiver,
reply_key_storage,
)
.start(self.runtime.handle())
.start()
}
fn start_gateway_client(
async fn start_gateway_client(
&mut self,
mixnet_message_sender: MixnetMessageSender,
ack_sender: AcknowledgementSender,
@@ -170,47 +156,44 @@ impl NymClient {
let gateway_identity = identity::PublicKey::from_base58_string(gateway_id)
.expect("provided gateway id is invalid!");
self.runtime.block_on(async {
#[cfg(feature = "coconut")]
let bandwidth_controller = BandwidthController::new(
self.config.get_base().get_validator_api_endpoints(),
*self.key_manager.identity_keypair().public_key(),
);
#[cfg(not(feature = "coconut"))]
let bandwidth_controller = BandwidthController::new(
self.config.get_base().get_eth_endpoint(),
self.config.get_base().get_eth_private_key(),
self.config.get_base().get_backup_bandwidth_token_keys_dir(),
)
.expect("Could not create bandwidth controller");
#[cfg(feature = "coconut")]
let bandwidth_controller = BandwidthController::new(
self.config.get_base().get_validator_api_endpoints(),
*self.key_manager.identity_keypair().public_key(),
);
#[cfg(not(feature = "coconut"))]
let bandwidth_controller = BandwidthController::new(
self.config.get_base().get_eth_endpoint(),
self.config.get_base().get_eth_private_key(),
self.config.get_base().get_backup_bandwidth_token_keys_dir(),
)
.expect("Could not create bandwidth controller");
let mut gateway_client = GatewayClient::new(
gateway_address,
self.key_manager.identity_keypair(),
gateway_identity,
Some(self.key_manager.gateway_shared_key()),
mixnet_message_sender,
ack_sender,
self.config.get_base().get_gateway_response_timeout(),
Some(bandwidth_controller),
);
let mut gateway_client = GatewayClient::new(
gateway_address,
self.key_manager.identity_keypair(),
gateway_identity,
Some(self.key_manager.gateway_shared_key()),
mixnet_message_sender,
ack_sender,
self.config.get_base().get_gateway_response_timeout(),
Some(bandwidth_controller),
);
if self.config.get_base().get_testnet_mode() {
gateway_client.set_testnet_mode(true)
}
if self.config.get_base().get_testnet_mode() {
gateway_client.set_testnet_mode(true)
}
gateway_client
.authenticate_and_start()
.await
.expect("could not authenticate and start up the gateway connection");
gateway_client
.authenticate_and_start()
.await
.expect("could not authenticate and start up the gateway connection");
gateway_client
})
gateway_client
}
// future responsible for periodically polling directory server and updating
// the current global view of topology
fn start_topology_refresher(&mut self, topology_accessor: TopologyAccessor) {
async fn start_topology_refresher(&mut self, topology_accessor: TopologyAccessor) {
let topology_refresher_config = TopologyRefresherConfig::new(
self.config.get_base().get_validator_api_endpoints(),
self.config.get_base().get_topology_refresh_rate(),
@@ -221,13 +204,10 @@ impl NymClient {
// 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());
topology_refresher.refresh().await;
// TODO: a slightly more graceful termination here
if !self
.runtime
.block_on(topology_refresher.is_topology_routable())
{
if !topology_refresher.is_topology_routable().await {
panic!(
"The current network topology seem to be insufficient to route any packets through\
- check if enough nodes and a gateway are online"
@@ -235,7 +215,7 @@ impl NymClient {
}
info!("Starting topology refresher...");
topology_refresher.start(self.runtime.handle());
topology_refresher.start();
}
// controller for sending sphinx packets to mixnet (either real traffic or cover traffic)
@@ -248,7 +228,7 @@ impl NymClient {
gateway_client: GatewayClient,
) {
info!("Starting mix traffic controller...");
MixTrafficController::new(mix_rx, gateway_client).start(self.runtime.handle());
MixTrafficController::new(mix_rx, gateway_client).start();
}
fn start_socks5_listener(
@@ -267,14 +247,13 @@ impl NymClient {
self.config.get_provider_mix_address(),
self.as_mix_recipient(),
);
self.runtime
.spawn(async move { sphinx_socks.serve(msg_input, buffer_requester).await });
tokio::spawn(async move { sphinx_socks.serve(msg_input, buffer_requester).await });
}
/// 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()) {
pub async fn run_forever(&mut self) {
self.start().await;
if let Err(e) = tokio::signal::ctrl_c().await {
error!(
"There was an error while capturing SIGINT - {:?}. We will terminate regardless",
e
@@ -286,7 +265,7 @@ impl NymClient {
);
}
pub fn start(&mut self) {
pub async fn start(&mut self) {
info!("Starting nym client");
// channels for inter-component communication
// TODO: make the channels be internally created by the relevant components
@@ -318,14 +297,17 @@ impl NymClient {
// 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_topology_refresher(shared_topology_accessor.clone())
.await;
self.start_received_messages_buffer_controller(
received_buffer_request_receiver,
mixnet_messages_receiver,
reply_key_storage.clone(),
);
let gateway_client = self.start_gateway_client(mixnet_messages_sender, ack_sender);
let gateway_client = self
.start_gateway_client(mixnet_messages_sender, ack_sender)
.await;
self.start_mix_traffic_controller(sphinx_message_receiver, gateway_client);
self.start_real_traffic_controller(
clients/socks5/src/commands/init.rs
+14 -20
View File
@@ -219,7 +219,7 @@ fn show_address(config: &Config) {
println!("\nThe address of this client is: {}", client_recipient);
}
pub fn execute(matches: &ArgMatches) {
pub async fn execute(matches: ArgMatches<'static>) {
println!("Initialising client...");
let id = matches.value_of("id").unwrap(); // required for now
@@ -238,7 +238,7 @@ pub fn execute(matches: &ArgMatches) {
// TODO: ideally that should be the last thing that's being done to config.
// However, we are later further overriding it with gateway id
config = override_config(config, matches);
config = override_config(config, &matches);
if matches.is_present("fastmode") {
config.get_base_mut().set_high_default_traffic_volume();
}
@@ -251,26 +251,20 @@ pub fn execute(matches: &ArgMatches) {
let chosen_gateway_id = matches.value_of("gateway");
let registration_fut = async {
let gate_details = gateway_details(
config.get_base().get_validator_api_endpoints(),
chosen_gateway_id,
)
.await;
config
.get_base_mut()
.with_gateway_id(gate_details.identity_key.to_base58_string());
let shared_keys =
register_with_gateway(&gate_details, key_manager.identity_keypair()).await;
(shared_keys, gate_details.clients_address())
};
// TODO: is there perhaps a way to make it work without having to spawn entire runtime?
let rt = tokio::runtime::Runtime::new().unwrap();
let (shared_keys, gateway_listener) = rt.block_on(registration_fut);
let gateway_details = gateway_details(
config.get_base().get_validator_api_endpoints(),
chosen_gateway_id,
)
.await;
config
.get_base_mut()
.with_gateway_listener(gateway_listener);
.with_gateway_id(gateway_details.identity_key.to_base58_string());
let shared_keys =
register_with_gateway(&gateway_details, key_manager.identity_keypair()).await;
config
.get_base_mut()
.with_gateway_listener(gateway_details.clients_address());
key_manager.insert_gateway_shared_key(shared_keys);
let pathfinder = ClientKeyPathfinder::new_from_config(config.get_base());
clients/socks5/src/commands/run.rs
+3 -3
View File
@@ -88,7 +88,7 @@ fn version_check(cfg: &Config) -> bool {
}
}
pub fn execute(matches: &ArgMatches) {
pub async fn execute(matches: ArgMatches<'static>) {
let id = matches.value_of("id").unwrap();
let mut config = match Config::load_from_file(Some(id)) {
@@ -99,12 +99,12 @@ pub fn execute(matches: &ArgMatches) {
}
};
config = override_config(config, matches);
config = override_config(config, &matches);
if !version_check(&config) {
error!("failed the local version check");
return;
}
NymClient::new(config).run_forever();
NymClient::new(config).run_forever().await;
}
clients/socks5/src/main.rs
+6 -5
View File
@@ -7,7 +7,8 @@ pub mod client;
mod commands;
pub mod socks;
fn main() {
#[tokio::main]
async fn main() {
dotenv::dotenv().ok();
setup_logging();
println!("{}", banner());
@@ -22,13 +23,13 @@ fn main() {
.subcommand(commands::upgrade::command_args())
.get_matches();
execute(arg_matches);
execute(arg_matches).await;
}
fn execute(matches: ArgMatches) {
async fn execute(matches: ArgMatches<'static>) {
match matches.subcommand() {
("init", Some(m)) => commands::init::execute(m),
("run", Some(m)) => commands::run::execute(m),
("init", Some(m)) => commands::init::execute(m.clone()).await,
("run", Some(m)) => commands::run::execute(m.clone()).await,
("upgrade", Some(m)) => commands::upgrade::execute(m),
_ => println!("{}", usage()),
}
mixnode/src/commands/init.rs
+42 -46
View File
@@ -8,7 +8,6 @@ use crate::node::MixNode;
use clap::{App, Arg, ArgMatches};
use config::NymConfig;
use crypto::asymmetric::{encryption, identity};
use tokio::runtime::Runtime;
pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> {
App::new("init")
@@ -66,59 +65,56 @@ pub fn command_args<'a, 'b>() -> clap::App<'a, 'b> {
)
}
pub fn execute(matches: &ArgMatches) {
pub async fn execute(matches: ArgMatches<'static>) {
// TODO: this should probably be made implicit by slapping `#[tokio::main]` on our main method
// and then removing runtime from mixnode itself in `run`
let rt = Runtime::new().unwrap();
rt.block_on(async {
let id = matches.value_of(ID_ARG_NAME).unwrap();
println!("Initialising mixnode {}...", id);
let id = matches.value_of(ID_ARG_NAME).unwrap();
println!("Initialising mixnode {}...", id);
let already_init = if Config::default_config_file_path(Some(id)).exists() {
println!("Mixnode \"{}\" was already initialised before! Config information will be overwritten (but keys will be kept)!", id);
true
} else {
false
};
let already_init = if Config::default_config_file_path(Some(id)).exists() {
println!("Mixnode \"{}\" was already initialised before! Config information will be overwritten (but keys will be kept)!", id);
true
} else {
false
};
let mut config = Config::new(id);
config = override_config(config, matches);
let mut config = Config::new(id);
config = override_config(config, &matches);
// if node was already initialised, don't generate new keys
if !already_init {
let mut rng = rand::rngs::OsRng;
// if node was already initialised, don't generate new keys
if !already_init {
let mut rng = rand::rngs::OsRng;
let identity_keys = identity::KeyPair::new(&mut rng);
let sphinx_keys = encryption::KeyPair::new(&mut rng);
let pathfinder = MixNodePathfinder::new_from_config(&config);
pemstore::store_keypair(
&identity_keys,
&pemstore::KeyPairPath::new(
pathfinder.private_identity_key().to_owned(),
pathfinder.public_identity_key().to_owned(),
),
)
.expect("Failed to save identity keys");
let identity_keys = identity::KeyPair::new(&mut rng);
let sphinx_keys = encryption::KeyPair::new(&mut rng);
let pathfinder = MixNodePathfinder::new_from_config(&config);
pemstore::store_keypair(
&identity_keys,
&pemstore::KeyPairPath::new(
pathfinder.private_identity_key().to_owned(),
pathfinder.public_identity_key().to_owned(),
),
)
.expect("Failed to save identity keys");
pemstore::store_keypair(
&sphinx_keys,
&pemstore::KeyPairPath::new(
pathfinder.private_encryption_key().to_owned(),
pathfinder.public_encryption_key().to_owned(),
),
)
.expect("Failed to save sphinx keys");
pemstore::store_keypair(
&sphinx_keys,
&pemstore::KeyPairPath::new(
pathfinder.private_encryption_key().to_owned(),
pathfinder.public_encryption_key().to_owned(),
),
)
.expect("Failed to save sphinx keys");
println!("Saved mixnet identity and sphinx keypairs");
}
println!("Saved mixnet identity and sphinx keypairs");
}
let config_save_location = config.get_config_file_save_location();
config
.save_to_file(None)
.expect("Failed to save the config file");
println!("Saved configuration file to {:?}", config_save_location);
println!("Mixnode configuration completed.\n\n\n");
let config_save_location = config.get_config_file_save_location();
config
.save_to_file(None)
.expect("Failed to save the config file");
println!("Saved configuration file to {:?}", config_save_location);
println!("Mixnode configuration completed.\n\n\n");
MixNode::new(config).print_node_details()
})
MixNode::new(config).print_node_details()
}
mixnode/src/commands/run.rs
+3 -3
View File
@@ -92,7 +92,7 @@ fn version_check(cfg: &Config) -> bool {
}
}
pub fn execute(matches: &ArgMatches) {
pub async fn execute(matches: ArgMatches<'static>) {
let id = matches.value_of(ID_ARG_NAME).unwrap();
println!("Starting mixnode {}...", id);
@@ -105,7 +105,7 @@ pub fn execute(matches: &ArgMatches) {
}
};
config = override_config(config, matches);
config = override_config(config, &matches);
if !version_check(&config) {
error!("failed the local version check");
@@ -123,5 +123,5 @@ pub fn execute(matches: &ArgMatches) {
Select the correct version and install it to your machine. You will need to provide the following: \n ");
mixnode.print_node_details();
mixnode.run()
mixnode.run().await
}
mixnode/src/main.rs
+6 -5
View File
@@ -10,7 +10,8 @@ mod commands;
mod config;
mod node;
fn main() {
#[tokio::main]
async fn main() {
dotenv::dotenv().ok();
setup_logging();
println!("{}", banner());
@@ -28,14 +29,14 @@ fn main() {
.subcommand(commands::node_details::command_args())
.get_matches();
execute(arg_matches);
execute(arg_matches).await;
}
fn execute(matches: ArgMatches) {
async fn execute(matches: ArgMatches<'static>) {
match matches.subcommand() {
("describe", Some(m)) => commands::describe::execute(m),
("init", Some(m)) => commands::init::execute(m),
("run", Some(m)) => commands::run::execute(m),
("init", Some(m)) => commands::init::execute(m.clone()).await,
("run", Some(m)) => commands::run::execute(m.clone()).await,
("sign", Some(m)) => commands::sign::execute(m),
("upgrade", Some(m)) => commands::upgrade::execute(m),
("node-details", Some(m)) => commands::node_details::execute(m),
mixnode/src/node/mod.rs
+19 -23
View File
@@ -27,7 +27,6 @@ use rand::thread_rng;
use std::net::SocketAddr;
use std::process;
use std::sync::Arc;
use tokio::runtime::Runtime;
use version_checker::parse_version;
pub(crate) mod http;
@@ -279,33 +278,30 @@ impl MixNode {
);
}
pub fn run(&mut self) {
pub async fn run(&mut self) {
info!("Starting nym mixnode");
let runtime = Runtime::new().unwrap();
runtime.block_on(async {
if let Some(duplicate_node_key) = self.check_if_same_ip_node_exists().await {
if duplicate_node_key == self.identity_keypair.public_key().to_base58_string() {
warn!("You seem to have bonded your mixnode before starting it - that's highly unrecommended as in the future it might result in slashing");
} else {
log::error!(
"Our announce-host is identical to an existing node's announce-host! (its key is {:?})",
duplicate_node_key
);
return;
}
if let Some(duplicate_node_key) = self.check_if_same_ip_node_exists().await {
if duplicate_node_key == self.identity_keypair.public_key().to_base58_string() {
warn!("You seem to have bonded your mixnode before starting it - that's highly unrecommended as in the future it might result in slashing");
} else {
log::error!(
"Our announce-host is identical to an existing node's announce-host! (its key is {:?})",
duplicate_node_key
);
return;
}
}
let (node_stats_pointer, node_stats_update_sender) = self.start_node_stats_controller();
let delay_forwarding_channel = self.start_packet_delay_forwarder(node_stats_update_sender.clone());
self.start_socket_listener(node_stats_update_sender, delay_forwarding_channel);
let (node_stats_pointer, node_stats_update_sender) = self.start_node_stats_controller();
let delay_forwarding_channel =
self.start_packet_delay_forwarder(node_stats_update_sender.clone());
self.start_socket_listener(node_stats_update_sender, delay_forwarding_channel);
let atomic_verloc_results= self.start_verloc_measurements();
self.start_http_api(atomic_verloc_results, node_stats_pointer);
let atomic_verloc_results = self.start_verloc_measurements();
self.start_http_api(atomic_verloc_results, node_stats_pointer);
info!("Finished nym mixnode startup procedure - it should now be able to receive mix traffic!");
self.wait_for_interrupt().await
});
info!("Finished nym mixnode startup procedure - it should now be able to receive mix traffic!");
self.wait_for_interrupt().await
}
}