markuze/aptos perf

Cargo.toml

@@ -284,7 +284,7 @@ assert_approx_eq = "1.1.0"
 assert_unordered = "0.1.1"
 async-stream = "0.3"
 async-trait = "0.1.53"
-axum = "0.5.16"
+axum = "0.6.1"
 base64 = "0.13.0"
 backtrace = "0.3.58"
 bcs = { git = "https://github.com/aptos-labs/bcs.git", rev = "d31fab9d81748e2594be5cd5cdf845786a30562d" }
@@ -342,7 +342,7 @@ hkdf = "0.10.0"
 hostname = "0.3.1"
 http = "0.2.3"
 httpmock = "0.6"
-hyper = { version = "0.14.18", features = ["full"] }
+hyper = { version = "0.14.23", features = ["full"] }
 hyper-tls = "0.5.0"
 include_dir = { version = "0.7.2", features = ["glob"] }
 indicatif = "0.15.0"

config/src/config/mod.rs

@@ -434,10 +434,12 @@ impl NodeConfig {
 
         if let Some(network) = self.validator_network.as_mut() {
             network.listen_address = crate::utils::get_available_port_in_multiaddr(true);
+            network.randomize_ports();
         }
 
         for network in self.full_node_networks.iter_mut() {
             network.listen_address = crate::utils::get_available_port_in_multiaddr(true);
+            network.randomize_ports();
         }
     }
 

config/src/config/network_config.rs

@@ -54,6 +54,8 @@ pub const INBOUND_TCP_RX_BUFFER_SIZE: u32 = 3 * 1024 * 1024; // 3MB ~6MB/s with
 pub const INBOUND_TCP_TX_BUFFER_SIZE: u32 = 512 * 1024; // 1MB use a bigger spoon
 pub const OUTBOUND_TCP_RX_BUFFER_SIZE: u32 = 3 * 1024 * 1024; // 3MB ~6MB/s with 500ms latency
 pub const OUTBOUND_TCP_TX_BUFFER_SIZE: u32 = 1024 * 1024; // 1MB use a bigger spoon
+pub const ENABLE_APTOS_NETPERF_CLIENT: bool = true;
+pub const DEFAULT_APTOS_NETPERF_CLIENT_PORT: u16 = 9107;
 
 #[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
 #[serde(default, deny_unknown_fields)]
@@ -80,6 +82,7 @@ pub struct NetworkConfig {
     // Select this to enforce that both peers should authenticate each other, otherwise
     // authentication only occurs for outgoing connections.
     pub mutual_authentication: bool,
+    pub netperf_client_port: Option<u16>,
     pub network_id: NetworkId,
     pub runtime_threads: Option<usize>,
     pub inbound_rx_buffer_size_bytes: Option<u32>,
@@ -120,7 +123,19 @@ impl Default for NetworkConfig {
     }
 }
 
