diff --git a/lez/wallet/src/lib.rs b/lez/wallet/src/lib.rs index ebe5028a..8fa65d78 100644 --- a/lez/wallet/src/lib.rs +++ b/lez/wallet/src/lib.rs @@ -87,6 +87,8 @@ pub enum ExecutionFailureKind { SignError(anyhow::Error), #[error("Sending transaction failed for each client")] MultiSequencerTransactionSendError, + #[error("Failed to join a task: {0}")] + JoinError(#[from] tokio::task::JoinError), } pub struct WalletCore { @@ -808,11 +810,7 @@ impl WalletCore { let call_res = self .multi_sequencer_client - .metered_send(async |client: &SequencerClient| { - client - .send_transaction(LeeTransaction::PrivacyPreserving(tx.clone())) - .await - }) + .metered_send_transaction(LeeTransaction::PrivacyPreserving(tx)) .await .into_iter() .find(std::result::Result::is_ok) @@ -877,17 +875,12 @@ impl WalletCore { let tx = lee::public_transaction::PublicTransaction::new(message, witness_set); - Ok(self - .multi_sequencer_client - .metered_send(async |client: &SequencerClient| { - client - .send_transaction(LeeTransaction::Public(tx.clone())) - .await - }) + self.multi_sequencer_client + .metered_send_transaction(LeeTransaction::Public(tx)) .await .into_iter() .find(std::result::Result::is_ok) - .ok_or(ExecutionFailureKind::MultiSequencerTransactionSendError)??) + .ok_or(ExecutionFailureKind::MultiSequencerTransactionSendError)? } pub async fn send_program_deployment_transaction(&self, bytecode: Vec) -> Result { @@ -896,11 +889,7 @@ impl WalletCore { Ok(self .multi_sequencer_client - .metered_send(async |client: &SequencerClient| { - client - .send_transaction(LeeTransaction::ProgramDeployment(transaction.clone())) - .await - }) + .metered_send_transaction(LeeTransaction::ProgramDeployment(transaction)) .await .into_iter() .find(std::result::Result::is_ok) diff --git a/lez/wallet/src/multi_client.rs b/lez/wallet/src/multi_client.rs index 0e7e2fef..3bb8a215 100644 --- a/lez/wallet/src/multi_client.rs +++ b/lez/wallet/src/multi_client.rs @@ -10,13 +10,17 @@ use std::{collections::HashMap, path::Path, sync::Arc}; use anyhow::{Context as _, Result}; +use common::{HashType, transaction::LeeTransaction}; use lee_core::BlockId; use sequencer_service_rpc::{RpcClient as _, SequencerClient, SequencerClientBuilder}; use serde::{Deserialize, Serialize}; -use tokio::sync::RwLock; +use tokio::{sync::RwLock, task::JoinSet}; use url::Url; -use crate::config::{MultiSequencerClientConfig, SequencerConnectionData}; +use crate::{ + ExecutionFailureKind, + config::{MultiSequencerClientConfig, SequencerConnectionData}, +}; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Statistics { @@ -131,13 +135,19 @@ impl MultiSequencerClient { // This actually runs in one thread, but eah exact future is parallelized. let (actualization_res, callibration_res) = tokio::join!( - multi_actualize_clients(&actualization_list), + multi_actualize_clients(actualization_list), multi_calibrate_clients( - &calibration_list, + calibration_list, multi_sequencer_client_config.calibration_limit ) ); + for (addr, statistic_opt) in callibration_res { + if let Some(statistic) = statistic_opt { + statistics.insert(addr, statistic); + } + } + for (addr, statistic_update_opt) in actualization_res { if let Some(statistic_mut) = statistics.get_mut(&addr) && let Some(statistic_update) = statistic_update_opt @@ -146,14 +156,8 @@ impl MultiSequencerClient { } } - for (addr, statistic_opt) in callibration_res { - if let Some(statistic) = statistic_opt { - statistics.insert(addr, statistic); - } - } - let leader_list = choose_leaders( - &client_list, + client_list, statistics, multi_sequencer_client_config.distribution_limit, ) @@ -165,7 +169,10 @@ impl MultiSequencerClient { ); } - log::info!("Chosen leaders is {leader_list:?}"); + log::info!( + "Chosen leaders is {:?}", + leader_list.iter().map(|(_, addr)| addr).collect::>() + ); Ok(leader_list) } @@ -194,7 +201,10 @@ impl MultiSequencerClient { ) -> Result<()> { let leader_list = Self::setup(conn_data, statistics, multi_sequencer_client_config).await?; - log::info!("Chosen leaders is {leader_list:#?}"); + log::info!( + "Chosen leaders is {:?}", + leader_list.iter().map(|(_, addr)| addr).collect::>() + ); self.leader_list = leader_list; @@ -288,32 +298,55 @@ impl MultiSequencerClient { resp } - /// Metered call for `distribution_limit` amount of leaders for sending data, usually - /// transaction. - pub async fn metered_send Result>( + /// Metered `send_transaction` for `distribution_limit` amount of leaders. + /// + /// Less abstract that it could be, clean way to implement it in a more general way is + /// "return-type notation". + /// + /// `ToDo`: Return to it, when "return-type notation" is stable. + pub async fn metered_send_transaction( &self, - call: I, - ) -> Vec> { - let leaders = self.leaders().iter().take(self.config().distribution_limit); - + tx: LeeTransaction, + ) -> Vec> { // Collecting all statistics into one map to lock updates only once let mut statistic_map: HashMap> = HashMap::new(); let mut results = vec![]; + let mut join_set = JoinSet::new(); - for (leader, leader_url) in leaders { - let (resp, statistics_update) = - tokio::join!(call(leader), actualize_client(leader.clone())); + for (leader, leader_url) in self.leaders() { + let curr_leader = leader.clone(); + let curr_tx = tx.clone(); + let curr_url = leader_url.clone(); - log::debug!( - "Metered call for {leader_url:?}, statistic updates is {statistics_update:?}", - ); + join_set.spawn(async move { + ( + tokio::join!( + curr_leader.send_transaction(curr_tx), + actualize_client(curr_leader.clone()) + ), + curr_url, + ) + }); + } - statistic_map - .entry(leader_url.clone()) - .or_default() - .push(statistics_update); - results.push(resp); + while let Some(resp) = join_set.join_next().await { + let res = resp + .map_err(Into::into) + .and_then(|((resp, statistics_update), leader_url)| { + log::debug!( + "Metered call for {leader_url:?}, statistic updates is {statistics_update:?}", + ); + + statistic_map + .entry(leader_url) + .or_default() + .push(statistics_update); + + resp.map_err(Into::into) + }); + + results.push(res); } { @@ -492,13 +525,14 @@ async fn actualize_client(client: SequencerClient) -> StatisticsUpdate { // ToDo: Return to it when "return-type notation" is stable pub async fn multi_actualize_clients( - clients: &[(Url, SequencerClient)], + clients: Vec<(Url, SequencerClient)>, ) -> Vec<(Url, Option)> { let mut handle_map = HashMap::new(); for (url, client) in clients { - let actualization_task = tokio::task::spawn(actualize_client(client.clone())); - handle_map.insert(url.clone(), actualization_task); + // `client` here must have 'static lifetime, so we can not use reference + let actualization_task = tokio::task::spawn(actualize_client(client)); + handle_map.insert(url, actualization_task); } let mut statistic_updates = vec![]; @@ -517,15 +551,15 @@ pub async fn multi_actualize_clients( } pub async fn multi_calibrate_clients( - clients: &[(Url, SequencerClient)], + clients: Vec<(Url, SequencerClient)>, calibration_limit: usize, ) -> Vec<(Url, Option)> { let mut handle_map = HashMap::new(); for (url, client) in clients { - let calibration_task = - tokio::task::spawn(calibrate_client(client.clone(), calibration_limit)); - handle_map.insert(url.clone(), calibration_task); + // `client` here must have 'static lifetime, so we can not use reference + let calibration_task = tokio::task::spawn(calibrate_client(client, calibration_limit)); + handle_map.insert(url, calibration_task); } let mut statistics = vec![]; @@ -559,13 +593,13 @@ pub async fn multi_calibrate_clients( /// /// Assumes that all clients have their statistics in `statistics`. fn choose_leaders( - client_vec: &[(Url, SequencerClient)], + client_vec: Vec<(Url, SequencerClient)>, statistics: &HashMap, distribution_limit: usize, ) -> Option> { // Sort out all unmetered clients let client_vec: Vec<_> = client_vec - .iter() + .into_iter() .filter(|item| statistics.contains_key(&item.0)) .collect(); @@ -594,10 +628,10 @@ fn choose_leaders( // Sort out all clients running late let mut res_vec: Vec<_> = client_vec .iter() - .filter_map(|x| { + .filter(|x| { let latest_block_id = statistics.get(&x.0).unwrap().latest_block_id; - (latest_block_id == max_block_id).then_some(*x) + latest_block_id == max_block_id }) .collect(); @@ -1026,7 +1060,7 @@ mod tests { }, ); - let leaders = choose_leaders(&client_list, &statistics, 1).unwrap(); + let leaders = choose_leaders(client_list, &statistics, 1).unwrap(); let (_, leader_url) = leaders.first().unwrap(); @@ -1083,7 +1117,7 @@ mod tests { }, ); - let leaders = choose_leaders(&client_list, &statistics, 1).unwrap(); + let leaders = choose_leaders(client_list, &statistics, 1).unwrap(); let (_, leader_url) = leaders.first().unwrap(); @@ -1140,7 +1174,7 @@ mod tests { }, ); - let leaders = choose_leaders(&client_list, &statistics, 1).unwrap(); + let leaders = choose_leaders(client_list, &statistics, 1).unwrap(); let (_, leader_url) = leaders.first().unwrap(); @@ -1197,7 +1231,7 @@ mod tests { }, ); - let leaders = choose_leaders(&client_list, &statistics, 1).unwrap(); + let leaders = choose_leaders(client_list, &statistics, 1).unwrap(); let (_, leader_url) = leaders.first().unwrap(); @@ -1254,7 +1288,7 @@ mod tests { }, ); - let leaders = choose_leaders(&client_list, &statistics, 2).unwrap(); + let leaders = choose_leaders(client_list, &statistics, 2).unwrap(); let mut url_set_origin = HashSet::new(); let mut url_set_res = HashSet::new(); @@ -1322,7 +1356,7 @@ mod tests { }, ); - let leaders = choose_leaders(&client_list, &statistics, 2).unwrap(); + let leaders = choose_leaders(client_list, &statistics, 2).unwrap(); let (_, helm) = leaders.first().unwrap(); @@ -1380,7 +1414,7 @@ mod tests { }, ); - let leaders = choose_leaders(&client_list, &statistics, 2).unwrap(); + let leaders = choose_leaders(client_list, &statistics, 2).unwrap(); let (_, leader_url_first) = leaders[0].clone(); let (_, leader_url_second) = leaders[1].clone(); @@ -1440,7 +1474,7 @@ mod tests { }, ); - let leaders = choose_leaders(&client_list, &statistics, 2).unwrap(); + let leaders = choose_leaders(client_list, &statistics, 2).unwrap(); let (_, leader_url_first) = leaders[0].clone(); let (_, leader_url_second) = leaders[1].clone(); @@ -1500,7 +1534,7 @@ mod tests { }, ); - let leaders = choose_leaders(&client_list, &statistics, 2).unwrap(); + let leaders = choose_leaders(client_list, &statistics, 2).unwrap(); let (_, leader_url_first) = leaders[0].clone(); let (_, leader_url_second) = leaders[1].clone();