[Network] Adding AptosPerf Client

.github/workflows/lint-test.yaml

@@ -122,7 +122,7 @@ jobs:
       - uses: taiki-e/install-action@v1.5.6
         with:
           tool: nextest
-      - run: cargo nextest run --profile ci --locked --workspace --exclude smoke-test --exclude aptos-testcases --retries 3 --no-fail-fast
+      - run: RUST_MIN_STACK=4194304 cargo nextest run --profile ci --locked --workspace --exclude smoke-test --exclude aptos-testcases --retries 3 --no-fail-fast
         env:
           INDEXER_DATABASE_URL: postgresql://postgres@localhost/postgres
 

Cargo.toml

@@ -287,7 +287,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" }
@@ -345,7 +345,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 = 9105;
 
 #[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/network.rs

@@ -184,7 +184,7 @@ impl NetworkSender {
         // Broadcast message over direct-send to all other validators.
         if let Err(err) = self
             .consensus_network_client
-            .send_to_many(other_validators.into_iter(), msg)
+            .send_to_many(other_validators.into_iter(), &msg)
         {
             warn!(error = ?err, "Error broadcasting message");
         }

consensus/src/network_interface.rs

@@ -149,7 +149,7 @@ impl<NetworkClient: NetworkClientInterface<ConsensusMsg>> ConsensusNetworkClient
     pub fn send_to_many(
         &self,
         peers: impl Iterator<Item = PeerId>,
-        message: ConsensusMsg,
+        message: &ConsensusMsg,
     ) -> Result<(), Error> {
         let peer_network_ids: Vec<PeerNetworkId> = peers
             .map(|peer| self.get_peer_network_id_for_peer(peer))

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/aptos-compression/src/metrics.rs

@@ -20,6 +20,7 @@ pub enum CompressionClient {
     Consensus,
     Mempool,
     StateSync,
+    NetPerf,
 }
 
 impl CompressionClient {
@@ -29,6 +30,7 @@ impl CompressionClient {
             Self::Consensus => "consensus",
             Self::Mempool => "mempool",
             Self::StateSync => "state_sync",
+            Self::NetPerf => "netperf",
         }
     }
 }

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 }