Move config parameters for ExecutionTimeEstimate mode into protocol and node configs.

Author: aschran

Cargo.lock

@@ -485,6 +485,7 @@ mod test {
                         target_utilization: rng.gen_range(1..=100),
                         allowed_txn_cost_overage_burst_limit_us: rng.gen_range(0..500_000),
                         max_txn_cost_overage_per_object_in_commit_us: 10_000_000_000,
+                        max_estimate_ms: 1_500,
                     },
                 ),
             ]

crates/sui-benchmark/tests/simtest.rs

@@ -16,6 +16,7 @@ bcs.workspace = true
 csv.workspace = true
 dirs.workspace = true
 fastcrypto.workspace = true
+nonzero_ext.workspace = true
 once_cell.workspace = true
 rand.workspace = true
 serde = { workspace = true, features = ["derive", "rc"] }

crates/sui-config/Cargo.toml

@@ -10,13 +10,14 @@ use crate::Config;
 use anyhow::Result;
 use consensus_config::Parameters as ConsensusParameters;
 use mysten_common::fatal;
+use nonzero_ext::nonzero;
 use once_cell::sync::OnceCell;
 use rand::rngs::OsRng;
 use serde::{Deserialize, Serialize};
 use serde_with::serde_as;
 use std::collections::{BTreeMap, BTreeSet};
 use std::net::SocketAddr;
-use std::num::NonZeroUsize;
+use std::num::{NonZeroU32, NonZeroUsize};
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::Duration;
@@ -33,7 +34,7 @@ use sui_types::traffic_control::{PolicyConfig, RemoteFirewallConfig};
 
 use sui_types::crypto::{get_key_pair_from_rng, AccountKeyPair, AuthorityKeyPair};
 use sui_types::multiaddr::Multiaddr;
-use tracing::{error, info};
+use tracing::info;
 
 // Default max number of concurrent requests served
 pub const DEFAULT_GRPC_CONCURRENCY_LIMIT: usize = 20000000000;
