[Forge] Add a netbench test (#10051)

### Description

Create netbench tests across a pair of nodes, one colocated and the other emulated on a high-latency (and low bandwidth) network.

For now there is no success criteria, but we should be able to add one based on throughput when needed.

### Test Plan

Run the tests in forge.

config/src/config/mod.rs

@@ -44,6 +44,7 @@ pub use indexer_grpc_config::*;
 pub use inspection_service_config::*;
 pub use logger_config::*;
 pub use mempool_config::*;
+pub use netbench::*;
 pub use network_config::*;
 pub use node_config::*;
 pub use node_config_loader::sanitize_node_config;

testsuite/forge-cli/src/main.rs

@@ -6,8 +6,8 @@
 
 use anyhow::{format_err, Context, Result};
 use aptos_config::config::{
-    BootstrappingMode, ConsensusConfig, ContinuousSyncingMode, MempoolConfig, NodeConfig,
-    StateSyncConfig,
+    BootstrappingMode, ConsensusConfig, ContinuousSyncingMode, MempoolConfig, NetbenchConfig,
+    NodeConfig, StateSyncConfig,
 };
 use aptos_forge::{
     args::TransactionTypeArg,
@@ -30,7 +30,9 @@ use aptos_testcases::{
     generate_traffic,
     load_vs_perf_benchmark::{LoadVsPerfBenchmark, TransactionWorkload, Workloads},
     modifiers::{CpuChaosTest, ExecutionDelayConfig, ExecutionDelayTest},
-    multi_region_network_test::MultiRegionNetworkEmulationTest,
+    multi_region_network_test::{
+        MultiRegionNetworkEmulationConfig, MultiRegionNetworkEmulationTest,
+    },
     network_bandwidth_test::NetworkBandwidthTest,
     network_loss_test::NetworkLossTest,
     network_partition_test::NetworkPartitionTest,
@@ -585,6 +587,8 @@ fn single_test_suite(
         "pfn_performance_with_network_chaos" => pfn_performance(duration, false, true),
         "pfn_performance_with_realistic_env" => pfn_performance(duration, true, true),
         "gather_metrics" => gather_metrics(),
+        "net_bench" => net_bench(),
+        "net_bench_two_region_env" => net_bench_two_region_env(),
         _ => return Err(format_err!("Invalid --suite given: {:?}", test_name)),
     };
     Ok(single_test_suite)
@@ -627,6 +631,15 @@ fn state_sync_config_fast_sync(state_sync_config: &mut StateSyncConfig) {
         ContinuousSyncingMode::ApplyTransactionOutputs;
 }
 
+fn wrap_with_two_region_env<T: NetworkTest + 'static>(test: T) -> CompositeNetworkTest {
+    CompositeNetworkTest::new(
+        MultiRegionNetworkEmulationTest::new_with_config(
+            MultiRegionNetworkEmulationConfig::two_region(),
+        ),
+        test,
+    )
+}
+
 fn run_consensus_only_realistic_env_max_tps() -> ForgeConfig {
     ForgeConfig::default()
         .with_initial_validator_count(NonZeroUsize::new(20).unwrap())
@@ -1346,6 +1359,45 @@ fn gather_metrics() -> ForgeConfig {
         .add_network_test(GatherMetrics)
 }
 
+fn netbench_config_100_megabytes_per_sec(netbench_config: &mut NetbenchConfig) {
+    netbench_config.enabled = true;
+    netbench_config.max_network_channel_size = 1000;
+    netbench_config.enable_direct_send_testing = true;
+    netbench_config.direct_send_data_size = 100000;
+    netbench_config.direct_send_per_second = 1000;
+}
+
+fn netbench_config_2_megabytes_per_sec(netbench_config: &mut NetbenchConfig) {
+    netbench_config.enabled = true;
+    netbench_config.max_network_channel_size = 1000;
+    netbench_config.enable_direct_send_testing = true;
+    netbench_config.direct_send_data_size = 100000;
+    netbench_config.direct_send_per_second = 20;
+}
+
+fn net_bench() -> ForgeConfig {
+    ForgeConfig::default()
+        .add_network_test(Delay::new(180))
+        .with_initial_validator_count(NonZeroUsize::new(2).unwrap())
+        .with_validator_override_node_config_fn(Arc::new(|config, _| {
+            let mut netbench_config = NetbenchConfig::default();
+            netbench_config_100_megabytes_per_sec(&mut netbench_config);
+            config.netbench = Some(netbench_config);
+        }))
+}
+
+fn net_bench_two_region_env() -> ForgeConfig {
+    ForgeConfig::default()
+        .add_network_test(wrap_with_two_region_env(Delay::new(180)))
+        .with_initial_validator_count(NonZeroUsize::new(2).unwrap())
+        .with_validator_override_node_config_fn(Arc::new(|config, _| {
+            // Not using 100 MBps here, as it will lead to throughput collapse
+            let mut netbench_config = NetbenchConfig::default();
+            netbench_config_2_megabytes_per_sec(&mut netbench_config);
+            config.netbench = Some(netbench_config);
+        }))
+}
+
 fn three_region_simulation_with_different_node_speed() -> ForgeConfig {
     ForgeConfig::default()
         .with_initial_validator_count(NonZeroUsize::new(30).unwrap())

testsuite/testcases/src/data/two_region_link_stats.csv

@@ -0,0 +1,3 @@
+sending_region,receiving_region,bitrate_bps,avgrtt
+aws--sa-east-1,aws--ap-northeast-1,32505856,255.289
+aws--ap-northeast-1,aws--sa-east-1,32243712,255.323

testsuite/testcases/src/multi_region_network_test.rs

@@ -11,17 +11,14 @@ use std::collections::BTreeMap;
 /// The link stats are obtained from https://github.com/doitintl/intercloud-throughput/blob/master/results_202202/results.csv
 /// The four regions were hand-picked from the dataset to simulate a multi-region setup
 /// with high latencies and low bandwidth.
-macro_rules! FOUR_REGION_LINK_STATS_CSV {
-    () => {
-        "data/four_region_link_stats.csv"
-    };
-}
+const FOUR_REGION_LINK_STATS: &[u8] = include_bytes!("data/four_region_link_stats.csv");
+/// The two regions were chosen as the most distant regions among the four regions set.
+const TWO_REGION_LINK_STATS: &[u8] = include_bytes!("data/two_region_link_stats.csv");
 
-fn get_link_stats_table() -> BTreeMap<String, BTreeMap<String, (u64, f64)>> {
+fn get_link_stats_table(csv: &[u8]) -> BTreeMap<String, BTreeMap<String, (u64, f64)>> {
     let mut stats_table = BTreeMap::new();
 
-    let mut rdr =
-        csv::Reader::from_reader(include_bytes!(FOUR_REGION_LINK_STATS_CSV!()).as_slice());
+    let mut rdr = csv::Reader::from_reader(csv);
     rdr.deserialize()
         .for_each(|result: Result<(String, String, u64, f64), _>| {
             if let Ok((from, to, bitrate, latency)) = result {
@@ -205,13 +202,22 @@ pub struct MultiRegionNetworkEmulationConfig {
 impl Default for MultiRegionNetworkEmulationConfig {
     fn default() -> Self {
         Self {
-            link_stats_table: get_link_stats_table(),
+            link_stats_table: get_link_stats_table(FOUR_REGION_LINK_STATS),
             inter_region_config: InterRegionNetEmConfig::default(),
             intra_region_config: Some(IntraRegionNetEmConfig::default()),
         }
     }
 }
 
+impl MultiRegionNetworkEmulationConfig {
+    pub fn two_region() -> Self {
+        Self {
+            link_stats_table: get_link_stats_table(TWO_REGION_LINK_STATS),
+            ..Default::default()
+        }
+    }
+}
+
 /// A test to emulate network conditions for a multi-region setup.
 #[derive(Default)]
 pub struct MultiRegionNetworkEmulationTest {