logos-blockchain/logos-execution-zone
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-execution-zone.git synced 2026-07-30 17:43:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(wallet): more parallelization

Browse Source
This commit is contained in:
Pravdyvy 2026-07-23 14:56:18 +03:00
parent 89240ae666
commit d34d269350
1 changed files with 150 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

252
lez/wallet/src/multi_client.rs
View File

@ -13,7 +13,7 @@ use anyhow::{Context as _, Result};
use lee_core::BlockId;
use sequencer_service_rpc::{RpcClient as _, SequencerClient, SequencerClientBuilder};
use serde::{Deserialize, Serialize};
use tokio::{sync::RwLock, task::JoinSet};
use tokio::sync::RwLock;
use url::Url;
use crate::config::{MultiSequencerClientConfig, SequencerConnectionData};
@ -93,7 +93,9 @@ impl MultiSequencerClient {
statistics: &mut HashMap<Url, Statistics>,
multi_sequencer_client_config: &MultiSequencerClientConfig,
) -> Result<Vec<(SequencerClient, Url)>> {
let mut client_list = HashMap::new();
let mut actualization_list = vec![];
let mut calibration_list = vec![];
let mut client_list = vec![];
for SequencerConnectionData {
sequencer_addr,
@ -118,30 +120,35 @@ impl MultiSequencerClient {
.context("Failed to create sequencer client")?
};
// If there is statistics for client, actualize it
if let Some(statistic_mut) = statistics.get_mut(sequencer_addr) {
let statistic_updates = actualize_client(&sequencer_client).await;
log::debug!(
"Metered call for {sequencer_addr:?}, statistic updates is {statistic_updates:?}"
);
statistic_mut.apply_updates(&[statistic_updates]);
client_list.insert(sequencer_addr.clone(), sequencer_client);
// Otherwise calibrate client data
} else if let Some(client_statistics) = calibrate_client(
&sequencer_client,
multi_sequencer_client_config.calibration_limit,
)
.await
{
statistics.insert(sequencer_addr.clone(), client_statistics);
client_list.insert(sequencer_addr.clone(), sequencer_client);
// There is no point in adding uncalibrated client
if statistics.contains_key(sequencer_addr) {
actualization_list.push((sequencer_addr.clone(), sequencer_client.clone()));
} else {
log::warn!("Client {sequencer_addr:?} failed all {} calibration attempts, it may be unhealthy.
\n Consider bumping calibration_limit or remove this client altogether", multi_sequencer_client_config.calibration_limit);
calibration_list.push((sequencer_addr.clone(), sequencer_client.clone()));
}
client_list.push((sequencer_addr.clone(), sequencer_client));
}
// 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_calibrate_clients(
&calibration_list,
multi_sequencer_client_config.calibration_limit
)
);
for (addr, statistic_update_opt) in actualization_res {
if let Some(statistic_mut) = statistics.get_mut(&addr)
&& let Some(statistic_update) = statistic_update_opt
{
statistic_mut.apply_updates(&[statistic_update]);
}
}
for (addr, statistic_opt) in callibration_res {
if let Some(statistic) = statistic_opt {
statistics.insert(addr, statistic);
}
}
@ -215,7 +222,8 @@ impl MultiSequencerClient {
leader_url: &Url,
statistic_map: &mut HashMap<Url, Vec<StatisticsUpdate>>,
) -> Result<R, E> {
let (resp, statistics_update) = tokio::join!(call(leader), actualize_client(leader));
let (resp, statistics_update) =
tokio::join!(call(leader), actualize_client(leader.clone()));
log::debug!("Metered call for {leader_url:?}, statistic updates is {statistics_update:?}",);
@ -290,7 +298,8 @@ impl MultiSequencerClient {
let mut results = vec![];
for (leader, leader_url) in leaders {
let (resp, statistics_update) = tokio::join!(call(leader), actualize_client(leader));
let (resp, statistics_update) =
tokio::join!(call(leader), actualize_client(leader.clone()));
log::debug!(
"Metered call for {leader_url:?}, statistic updates is {statistics_update:?}",
@ -414,17 +423,16 @@ async fn measure_request_duration(client: &SequencerClient) -> (u128, Option<Blo
)
}
pub async fn calibrate_client(
client: &SequencerClient,
calibration_limit: usize,
) -> Option<Statistics> {
/// Calibrate statistics for one client. Takes `client` by value deliberately, cloning
/// `SequencerClient` is cheap.
async fn calibrate_client(client: SequencerClient, calibration_limit: usize) -> Option<Statistics> {
let mut latencies = vec![];
let mut latest_block_id = 0;
let mut errors: u64 = 0;
// ToDo: Add some DDoS adaptation
for _ in 0..calibration_limit {
let (latency, block_id) = measure_request_duration(client).await;
let (latency, block_id) = measure_request_duration(&client).await;
let Some(block_id) = block_id else {
errors = errors.saturating_add(1);
@ -458,21 +466,9 @@ pub async fn calibrate_client(
})
}
async fn actualize_client(client: &SequencerClient) -> StatisticsUpdate {
let (latency, block_id) = measure_request_duration(client).await;
#[expect(clippy::as_conversions, reason = "int to float conversion is safe")]
let latency = latency as f32;
block_id.map_or(StatisticsUpdate::Failure, |new_latest_block_id| {
StatisticsUpdate::Success {
latency,
new_latest_block_id,
}
})
}
async fn actualize_client_owned(client: SequencerClient) -> StatisticsUpdate {
/// Actualize statistics for one client. Takes `client` by value deliberately, cloning
/// `SequencerClient` is cheap.
async fn actualize_client(client: SequencerClient) -> StatisticsUpdate {
let (latency, block_id) = measure_request_duration(&client).await;
#[expect(clippy::as_conversions, reason = "int to float conversion is safe")]
@ -486,32 +482,85 @@ async fn actualize_client_owned(client: SequencerClient) -> StatisticsUpdate {
})
}
pub async fn multi_actualize_clients(clients: &[SequencerClient]) -> Vec<StatisticsUpdate> {
let mut join_set = JoinSet::new();
// Next 2 functions can be done in uniform way right now, but it will be incredibly cursed.
// ToDo: Return to it when "return-type notation" is stable
for client in clients {
join_set.spawn(actualize_client_owned(client.clone()));
pub async fn multi_actualize_clients(
clients: &[(Url, SequencerClient)],
) -> Vec<(Url, Option<StatisticsUpdate>)> {
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);
}
let mut statistic_updates = vec![];
while let Some(statistic_resp) = join_set.join_next().await {
statistic_updates.push(statistic_resp.unwrap_or(StatisticsUpdate::Failure));
// We anyhow need results of each one
#[expect(
clippy::iter_over_hash_type,
reason = "Ordering of map updates is not important"
)]
for (url, task) in handle_map {
let task_opt = task.await.ok();
statistic_updates.push((url, task_opt));
}
statistic_updates
}
pub async fn multi_calibrate_clients(
clients: &[(Url, SequencerClient)],
calibration_limit: usize,
) -> Vec<(Url, Option<Statistics>)> {
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);
}
let mut statistics = vec![];
// We anyhow need results of each one
#[expect(
clippy::iter_over_hash_type,
reason = "Ordering of map updates is not important"
)]
for (url, task) in handle_map {
let task_opt = task
.await
.ok()
.inspect(|val| {
if val.is_none() {
log::warn!(
"Client {url:?} failed all {calibration_limit} calibration attempts, it may be unhealthy.
\n Consider bumping calibration_limit or remove this client altogether"
);
}
})
.flatten();
statistics.push((url, task_opt));
}
statistics
}
#[must_use]
pub fn choose_leaders(
client_list: &HashMap<Url, SequencerClient>,
/// Choosing leaders up to a `distribution_limit`.
///
/// Assumes that all clients have their statistics in `statistics`.
fn choose_leaders(
client_vec: &[(Url, SequencerClient)],
statistics: &HashMap<Url, Statistics>,
distribution_limit: usize,
) -> Option<Vec<(SequencerClient, Url)>> {
// Sort out all unmetered clients
let mut client_vec: Vec<_> = client_list
.keys()
.filter(|item| statistics.contains_key(*item))
let client_vec: Vec<_> = client_vec
.iter()
.filter(|item| statistics.contains_key(&item.0))
.collect();
if client_vec.is_empty() {
@ -519,23 +568,28 @@ pub fn choose_leaders(
}
// Considering the nature of our requests, the latest_block_id is the dominant characteristic
let max_block_id_addr = client_vec.iter().fold(client_vec[0], |acc, x| {
let old_latest_block_id = statistics.get(acc).unwrap().latest_block_id;
let new_latest_block_id = statistics.get(*x).unwrap().latest_block_id;
if new_latest_block_id > old_latest_block_id {
*x
} else {
acc
}
});
let max_block_id_addr = client_vec
.iter()
.fold(client_vec.first().unwrap(), |acc, x| {
let old_latest_block_id = statistics.get(&acc.0).unwrap().latest_block_id;
let new_latest_block_id = statistics.get(&x.0).unwrap().latest_block_id;
if new_latest_block_id > old_latest_block_id {
x
} else {
acc
}
});
let max_block_id = statistics.get(max_block_id_addr).unwrap().latest_block_id;
let max_block_id = statistics
.get(&max_block_id_addr.0)
.unwrap()
.latest_block_id;
// Sort out all clients running late
client_vec = client_vec
let mut res_vec: Vec<_> = client_vec
.iter()
.filter_map(|x| {
let latest_block_id = statistics.get(*x).unwrap().latest_block_id;
let latest_block_id = statistics.get(&x.0).unwrap().latest_block_id;
(latest_block_id == max_block_id).then_some(*x)
})
@ -543,21 +597,21 @@ pub fn choose_leaders(
// Get the clients with lesser or equal to average error ratio
#[expect(clippy::as_conversions, reason = "int to float conversion is safe")]
let avg_err_ratio = client_vec.iter().fold(0_f32, |acc, x| {
let avg_err_ratio = res_vec.iter().fold(0_f32, |acc, x| {
acc + error_ratio(
statistics.get(*x).unwrap().errors,
statistics.get(*x).unwrap().sample_size,
statistics.get(&x.0).unwrap().errors,
statistics.get(&x.0).unwrap().sample_size,
)
}) / (client_vec.len() as f32);
}) / (res_vec.len() as f32);
client_vec.sort_by(|a, b| {
res_vec.sort_by(|a, b| {
let err_ratio_a = error_ratio(
statistics.get(*a).unwrap().errors,
statistics.get(*a).unwrap().sample_size,
statistics.get(&a.0).unwrap().errors,
statistics.get(&a.0).unwrap().sample_size,
);
let err_ratio_b = error_ratio(
statistics.get(*b).unwrap().errors,
statistics.get(*b).unwrap().sample_size,
statistics.get(&b.0).unwrap().errors,
statistics.get(&b.0).unwrap().sample_size,
);
err_ratio_a
@ -565,22 +619,22 @@ pub fn choose_leaders(
.expect("Ratios must be a valid numbers")
});
let mut client_vec = client_vec[..(client_vec
res_vec = res_vec[..(res_vec
.iter()
.position(|item| {
error_ratio(
statistics.get(*item).unwrap().errors,
statistics.get(*item).unwrap().sample_size,
statistics.get(&item.0).unwrap().errors,
statistics.get(&item.0).unwrap().sample_size,
) > avg_err_ratio
})
.unwrap_or(client_vec.len()))]
.unwrap_or(res_vec.len()))]
.to_vec();
// Choose clients with least latency and variance
client_vec.sort_by(|a, b| {
let left = statistics.get(*a).unwrap();
res_vec.sort_by(|a, b| {
let left = statistics.get(&a.0).unwrap();
let (left_lat, left_var) = (left.latency_avg, left.latency_var);
let right = statistics.get(*b).unwrap();
let right = statistics.get(&b.0).unwrap();
let (right_lat, right_var) = (right.latency_avg, right.latency_var);
let right_std = right_var.sqrt();
@ -605,16 +659,10 @@ pub fn choose_leaders(
});
Some(
client_vec
.iter()
res_vec
.into_iter()
.map(|(a, b)| (b.clone(), a.clone()))
.take(distribution_limit)
.map(|addr| {
let client = client_list
.get(*addr)
.expect("Missing clients already sorted out");
(client.clone(), (*addr).clone())
})
.collect(),
)
}
@ -721,7 +769,7 @@ mod tests {
builder.build(url).unwrap()
}
fn four_client_list() -> (HashMap<Url, SequencerClient>, [Url; 4]) {
fn four_client_list() -> (Vec<(Url, SequencerClient)>, [Url; 4]) {
let addr_leader = Url::parse("http://127.0.0.1:3040").unwrap();
let addr_1 = Url::parse("http://127.0.0.1:3041").unwrap();
let addr_2 = Url::parse("http://127.0.0.1:3042").unwrap();
@ -732,12 +780,12 @@ mod tests {
let client_2 = client_from_url_unchecked(&addr_2);
let client_3 = client_from_url_unchecked(&addr_3);
let mut client_list = HashMap::new();
client_list.insert(addr_leader.clone(), leader);
client_list.insert(addr_1.clone(), client_1);
client_list.insert(addr_2.clone(), client_2);
client_list.insert(addr_3.clone(), client_3);
let client_list = vec![
(addr_leader.clone(), leader),
(addr_1.clone(), client_1),
(addr_2.clone(), client_2),
(addr_3.clone(), client_3),
];
(client_list, [addr_leader, addr_1, addr_2, addr_3])
}