@@ -202,17 +203,95 @@ pub struct NodeConfig {
     #[serde(skip_serializing_if = "Option::is_none")]
     pub enable_db_write_stall: Option<bool>,
 
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub execution_time_observer_config: Option<ExecutionTimeObserverConfig>,
+}
+
+#[derive(Clone, Debug, Default, Deserialize, Serialize)]
+#[serde(rename_all = "kebab-case")]
+pub struct ExecutionTimeObserverConfig {
     /// Size of the channel used for buffering local execution time observations.
     ///
     /// If unspecified, this will default to `128`.
-    #[serde(default = "default_local_execution_time_channel_capacity")]
-    pub local_execution_time_channel_capacity: usize,
+    pub observation_channel_capacity: Option<NonZeroUsize>,
 
     /// Size of the LRU cache used for storing local execution time observations.
     ///
-    /// If unspecified, this will default to `10000`.
-    #[serde(default = "default_local_execution_time_cache_size")]
-    pub local_execution_time_cache_size: usize,
+    /// If unspecified, this will default to `10_000`.
+    pub observation_cache_size: Option<NonZeroUsize>,
+
+    /// Size of the LRU cache used for tracking object utilization.
+    ///
+    /// If unspecified, this will default to `50_000`.
+    pub object_utilization_cache_size: Option<NonZeroUsize>,
+
+    /// Unless target object utilization is exceeded by at least this amount, no observation
+    /// will be shared with consensus.
+    ///
+    /// If unspecified, this will default to `100` milliseconds.
+    pub observation_sharing_object_utilization_threshold: Option<Duration>,
+
+    /// Unless the current local observation differs from the last one we shared by at least this
+    /// percentage, no observation will be shared with consensus.
+    ///
+    /// If unspecified, this will default to `0.05`.
+    pub observation_sharing_diff_threshold: Option<f64>,
+
+    /// Minimum interval between sharing multiple observations of the same key.
+    ///
+    /// If unspecified, this will default to `5` seconds.
+    pub observation_sharing_min_interval: Option<Duration>,
+
+    /// Global per-second rate limit for sharing observations. This is a safety valve and
+    /// should not trigger during normal operation.
+    ///
+    /// If unspecified, this will default to `10` observations per second.
+    pub observation_sharing_rate_limit: Option<NonZeroU32>,
+
+    /// Global burst limit for sharing observations.
+    ///
+    /// If unspecified, this will default to `100` observations.
+    pub observation_sharing_burst_limit: Option<NonZeroU32>,
+}
+
+impl ExecutionTimeObserverConfig {
+    pub fn observation_channel_capacity(&self) -> NonZeroUsize {
+        self.observation_channel_capacity
+            .unwrap_or(nonzero!(128usize))
+    }
+
+    pub fn observation_cache_size(&self) -> NonZeroUsize {
+        self.observation_cache_size.unwrap_or(nonzero!(10_000usize))
+    }
+
+    pub fn object_utilization_cache_size(&self) -> NonZeroUsize {
+        self.object_utilization_cache_size
+            .unwrap_or(nonzero!(50_000usize))
+    }
+
+    pub fn observation_sharing_object_utilization_threshold(&self) -> Duration {
+        self.observation_sharing_object_utilization_threshold
+            .unwrap_or(Duration::from_millis(100))
+    }
+
+    pub fn observation_sharing_diff_threshold(&self) -> f64 {
+        self.observation_sharing_diff_threshold.unwrap_or(0.05)
+    }
+
+    pub fn observation_sharing_min_interval(&self) -> Duration {
+        self.observation_sharing_min_interval
+            .unwrap_or(Duration::from_secs(5))
+    }
+
+    pub fn observation_sharing_rate_limit(&self) -> NonZeroU32 {
+        self.observation_sharing_rate_limit
+            .unwrap_or(nonzero!(10u32))
+    }
+
+    pub fn observation_sharing_burst_limit(&self) -> NonZeroU32 {
+        self.observation_sharing_burst_limit
+            .unwrap_or(nonzero!(100u32))
+    }
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
@@ -552,14 +631,6 @@ pub fn default_end_of_epoch_broadcast_channel_capacity() -> usize {
     128
 }
 
-pub fn default_local_execution_time_channel_capacity() -> usize {
-    128
-}
-
-pub fn default_local_execution_time_cache_size() -> usize {
-    10000
-}
-
 pub fn bool_true() -> bool {
     true
 }
@@ -655,13 +726,6 @@ impl NodeConfig {
     pub fn rpc(&self) -> Option<&sui_rpc_api::Config> {
         self.rpc.as_ref()
     }
-
-    pub fn local_execution_time_cache_size(&self) -> NonZeroUsize {
-        NonZeroUsize::new(self.local_execution_time_cache_size).unwrap_or_else(|| {
-            error!("local_execution_time_cache_size must be non-zero - defaulting to 10000");
-            NonZeroUsize::new(10000).unwrap()
-        })
-    }
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize)]

crates/sui-config/src/node.rs