+fn netperf_client_port(enabled: bool) -> Option<u16> {
+    if enabled {
+        Some(DEFAULT_APTOS_NETPERF_CLIENT_PORT)
+    } else {
+        None
+    }
+}
+
 impl NetworkConfig {
+    pub fn randomize_ports(&mut self) {
+        self.netperf_client_port = Some(utils::get_available_port());
+    }
+
     pub fn network_with_id(network_id: NetworkId) -> NetworkConfig {
         let mutual_authentication = network_id.is_validator_network();
         let mut config = Self {
@@ -129,6 +144,7 @@ impl NetworkConfig {
             identity: Identity::None,
             listen_address: "/ip4/0.0.0.0/tcp/6180".parse().unwrap(),
             mutual_authentication,
+            netperf_client_port: netperf_client_port(ENABLE_APTOS_NETPERF_CLIENT),
             network_id,
             runtime_threads: None,
             seed_addrs: HashMap::new(),

consensus/src/payload_manager.rs

@@ -18,6 +18,7 @@ use tokio::sync::oneshot;
 
 /// Responsible to extract the transactions out of the payload and notify QuorumStore about commits.
 /// If QuorumStore is enabled, has to ask BatchReader for the transaction behind the proofs of availability in the payload.
+#[allow(dead_code)]
 pub enum PayloadManager {
     DirectMempool,
     InQuorumStore(BatchReader, Mutex<Sender<PayloadRequest>>),

consensus/src/state_computer.rs

@@ -37,10 +37,12 @@ type NotificationType = (
     Vec<ContractEvent>,
 );
 
+#[allow(dead_code)]
 type CommitType = (u64, Round, Vec<Payload>);
 
 /// Basic communication with the Execution module;
 /// implements StateComputer traits.
+#[allow(dead_code)]
 pub struct ExecutionProxy {
     executor: Arc<dyn BlockExecutorTrait>,
     txn_notifier: Arc<dyn TxnNotifier>,

crates/channel/src/message_queues.rs

@@ -123,6 +123,7 @@ impl<K: Eq + Hash + Clone, T> PerKeyQueue<K, T> {
             // For example, many of our queues have a max capacity of 1024. To
             // handle a single rpc from a transient peer, we would end up
             // allocating ~ 96 b * 1024 ~ 64 Kib per queue.
+            //TODO: Reconsider this. Maybe true for VFN but not Validators
             .or_insert_with(|| VecDeque::with_capacity(1));
 
         // Add the key to our round-robin queue if it's not already there

network/Cargo.toml

@@ -33,8 +33,10 @@ aptos-short-hex-str = { workspace = true }
 aptos-time-service = { workspace = true }
 aptos-types = { workspace = true }
 async-trait = { workspace = true }
+axum = { workspace = true }
 bcs = { workspace = true }
 bytes = { workspace = true }
+dashmap = { worksapce = true }
 futures = { workspace = true }
 futures-util = { workspace = true }
 hex = { workspace = true }

network/builder/src/builder.rs

@@ -24,7 +24,10 @@ use aptos_infallible::RwLock;
 use aptos_logger::prelude::*;
 use aptos_netcore::transport::tcp::TCPBufferCfg;
 use aptos_network::{
-    application::storage::PeerMetadataStorage,
+    application::{
+        netperf::{builder::NetPerfBuilder, NetPerf},
+        storage::PeerMetadataStorage,
+    },
     connectivity_manager::{builder::ConnectivityManagerBuilder, ConnectivityRequest},
     constants::MAX_MESSAGE_SIZE,
     logging::NetworkSchema,
@@ -70,6 +73,7 @@ pub struct NetworkBuilder {
     health_checker_builder: Option<HealthCheckerBuilder>,
     peer_manager_builder: PeerManagerBuilder,
     peer_metadata_storage: Arc<PeerMetadataStorage>,
+    netperf_builder: Option<NetPerfBuilder>,
 }
 
 impl NetworkBuilder {
@@ -125,6 +129,7 @@ impl NetworkBuilder {
             health_checker_builder: None,
             peer_manager_builder,
             peer_metadata_storage,
+            netperf_builder: None,
         }
     }
 
@@ -226,6 +231,10 @@ impl NetworkBuilder {
             config.ping_failures_tolerated,
         );
 
+        if let Some(netperf_port) = config.netperf_client_port {
+            network_builder.add_network_perf(netperf_port);
+        }
+
         // Always add a connectivity manager to keep track of known peers
         let seeds = merge_seeds(config);
 
@@ -315,6 +324,16 @@ impl NetworkBuilder {
             );
         }
 
+        if let Some(netperf_builder) = self.netperf_builder.as_mut() {
+            if self.network_context.role() == RoleType::Validator {
+                netperf_builder.start(executor);
+                debug!(
+                    NetworkSchema::new(&self.network_context),
+                    "{} Started Aptos NetPerf", self.network_context
+                );
+            }
+        }
+
         if let Some(discovery_listeners) = self.discovery_listeners.take() {
             discovery_listeners
                 .into_iter()
@@ -426,6 +445,25 @@ impl NetworkBuilder {
             .push(listener);
     }
 
+    /// Add a Aptos NetPerf to the network.
+    fn add_network_perf(&mut self, netperf_port: u16) -> &mut Self {
+        let (netperf_tx, netperf_rx) =
+            self.add_client_and_service(&NetPerf::network_endpoint_config());
+        self.netperf_builder = Some(NetPerfBuilder::new(
+            self.network_context(),
+            self.peer_metadata_storage.clone(),
+            Arc::new(netperf_tx),
+            netperf_rx,
+            netperf_port,
+        ));
+        debug!(
+            NetworkSchema::new(&self.network_context),
+            "{} Created Aptos NetPerf", self.network_context
+        );
+
+        self
+    }
+
     /// Add a HealthChecker to the network.
     fn add_connection_monitoring(
         &mut self,

network/src/application/mod.rs

@@ -3,6 +3,7 @@
 
 pub mod error;
 pub mod interface;
+pub mod netperf;
 pub mod storage;
 pub mod types;