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