@@ -404,7 +404,7 @@ pub struct AuthorityPerEpochStore {
     randomness_reporter: OnceCell<RandomnessReporter>,
 
     /// Manages recording execution time observations and generating estimates.
-    execution_time_estimator: tokio::sync::Mutex<ExecutionTimeEstimator>,
+    execution_time_estimator: tokio::sync::Mutex<Option<ExecutionTimeEstimator>>,
     tx_local_execution_time:
         OnceCell<mpsc::Sender<(ProgrammableTransaction, Vec<ExecutionTiming>, Duration)>>,
     // Saved at end of epoch for propagating observations to the next.
@@ -892,23 +892,31 @@ impl AuthorityPerEpochStore {
             .execution_time_observations
             .safe_iter()
             .collect::<Result<Vec<_>, _>>()?;
-        let execution_time_estimator = ExecutionTimeEstimator::new(
-            committee.clone(),
-            // Load observations stored at end of previous epoch.
-            Self::get_stored_execution_time_observations(
-                &protocol_config,
-                committee.clone(),
-                &*object_store,
-            )
-            // Load observations stored during the current epoch.
-            .chain(execution_time_observations.into_iter().flat_map(
-                |((generation, source), observations)| {
-                    observations
-                        .into_iter()
-                        .map(move |(key, duration)| (source, generation, key, duration))
-                },
-            )),
-        );
+        let execution_time_estimator =
+            if let PerObjectCongestionControlMode::ExecutionTimeEstimate(protocol_params) =
+                protocol_config.per_object_congestion_control_mode()
+            {
+                Some(ExecutionTimeEstimator::new(
+                    committee.clone(),
+                    protocol_params,
+                    // Load observations stored at end of previous epoch.
+                    Self::get_stored_execution_time_observations(
+                        &protocol_config,
+                        committee.clone(),
+                        &*object_store,
+                    )
+                    // Load observations stored during the current epoch.
+                    .chain(execution_time_observations.into_iter().flat_map(
+                        |((generation, source), observations)| {
+                            observations
+                                .into_iter()
+                                .map(move |(key, duration)| (source, generation, key, duration))
+                        },
+                    )),
+                ))
+            } else {
+                None
+            };
 
         let s = Arc::new(Self {
             name,
@@ -1993,7 +2001,7 @@ impl AuthorityPerEpochStore {
 
     fn should_defer(
         &self,
-        execution_time_estimator: &ExecutionTimeEstimator,
+        execution_time_estimator: Option<&ExecutionTimeEstimator>,
         cert: &VerifiedExecutableTransaction,
         commit_info: &ConsensusCommitInfo,
         dkg_failed: bool,
@@ -3001,12 +3009,12 @@ impl AuthorityPerEpochStore {
             estimates,
         } in execution_time_observations
         {
+            let Some(estimator) = execution_time_estimator.as_mut() else {
+                error!("dropping ExecutionTimeObservation from possibly-Byzantine authority {authority:?} sent when ExecutionTimeEstimate mode is not enabled");
+                continue;
+            };
             let authority_index = self.committee.authority_index(&authority).unwrap();
-            execution_time_estimator.process_observations_from_consensus(
-                authority_index,
-                generation,
-                &estimates,
-            );
+            estimator.process_observations_from_consensus(authority_index, generation, &estimates);
             output.insert_execution_time_observation(authority_index, generation, estimates);
         }
 
@@ -3063,7 +3071,7 @@ impl AuthorityPerEpochStore {
                 randomness_manager.as_deref_mut(),
                 dkg_failed,
                 randomness_round,
-                &execution_time_estimator,
+                execution_time_estimator.as_ref(),
                 authority_metrics,
             )
             .await?;
@@ -3073,11 +3081,13 @@ impl AuthorityPerEpochStore {
         // If this is the final round, record execution time observations for storage in the
         // end-of-epoch tx.
         if final_round {
-            self.end_of_epoch_execution_time_observations
-                .set(execution_time_estimator.take_observations())
+            if let Some(estimator) = execution_time_estimator.as_mut() {
+                self.end_of_epoch_execution_time_observations
+                .set(estimator.take_observations())
                 .expect(
                     "`stored_execution_time_observations` should only be set once at end of epoch",
                 );
+            }
             drop(execution_time_estimator); // make sure this is not used after `take_observations`
         }
 
@@ -3404,7 +3414,7 @@ impl AuthorityPerEpochStore {
         mut randomness_manager: Option<&mut RandomnessManager>,
         dkg_failed: bool,
         randomness_round: Option<RandomnessRound>,
-        execution_time_estimator: &ExecutionTimeEstimator,
+        execution_time_estimator: Option<&ExecutionTimeEstimator>,
         authority_metrics: &Arc<AuthorityMetrics>,
     ) -> SuiResult<(
         Vec<VerifiedExecutableTransaction>,    // transactions to schedule
@@ -3707,7 +3717,7 @@ impl AuthorityPerEpochStore {
         dkg_failed: bool,
         generating_randomness: bool,
         shared_object_congestion_tracker: &mut SharedObjectCongestionTracker,
-        execution_time_estimator: &ExecutionTimeEstimator,
+        execution_time_estimator: Option<&ExecutionTimeEstimator>,
         authority_metrics: &Arc<AuthorityMetrics>,
     ) -> SuiResult<ConsensusCertificateResult> {
         let _scope = monitored_scope("ConsensusCommitHandler::process_consensus_transaction");
@@ -3979,7 +3989,7 @@ impl AuthorityPerEpochStore {
         dkg_failed: bool,
         generating_randomness: bool,
         shared_object_congestion_tracker: &mut SharedObjectCongestionTracker,
-        execution_time_estimator: &ExecutionTimeEstimator,
+        execution_time_estimator: Option<&ExecutionTimeEstimator>,
         authority_metrics: &Arc<AuthorityMetrics>,
     ) -> SuiResult<ConsensusCertificateResult> {
         let _scope = monitored_scope("ConsensusCommitHandler::process_consensus_user_transaction");