Skip to content

improve_perf_test #188

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mubarizkyc wants to merge 1 commit into
base: main
Choose a base branch
from
+169 −142
Open

improve_perf_test #188

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
311 changes: 169 additions & 142 deletions chitchat/tests/perf_test.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,56 +1,66 @@
use std::collections::BTreeMap;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;

use chitchat::transport::{ChannelTransport, Transport, TransportExt};
use chitchat::{
ChitchatConfig, ChitchatHandle, ChitchatId, FailureDetectorConfig, NodeState, spawn_chitchat,
transport::{ChannelTransport, Transport, TransportExt},
};
use std::{collections::BTreeMap, net::SocketAddr, sync::Arc, time::Duration};
use tokio::time::Instant;
use tokio_stream::StreamExt;
use tracing::info;

async fn spawn_one(chitchat_id: u16, transport: &dyn Transport) -> ChitchatHandle {
let listen_addr: SocketAddr = ([127, 0, 0, 1], 10_000u16 + chitchat_id).into();
let chitchat_id = ChitchatId {
node_id: Arc::from(format!("node_{chitchat_id}").as_str()),
generation_id: 0,
gossip_advertise_addr: listen_addr,
};
let gossip_interval = Duration::from_millis(300);
let config = ChitchatConfig {
chitchat_id,
cluster_id: "default-cluster".to_string(),
gossip_interval,
listen_addr,
seed_nodes: vec!["127.0.0.1:10000".to_string()],
failure_detector_config: FailureDetectorConfig {
initial_interval: gossip_interval,
..Default::default()
},
marked_for_deletion_grace_period: Duration::from_secs(10_000),
catchup_callback: None,
extra_liveness_predicate: None,
};
spawn_chitchat(config, Vec::new(), transport).await.unwrap()
const DEFAULT_MTU: usize = 65_507;
const DETECTION_TIMEOUT_SECS: u64 = 10;
const FAULTY_DETECTION_TIMEOUT_SECS: u64 = 15;
const PACKET_LOSS_RATE: f64 = 0.5;
const MEASUREMENT_FRAME_SECS: u64 = 3;
const MAX_MESSAGES_PER_SEC_PER_NODE: u64 = 50;

fn init_tracing() {
let _ = tracing_subscriber::fmt::try_init();
}

/// Spawn "num_nodes" chitchat nodes
async fn spawn_nodes(num_nodes: u16, transport: &dyn Transport) -> Vec<ChitchatHandle> {
let mut handles = Vec::new();
let mut handles: Vec<ChitchatHandle> = Vec::new();
// per-node config
let gossip_interval = Duration::from_millis(300);
let cluster_id: String = "default-cluster".to_string();

for chitchat_id in 0..num_nodes {
let handle = spawn_one(chitchat_id, transport).await;
handles.push(handle);
let listen_addr: SocketAddr = ([127, 0, 0, 1], 10_000u16 + chitchat_id).into();
let config = ChitchatConfig {
chitchat_id: ChitchatId {
node_id: Arc::from(format!("node_{chitchat_id}").as_str()),
generation_id: 0,
gossip_advertise_addr: listen_addr,
},
cluster_id: cluster_id.clone(),
gossip_interval,
listen_addr,
seed_nodes: vec!["127.0.0.1:10000".to_string()],
failure_detector_config: FailureDetectorConfig {
initial_interval: gossip_interval,
..Default::default()
},
marked_for_deletion_grace_period: Duration::from_secs(10_000),
catchup_callback: None,
extra_liveness_predicate: None,
};
let spawned_node = spawn_chitchat(config, Vec::new(), transport).await.unwrap();
handles.push(spawned_node);
}
handles
}

async fn wait_until<P: Fn(&BTreeMap<ChitchatId, NodeState>) -> bool>(
handle: &ChitchatHandle,
node_handle: &ChitchatHandle,
predicate: P,
) -> Duration {
let start = Instant::now();
let mut node_watcher = handle.chitchat().lock().await.live_nodes_watch_stream();
let mut node_watcher = node_handle
.chitchat()
.lock()
.await
.live_nodes_watch_stream();
while let Some(nodes) = node_watcher.next().await {
if predicate(&nodes) {
break;
Expand All @@ -61,141 +71,160 @@ async fn wait_until<P: Fn(&BTreeMap<ChitchatId, NodeState>) -> bool>(

async fn delay_before_detection_sample(num_nodes: usize, transport: &dyn Transport) -> Duration {
assert!(num_nodes > 2);
let mut handles = spawn_nodes(num_nodes as u16, transport).await;
// spawned n nodes
let mut node_handles = spawn_nodes(num_nodes as u16, transport).await;
info!("spawn {num_nodes} nodes");
let _delay = wait_until(&handles[1], |nodes| nodes.len() == num_nodes).await;

// wait until all nodes are visible to node 1
wait_until(&node_handles[1], |nodes| nodes.len() == num_nodes).await;
info!("converged on {num_nodes} nodes");
handles.pop();

// one node died
node_handles.pop();

// time to detect failure (or to realize that we have n-1 nodes now)
let time_to_death_detection =
wait_until(&handles[1], |nodes| nodes.len() == num_nodes - 1).await;
for handle in handles {
wait_until(&node_handles[1], |nodes| nodes.len() == num_nodes - 1).await;

for handle in node_handles {
handle.shutdown().await.unwrap();
}
info!(time_to_death_detection=?time_to_death_detection);
time_to_death_detection
}

#[tokio::test]
async fn test_delay_before_dead_detection_10() {
// let _ = tracing_subscriber::fmt::try_init();
let transport = ChannelTransport::with_mtu(65_507);
let delay = delay_before_detection_sample(40, &transport).await;
assert!(
delay < Duration::from_secs(10),
"Delay exceeded: {:?}",
delay
);
}

#[tokio::test]
async fn test_delay_before_dead_detection_20() {
// let _ = tracing_subscriber::fmt::try_init();
let transport = ChannelTransport::with_mtu(65_507);
let delay = delay_before_detection_sample(20, &transport).await;
assert!(
delay < Duration::from_secs(10),
"Delay exceeded: {:?}",
delay
);
}

#[tokio::test]
async fn test_delay_before_dead_detection_40() {
// let _ = tracing_subscriber::fmt::try_init();
let transport = ChannelTransport::with_mtu(65_507);
let delay = delay_before_detection_sample(40, &transport).await;
assert!(
delay < Duration::from_secs(10),
"Delay exceeded: {:?}",
delay
);
}

#[tokio::test]
async fn test_delay_before_dead_detection_100() {
// let _ = tracing_subscriber::fmt::try_init();
let transport = ChannelTransport::with_mtu(65_507);
let delay = delay_before_detection_sample(100, &transport).await;
assert!(
delay < Duration::from_secs(10),
"Delay exceeded: {:?}",
delay
);
macro_rules! death_detection_test {
($name:ident, $num_nodes:expr, $timeout_secs:expr) => {
#[tokio::test]
async fn $name() {
init_tracing();
let transport = ChannelTransport::with_mtu(DEFAULT_MTU);
let delay = delay_before_detection_sample($num_nodes, &transport).await;
assert!(
delay < Duration::from_secs($timeout_secs),
"Delay Exceeded {:?}",
delay,
)
}
};
}

#[tokio::test]
async fn test_delay_before_dead_detection_100_faulty() {
let _ = tracing_subscriber::fmt::try_init();
let transport = ChannelTransport::with_mtu(65_507).drop_message(0.5f64);
let delay = delay_before_detection_sample(100, &*transport).await;
assert!(
delay < Duration::from_secs(15),
"Delay exceeded: {:?}",
delay
);
macro_rules! death_detection_test_faulty {
($name:ident, $num_nodes:expr, $timeout_secs:expr, $loss_rate:expr) => {
#[tokio::test]
async fn $name() {
init_tracing();
let transport = ChannelTransport::with_mtu(DEFAULT_MTU).drop_message($loss_rate);
let delay = delay_before_detection_sample($num_nodes, &*transport).await;
assert!(
delay < Duration::from_secs($timeout_secs),
"Delay Exceeded {:?}",
delay,
)
}
};
}

async fn test_bandwidth_aux(num_nodes: usize) -> u64 {
let _ = tracing_subscriber::fmt::try_init();
death_detection_test!(
test_delay_before_dead_detection_10,
10,
DETECTION_TIMEOUT_SECS
);
death_detection_test!(
test_delay_before_dead_detection_20,
20,
DETECTION_TIMEOUT_SECS
);
death_detection_test!(
test_delay_before_dead_detection_40,
40,
DETECTION_TIMEOUT_SECS
);
death_detection_test!(
test_delay_before_dead_detection_100,
100,
DETECTION_TIMEOUT_SECS
);
death_detection_test_faulty!(
test_delay_before_dead_detection_100_faulty,
100,
FAULTY_DETECTION_TIMEOUT_SECS,
PACKET_LOSS_RATE
);

struct BandwidthMetrics {
bytes_per_sec_per_node: u64,
messages_per_sec_per_node: u64,
}

async fn measure_bandwidth(num_nodes: usize) -> BandwidthMetrics {
assert!(num_nodes > 2);
let transport = ChannelTransport::with_mtu(65_507);
let transport = ChannelTransport::with_mtu(DEFAULT_MTU);
let handles = spawn_nodes(num_nodes as u16, &transport).await;
let instant = Instant::now();
for handle in &handles {
wait_until(handle, |nodes| nodes.len() == num_nodes).await;
info!("success");
let start = Instant::now();

for node_handle in &handles {
// wait for convergence
wait_until(node_handle, |nodes| nodes.len() == num_nodes).await;
}
let cluster_convergence = instant.elapsed();
info!(cluster_convergence=?cluster_convergence);
const MEASUREMENT_FRAME_SECS: u64 = 3;
let stat_before = transport.statistics();
let convergence_time = start.elapsed();
info!(?convergence_time, "Cluster converged");

let stats_before = transport.statistics();
tokio::time::sleep(Duration::from_secs(MEASUREMENT_FRAME_SECS)).await;
let stat_after = transport.statistics();
let bytes_per_sec_per_node = (stat_after.num_bytes_total - stat_before.num_bytes_total)
/ (num_nodes as u64 * MEASUREMENT_FRAME_SECS);
let num_messages_per_sec_per_node = (stat_after.num_messages_total
- stat_before.num_messages_total)
/ (num_nodes as u64 * MEASUREMENT_FRAME_SECS);
info!(num_messages_per_sec_per_node=num_messages_per_sec_per_node, bandwidth_per_node=%bytes_per_sec_per_node);
assert!(num_messages_per_sec_per_node < 50);
bytes_per_sec_per_node
}
let stats_after = transport.statistics();

#[tokio::test]
async fn test_bandwidth_10() {
let _ = tracing_subscriber::fmt::try_init();
assert!(test_bandwidth_aux(10).await < 15_000);
}
let total_bytes = stats_after.num_bytes_total - stats_before.num_bytes_total;
let total_messages = stats_after.num_messages_total - stats_before.num_messages_total;

#[tokio::test]
async fn test_bandwidth_20() {
let _ = tracing_subscriber::fmt::try_init();
test_bandwidth_aux(20).await;
assert!(test_bandwidth_aux(20).await < 25_000);
BandwidthMetrics {
bytes_per_sec_per_node: total_bytes / (num_nodes as u64 * MEASUREMENT_FRAME_SECS),
messages_per_sec_per_node: total_messages / (num_nodes as u64 * MEASUREMENT_FRAME_SECS),
}
}

#[tokio::test]
async fn test_bandwidth_40() {
let _ = tracing_subscriber::fmt::try_init();
assert!(test_bandwidth_aux(40).await < 50_000);
macro_rules! bandwidth_test {
($name:ident, $num_nodes:expr, $max_bytes_per_sec:expr) => {
#[tokio::test]
async fn $name() {
init_tracing();
let metrics = measure_bandwidth($num_nodes).await;

assert!(
metrics.messages_per_sec_per_node < MAX_MESSAGES_PER_SEC_PER_NODE,
"messages rate too high: {} msg's per node (limit:{} msg's/node) for {} nodes",
metrics.messages_per_sec_per_node,
MAX_MESSAGES_PER_SEC_PER_NODE,
$num_nodes
);
assert!(
metrics.bytes_per_sec_per_node < $max_bytes_per_sec,
"bandwidth too high: {} byte's per node (limit:{} bytes's/node) for {} nodes",
metrics.bytes_per_sec_per_node,
$max_bytes_per_sec,
$num_nodes
);
}
};
}

#[tokio::test]
async fn test_bandwidth_100() {
let _ = tracing_subscriber::fmt::try_init();
assert!(test_bandwidth_aux(100).await < 120_000);
}
bandwidth_test!(test_bandwidth_10, 10, 15_000);
bandwidth_test!(test_bandwidth_20, 20, 25_000);
bandwidth_test!(test_bandwidth_40, 40, 50_000);
bandwidth_test!(test_bandwidth_100, 100, 120_000);

async fn test_faulty_network_stability_aux(num_nodes: usize, transport: &dyn Transport) {
// 50% messages are dropped.
assert!(num_nodes > 2);
let handles = spawn_nodes(num_nodes as u16, transport).await;
let start = Instant::now();

for handle in &handles {
wait_until(handle, |nodes| nodes.len() == num_nodes).await;
}
let elapsed = start.elapsed();
info!("Convergence took {elapsed:?}");

let lost_one_node = wait_until(&handles[1], |nodes| nodes.len() != num_nodes);
// We want this to timeout!
tokio::time::timeout(Duration::from_secs(30), lost_one_node)
Expand All @@ -205,18 +234,16 @@ async fn test_faulty_network_stability_aux(num_nodes: usize, transport: &dyn Tra

#[tokio::test]
async fn test_faulty_network_stability_10() {
let _ = tracing_subscriber::fmt::try_init();
init_tracing();
// 50% messages are dropped.
use chitchat::transport::TransportExt;
let transport: Box<dyn Transport> = ChannelTransport::with_mtu(65_507).drop_message(0.5f64);
let transport = ChannelTransport::with_mtu(DEFAULT_MTU).drop_message(PACKET_LOSS_RATE);
test_faulty_network_stability_aux(10, &*transport).await;
}

#[tokio::test]
async fn test_faulty_network_stability_100() {
let _ = tracing_subscriber::fmt::try_init();
init_tracing();
// 50% messages are dropped.
use chitchat::transport::TransportExt;
let transport: Box<dyn Transport> = ChannelTransport::with_mtu(65_507).drop_message(0.5f64);
test_faulty_network_stability_aux(10, &*transport).await;
let transport = ChannelTransport::with_mtu(DEFAULT_MTU).drop_message(PACKET_LOSS_RATE);
test_faulty_network_stability_aux(100, &*transport).await;
}