From 56a8118ea3dcd824ba38e5f496a17a7b802780a0 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 19 Dec 2019 12:55:18 +0000 Subject: [PATCH 1/4] Sending presence inside tokio reactor --- .gitignore | 6 ++++-- src/main.rs | 6 ------ src/provider/mod.rs | 18 +++++++++++++++--- src/provider/presence.rs | 21 ++++++++++++++------- 4 files changed, 33 insertions(+), 18 deletions(-) diff --git a/.gitignore b/.gitignore index c384ecb2ef..7fd7f19fc0 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,6 @@ -/target +/targetString **/*.rs.bk /*/target -/test_inbox \ No newline at end of file +/test_inbox +.idea +target \ No newline at end of file diff --git a/src/main.rs b/src/main.rs index 00662a0f59..24309f7440 100644 --- a/src/main.rs +++ b/src/main.rs @@ -72,12 +72,6 @@ fn run(matches: &ArgMatches) { let config = new_config(matches); let provider = ServiceProvider::new(&config); - // Start sending presence notifications in a separate thread - thread::spawn(move || { - let notifier = presence::Notifier::new(&config); - notifier.run(); - }); - provider.start().unwrap() } diff --git a/src/provider/mod.rs b/src/provider/mod.rs index 3bb6557a52..a5cb8a2dfb 100644 --- a/src/provider/mod.rs +++ b/src/provider/mod.rs @@ -34,8 +34,8 @@ pub struct Config { } impl Config { - pub fn public_key_string(&self) -> String { - let key_bytes = self.public_key.to_bytes().to_vec(); + pub fn public_key_string(public_key: MontgomeryPoint) -> String { + let key_bytes = public_key.to_bytes().to_vec(); base64::encode_config(&key_bytes, base64::URL_SAFE) } } @@ -93,9 +93,11 @@ impl ClientLedger { } pub struct ServiceProvider { + directory_server: String, mix_network_address: SocketAddr, client_network_address: SocketAddr, secret_key: Scalar, + public_key: MontgomeryPoint, store_dir: PathBuf, registered_clients_ledger: ClientLedger, } @@ -106,9 +108,11 @@ impl ServiceProvider { mix_network_address: config.mix_socket_address, client_network_address: config.client_socket_address, secret_key: config.secret_key, + public_key: config.public_key, store_dir: PathBuf::from(config.store_dir.clone()), // TODO: load initial ledger from file registered_clients_ledger: ClientLedger::new(), + directory_server: config.directory_server.clone(), } } @@ -268,6 +272,13 @@ impl ServiceProvider { let mut rt = Runtime::new()?; // let mut h = rt.handle(); + let presence_notifier = presence::Notifier::new( + self.directory_server, + self.mix_network_address.clone(), + self.public_key, + ); + + let presence_future = rt.spawn(presence_notifier.run()); let mix_future = rt.spawn(ServiceProvider::start_mixnet_listening( self.mix_network_address, self.secret_key, @@ -281,7 +292,8 @@ impl ServiceProvider { )); // Spawn the root task rt.block_on(async { - let future_results = futures::future::join(mix_future, client_future).await; + let future_results = + futures::future::join3(mix_future, client_future, presence_future).await; assert!(future_results.0.is_ok() && future_results.1.is_ok()); }); diff --git a/src/provider/presence.rs b/src/provider/presence.rs index 5368214a5d..9b08de810e 100644 --- a/src/provider/presence.rs +++ b/src/provider/presence.rs @@ -1,11 +1,13 @@ use crate::provider; use nym_client::clients::directory; use nym_client::clients::directory::DirectoryClient; -use std::thread; use std::time::Duration; +use crate::provider::Config; +use curve25519_dalek::montgomery::MontgomeryPoint; use nym_client::clients::directory::presence::MixProviderPresence; use nym_client::clients::directory::requests::presence_providers_post::PresenceMixProviderPoster; +use std::net::SocketAddr; pub struct Notifier { pub net_client: directory::Client, @@ -13,14 +15,18 @@ pub struct Notifier { } impl Notifier { - pub fn new(config: &provider::Config) -> Notifier { + pub fn new( + directory_server_address: String, + host: SocketAddr, + pub_key: MontgomeryPoint, + ) -> Notifier { let directory_config = directory::Config { - base_url: config.directory_server.clone(), + base_url: directory_server_address, }; let net_client = directory::Client::new(directory_config); let presence = MixProviderPresence { - host: config.mix_socket_address.to_string(), - pub_key: config.public_key_string(), + host: host.to_string(), + pub_key: Config::public_key_string(pub_key), registered_clients: vec![], }; Notifier { @@ -36,10 +42,11 @@ impl Notifier { .unwrap(); } - pub fn run(&self) { + pub async fn run(self) { loop { self.notify(); - thread::sleep(Duration::from_secs(5)); + let delay_duration = Duration::from_secs(5); + tokio::time::delay_for(delay_duration).await; } } } From a8eb5ec4c228d24a1f3ec5499251920d092be594 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 19 Dec 2019 13:07:07 +0000 Subject: [PATCH 2/4] Moved client processing data behind just an arc but put the ledger behind arc+mutex --- src/provider/client_handling/mod.rs | 37 +++++++++++++---------------- src/provider/mod.rs | 15 ++++++++---- 2 files changed, 28 insertions(+), 24 deletions(-) diff --git a/src/provider/client_handling/mod.rs b/src/provider/client_handling/mod.rs index 6d22cad3d2..d252fc3cc2 100644 --- a/src/provider/client_handling/mod.rs +++ b/src/provider/client_handling/mod.rs @@ -52,14 +52,14 @@ impl From for ClientProcessingError { #[derive(Debug)] pub(crate) struct ClientProcessingData { store_dir: PathBuf, - registered_clients_ledger: ClientLedger, + registered_clients_ledger: Arc>, secret_key: Scalar, } impl ClientProcessingData { pub(crate) fn new( store_dir: PathBuf, - registered_clients_ledger: ClientLedger, + registered_clients_ledger: Arc>, secret_key: Scalar, ) -> Self { ClientProcessingData { @@ -69,17 +69,17 @@ impl ClientProcessingData { } } - pub(crate) fn add_arc_futures_mutex(self) -> Arc> { - Arc::new(FMutex::new(self)) + pub(crate) fn add_arc(self) -> Arc { + Arc::new(self) } } -pub(crate) struct ClientRequestProcessor(()); +pub(crate) struct ClientRequestProcessor; impl ClientRequestProcessor { pub(crate) async fn process_client_request( data: &[u8], - processing_data: Arc>, + processing_data: Arc, ) -> Result, ClientProcessingError> { let client_request = ProviderRequests::from_bytes(&data)?; println!("Received the following request: {:?}", client_request); @@ -100,20 +100,19 @@ impl ClientRequestProcessor { async fn process_pull_messages_request( req: PullRequest, - processing_data: Arc>, + processing_data: Arc, ) -> Result { // TODO: this lock is completely unnecessary as we're only reading the data. // Wait for https://github.com/nymtech/nym-sfw-provider/issues/19 to resolve. - let unlocked = processing_data.lock().await; + let unlocked_ledger = processing_data.registered_clients_ledger.lock().await; println!("Processing pull!"); - if unlocked.registered_clients_ledger.has_token(req.auth_token) { - let store_dir_clone = unlocked.store_dir.clone(); + if unlocked_ledger.has_token(req.auth_token) { // drop the mutex so that we could do IO without blocking others wanting to get the lock - drop(unlocked); + drop(unlocked_ledger); let retrieved_messages = ClientStorage::retrieve_client_files( req.destination_address, - store_dir_clone.as_path(), + processing_data.store_dir.as_path(), )?; Ok(PullResponse::new(retrieved_messages)) } else { @@ -123,22 +122,20 @@ impl ClientRequestProcessor { async fn register_new_client( req: RegisterRequest, - processing_data: Arc>, + processing_data: Arc, ) -> Result { println!("Processing register new client request!"); - let mut unlocked = processing_data.lock().await; + let mut unlocked_ledger = processing_data.registered_clients_ledger.lock().await; let auth_token = ClientRequestProcessor::generate_new_auth_token( req.destination_address.to_vec(), - unlocked.secret_key, + processing_data.secret_key, ); - if !unlocked.registered_clients_ledger.has_token(auth_token) { - unlocked - .registered_clients_ledger - .insert_token(auth_token, req.destination_address); + if !unlocked_ledger.has_token(auth_token) { + unlocked_ledger.insert_token(auth_token, req.destination_address); ClientRequestProcessor::create_storage_dir( req.destination_address, - unlocked.store_dir.as_path(), + processing_data.store_dir.as_path(), )?; } Ok(RegisterResponse::new(auth_token)) diff --git a/src/provider/mod.rs b/src/provider/mod.rs index a5cb8a2dfb..bdbd337b29 100644 --- a/src/provider/mod.rs +++ b/src/provider/mod.rs @@ -74,6 +74,10 @@ impl ClientLedger { ClientLedger(HashMap::new()) } + fn add_arc_futures_mutex(self) -> Arc> { + Arc::new(FMutex::new(self)) + } + fn has_token(&self, auth_token: AuthToken) -> bool { return self.0.contains_key(&auth_token); } @@ -176,7 +180,7 @@ impl ServiceProvider { // TODO: FIGURE OUT HOW TO SET READ_DEADLINES IN TOKIO async fn process_client_socket_connection( mut socket: tokio::net::TcpStream, - processing_data: Arc>, + processing_data: Arc, ) { let mut buf = [0; 1024]; @@ -246,12 +250,12 @@ impl ServiceProvider { async fn start_client_listening( address: SocketAddr, store_dir: PathBuf, - client_ledger: ClientLedger, + client_ledger: Arc>, secret_key: Scalar, ) -> Result<(), ProviderError> { let mut listener = tokio::net::TcpListener::bind(address).await?; let processing_data = - ClientProcessingData::new(store_dir, client_ledger, secret_key).add_arc_futures_mutex(); + ClientProcessingData::new(store_dir, client_ledger, secret_key).add_arc(); loop { let (socket, _) = listener.accept().await?; @@ -272,6 +276,9 @@ impl ServiceProvider { let mut rt = Runtime::new()?; // let mut h = rt.handle(); + let initial_client_ledger = self.registered_clients_ledger; + let thread_shareable_ledger = initial_client_ledger.add_arc_futures_mutex(); + let presence_notifier = presence::Notifier::new( self.directory_server, self.mix_network_address.clone(), @@ -287,7 +294,7 @@ impl ServiceProvider { let client_future = rt.spawn(ServiceProvider::start_client_listening( self.client_network_address, self.store_dir.clone(), - self.registered_clients_ledger, // we're just cloning the initial ledger state + thread_shareable_ledger.clone(), self.secret_key, )); // Spawn the root task From 76f503a6641ea313ed2cff608a4e00a92276f819 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 19 Dec 2019 13:49:04 +0000 Subject: [PATCH 3/4] Provider sending registered clients with presence --- src/provider/mod.rs | 12 ++++++++++- src/provider/presence.rs | 44 +++++++++++++++++++++++++--------------- 2 files changed, 39 insertions(+), 17 deletions(-) diff --git a/src/provider/mod.rs b/src/provider/mod.rs index bdbd337b29..3b588fcb6d 100644 --- a/src/provider/mod.rs +++ b/src/provider/mod.rs @@ -5,6 +5,7 @@ use curve25519_dalek::montgomery::MontgomeryPoint; use curve25519_dalek::scalar::Scalar; use futures::io::Error; use futures::lock::Mutex as FMutex; +use nym_client::clients::directory::presence::MixProviderClient; use sfw_provider_requests::AuthToken; use sphinx::route::DestinationAddressBytes; use std::collections::HashMap; @@ -90,6 +91,14 @@ impl ClientLedger { self.0.insert(auth_token, client_address) } + fn current_clients(&self) -> Vec { + self.0 + .iter() + .map(|(_, &v)| Config::public_key_string(MontgomeryPoint(v))) + .map(|pub_key| MixProviderClient { pub_key }) + .collect() + } + #[allow(dead_code)] fn load(_file: PathBuf) -> Self { unimplemented!() @@ -283,6 +292,7 @@ impl ServiceProvider { self.directory_server, self.mix_network_address.clone(), self.public_key, + thread_shareable_ledger.clone(), ); let presence_future = rt.spawn(presence_notifier.run()); @@ -294,7 +304,7 @@ impl ServiceProvider { let client_future = rt.spawn(ServiceProvider::start_client_listening( self.client_network_address, self.store_dir.clone(), - thread_shareable_ledger.clone(), + thread_shareable_ledger, self.secret_key, )); // Spawn the root task diff --git a/src/provider/presence.rs b/src/provider/presence.rs index 9b08de810e..d1790b685a 100644 --- a/src/provider/presence.rs +++ b/src/provider/presence.rs @@ -1,17 +1,19 @@ -use crate::provider; -use nym_client::clients::directory; -use nym_client::clients::directory::DirectoryClient; -use std::time::Duration; - -use crate::provider::Config; +use crate::provider::{ClientLedger, Config}; use curve25519_dalek::montgomery::MontgomeryPoint; +use futures::lock::Mutex as FMutex; +use nym_client::clients::directory; use nym_client::clients::directory::presence::MixProviderPresence; use nym_client::clients::directory::requests::presence_providers_post::PresenceMixProviderPoster; +use nym_client::clients::directory::DirectoryClient; use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Duration; pub struct Notifier { pub net_client: directory::Client, - presence: MixProviderPresence, + client_ledger: Arc>, + host: String, + pub_key: String, } impl Notifier { @@ -19,32 +21,42 @@ impl Notifier { directory_server_address: String, host: SocketAddr, pub_key: MontgomeryPoint, + client_ledger: Arc>, ) -> Notifier { let directory_config = directory::Config { base_url: directory_server_address, }; let net_client = directory::Client::new(directory_config); - let presence = MixProviderPresence { - host: host.to_string(), - pub_key: Config::public_key_string(pub_key), - registered_clients: vec![], - }; + Notifier { net_client, - presence, + host: host.to_string(), + pub_key: Config::public_key_string(pub_key), + client_ledger, } } - pub fn notify(&self) { + async fn make_presence(&self) -> MixProviderPresence { + let unlocked_ledger = self.client_ledger.lock().await; + + MixProviderPresence { + host: self.host.clone(), + pub_key: self.pub_key.clone(), + registered_clients: unlocked_ledger.current_clients(), + } + } + + pub fn notify(&self, presence: MixProviderPresence) { self.net_client .presence_providers_post - .post(&self.presence) + .post(&presence) .unwrap(); } pub async fn run(self) { loop { - self.notify(); + let presence = self.make_presence().await; + self.notify(presence); let delay_duration = Duration::from_secs(5); tokio::time::delay_for(delay_duration).await; } From cbac85fbf4284f29ec7c66c9f9e98253bb837139 Mon Sep 17 00:00:00 2001 From: Jedrzej Stuczynski Date: Thu, 19 Dec 2019 13:49:45 +0000 Subject: [PATCH 4/4] Fixed compiler warnings --- src/main.rs | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/main.rs b/src/main.rs index 24309f7440..249b26fe14 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,11 +1,9 @@ -use crate::provider::presence; use crate::provider::ServiceProvider; use clap::{App, Arg, ArgMatches, SubCommand}; use nym_client::identity::mixnet::KeyPair; use std::net::ToSocketAddrs; use std::path::PathBuf; use std::process; -use std::thread; pub mod provider;