Gossip: Add dynamic stake weighting based on fraction of unstaked nodes - Fixed Point Math (#7098)

* add dynamic weighting based on fraction of unstaked nodes. fixed point math

* Add setup for account control

* address comments - add explicit types, bundle config into Dynamic, refactor, etc

* update to vanity key: gosW... (gossip weight)

* agave-unstable-api implementation

* refactor add apply_cfg tests

* refactor and address comments

* update interpolate. change lpf

* fix deser bug. dont forget to rm logs

* update WeightingConfig to match record program

* address comments and make default setting static for PushActiveSet

Author: gregcusack

Cargo.lock

@@ -51,6 +51,7 @@ members = [
     "ledger-tool",
     "local-cluster",
     "log-analyzer",
+    "low-pass-filter",
     "measure",
     "memory-management",
     "merkle-tree",
@@ -452,6 +453,7 @@ solana-loader-v4-interface = "2.2.1"
 solana-loader-v4-program = { path = "programs/loader-v4", version = "=3.0.0" }
 solana-local-cluster = { path = "local-cluster", version = "=3.0.0" }
 solana-logger = "3.0.0"
+solana-low-pass-filter = { path = "low-pass-filter", version = "=3.0.0" }
 solana-measure = { path = "measure", version = "=3.0.0" }
 solana-merkle-tree = { path = "merkle-tree", version = "=3.0.0" }
 solana-message = "2.4.0"

Cargo.toml

@@ -33,6 +33,7 @@ frozen-abi = [
     "solana-vote/frozen-abi",
     "solana-vote-program/frozen-abi",
 ]
+agave-unstable-api = ["solana-low-pass-filter/agave-unstable-api"]
 
 [dependencies]
 agave-feature-set = { workspace = true }
@@ -56,6 +57,7 @@ serde-big-array = { workspace = true }
 serde_bytes = { workspace = true }
 serde_derive = { workspace = true }
 siphasher = { workspace = true }
+solana-account = { workspace = true }
 solana-bloom = { workspace = true }
 solana-clap-utils = { workspace = true }
 solana-client = { workspace = true }
@@ -73,6 +75,7 @@ solana-hash = "=2.3.0"
 solana-keypair = "=2.2.1"
 solana-ledger = { workspace = true, features = ["agave-unstable-api"] }
 solana-logger = "=3.0.0"
+solana-low-pass-filter = { workspace = true, features = ["agave-unstable-api"] }
 solana-measure = { workspace = true }
 solana-metrics = { workspace = true }
 solana-native-token = "=2.2.2"

gossip/Cargo.toml

@@ -62,7 +62,7 @@ use {
     },
     solana_pubkey::Pubkey,
     solana_rayon_threadlimit::get_thread_count,
-    solana_runtime::bank_forks::BankForks,
+    solana_runtime::{bank::Bank, bank_forks::BankForks},
     solana_sanitize::Sanitize,
     solana_signature::Signature,
     solana_signer::Signer,
@@ -130,6 +130,8 @@ pub const DEFAULT_CONTACT_DEBUG_INTERVAL_MILLIS: u64 = 10_000;
 pub const DEFAULT_CONTACT_SAVE_INTERVAL_MILLIS: u64 = 60_000;
 // Limit number of unique pubkeys in the crds table.
 pub(crate) const CRDS_UNIQUE_PUBKEY_CAPACITY: usize = 8192;
+// Interval between push active set refreshes.
+pub const REFRESH_PUSH_ACTIVE_SET_INTERVAL_MS: u64 = CRDS_GOSSIP_PULL_CRDS_TIMEOUT_MS / 2;
 
 // Must have at least one socket to monitor the TVU port
 pub const MINIMUM_NUM_TVU_RECEIVE_SOCKETS: NonZeroUsize = NonZeroUsize::new(1).unwrap();
@@ -214,6 +216,7 @@ impl ClusterInfo {
         stakes: &HashMap<Pubkey, u64>,
         gossip_validators: Option<&HashSet<Pubkey>>,
         sender: &impl ChannelSend<PacketBatch>,
+        maybe_bank_ref: Option<&Bank>,
     ) {
         let shred_version = self.my_contact_info.read().unwrap().shred_version();
         let self_keypair: Arc<Keypair> = self.keypair().clone();
@@ -226,6 +229,7 @@ impl ClusterInfo {
             &self.ping_cache,
             &mut pings,
             &self.socket_addr_space,
+            maybe_bank_ref,
         );
         let pings = pings
             .into_iter()
@@ -1448,7 +1452,7 @@ impl ClusterInfo {
             .thread_name(|i| format!("solGossipRun{i:02}"))
             .build()
             .unwrap();
-        let mut epoch_specs = bank_forks.map(EpochSpecs::from);
+        let mut epoch_specs = bank_forks.clone().map(EpochSpecs::from);
         Builder::new()
             .name("solGossip".to_string())
             .spawn(move || {
@@ -1504,13 +1508,19 @@ impl ClusterInfo {
                     entrypoints_processed = entrypoints_processed || self.process_entrypoints();
                     //TODO: possibly tune this parameter
                     //we saw a deadlock passing an self.read().unwrap().timeout into sleep
-                    if start - last_push > CRDS_GOSSIP_PULL_CRDS_TIMEOUT_MS / 2 {
+                    if start - last_push > REFRESH_PUSH_ACTIVE_SET_INTERVAL_MS {
+                        let maybe_bank = bank_forks
+                            .as_ref()
+                            .and_then(|bf| bf.read().ok())
+                            .map(|forks| forks.root_bank());
+                        let maybe_bank_ref = maybe_bank.as_deref();
                         self.refresh_my_gossip_contact_info();
                         self.refresh_push_active_set(
                             &recycler,
                             &stakes,
                             gossip_validators.as_ref(),
                             &sender,
+                            maybe_bank_ref,
                         );
                         last_push = timestamp();
                     }
@@ -2831,6 +2841,7 @@ mod tests {
             &cluster_info.ping_cache,
             &mut Vec::new(), // pings
             &SocketAddrSpace::Unspecified,
+            None,
         );
         let mut reqs = cluster_info.generate_new_gossip_requests(
             &thread_pool,
@@ -2972,6 +2983,7 @@ mod tests {
             &cluster_info.ping_cache,
             &mut Vec::new(), // pings
             &SocketAddrSpace::Unspecified,
+            None,
         );
         //check that all types of gossip messages are signed correctly
         cluster_info.flush_push_queue();

gossip/src/cluster_info.rs

@@ -27,6 +27,7 @@ use {
     solana_keypair::Keypair,
     solana_ledger::shred::Shred,
     solana_pubkey::Pubkey,
+    solana_runtime::bank::Bank,
     solana_signer::Signer,
     solana_streamer::socket::SocketAddrSpace,
     solana_time_utils::timestamp,
@@ -186,6 +187,7 @@ impl CrdsGossip {
         ping_cache: &Mutex<PingCache>,
         pings: &mut Vec<(SocketAddr, Ping)>,
         socket_addr_space: &SocketAddrSpace,
+        maybe_bank_ref: Option<&Bank>,
     ) {
         self.push.refresh_push_active_set(
             &self.crds,
@@ -196,6 +198,7 @@ impl CrdsGossip {
             ping_cache,
             pings,
             socket_addr_space,
+            maybe_bank_ref,
         )
     }
 
@@ -445,6 +448,7 @@ mod test {
             &ping_cache,
             &mut Vec::new(), // pings
             &SocketAddrSpace::Unspecified,
+            None,
         );
         let now = timestamp();
         //incorrect dest

gossip/src/crds_gossip.rs

@@ -21,10 +21,12 @@ use {
         protocol::{Ping, PingCache},
         push_active_set::PushActiveSet,
         received_cache::ReceivedCache,
+        stake_weighting_config::{get_gossip_config_from_account, WeightingConfig},
     },
     itertools::Itertools,
     solana_keypair::Keypair,
     solana_pubkey::Pubkey,
+    solana_runtime::bank::Bank,
     solana_signer::Signer,
     solana_streamer::socket::SocketAddrSpace,
     solana_time_utils::timestamp,
@@ -49,6 +51,7 @@ const CRDS_GOSSIP_PRUNE_MSG_TIMEOUT_MS: u64 = 500;
 const CRDS_GOSSIP_PRUNE_STAKE_THRESHOLD_PCT: f64 = 0.15;
 const CRDS_GOSSIP_PRUNE_MIN_INGRESS_NODES: usize = 2;
 const CRDS_GOSSIP_PUSH_ACTIVE_SET_SIZE: usize = CRDS_GOSSIP_PUSH_FANOUT + 3;
+const CONFIG_REFRESH_INTERVAL_MS: u64 = 60_000;
 
 pub struct CrdsGossipPush {
     /// Active set of validators for push
@@ -65,12 +68,13 @@ pub struct CrdsGossipPush {
     pub num_total: AtomicUsize,
     pub num_old: AtomicUsize,
     pub num_pushes: AtomicUsize,
+    last_cfg_poll_ms: Mutex<u64>,
 }
 
 impl Default for CrdsGossipPush {
     fn default() -> Self {
         Self {
-            active_set: RwLock::default(),
+            active_set: RwLock::new(PushActiveSet::new_static()),
             crds_cursor: Mutex::default(),
             received_cache: Mutex::new(ReceivedCache::new(2 * CRDS_UNIQUE_PUBKEY_CAPACITY)),
             push_fanout: CRDS_GOSSIP_PUSH_FANOUT,
@@ -79,6 +83,7 @@ impl Default for CrdsGossipPush {
             num_total: AtomicUsize::default(),
             num_old: AtomicUsize::default(),
             num_pushes: AtomicUsize::default(),
+            last_cfg_poll_ms: Mutex::new(0),
         }
     }
 }
@@ -238,6 +243,22 @@ impl CrdsGossipPush {
         active_set.prune(self_pubkey, peer, origins, stakes);
     }
 
+    fn maybe_refresh_weighting_config(
+        &self,
+        maybe_bank_ref: Option<&Bank>,
+        now_ms: u64,
+    ) -> Option<WeightingConfig> {
+        let bank = maybe_bank_ref?;
+        {
+            let mut last = self.last_cfg_poll_ms.lock().unwrap();
+            if now_ms.saturating_sub(*last) < CONFIG_REFRESH_INTERVAL_MS {
+                return None;
+            }
+            *last = now_ms;
+        }
+        get_gossip_config_from_account(bank)
+    }
+
     /// Refresh the push active set.
     #[allow(clippy::too_many_arguments)]
     pub(crate) fn refresh_push_active_set(
@@ -250,6 +271,7 @@ impl CrdsGossipPush {
         ping_cache: &Mutex<PingCache>,
         pings: &mut Vec<(SocketAddr, Ping)>,
         socket_addr_space: &SocketAddrSpace,
+        maybe_bank_ref: Option<&Bank>,
     ) {
         let mut rng = rand::thread_rng();
         // Active and valid gossip nodes with matching shred-version.
@@ -280,13 +302,18 @@ impl CrdsGossipPush {
             return;
         }
         let cluster_size = crds.read().unwrap().num_pubkeys().max(stakes.len());
+        let maybe_cfg = self.maybe_refresh_weighting_config(maybe_bank_ref, timestamp());
         let mut active_set = self.active_set.write().unwrap();
+        if let Some(cfg) = maybe_cfg {
+            active_set.apply_cfg(&cfg);
+        }
         active_set.rotate(
             &mut rng,
             CRDS_GOSSIP_PUSH_ACTIVE_SET_SIZE,
             cluster_size,
             &nodes,
             stakes,
+            &self_keypair.pubkey(),
         )
     }
 }
@@ -447,6 +474,7 @@ mod tests {
             &ping_cache,
             &mut Vec::new(), // pings
             &SocketAddrSpace::Unspecified,
+            None,
         );
 
         let new_msg = CrdsValue::new_unsigned(CrdsData::from(ContactInfo::new_localhost(
@@ -514,6 +542,7 @@ mod tests {
             &ping_cache,
             &mut Vec::new(),
             &SocketAddrSpace::Unspecified,
+            None,
         );
 
         // push 3's contact info to 1 and 2 and 3
@@ -557,6 +586,7 @@ mod tests {
             &ping_cache,
             &mut Vec::new(), // pings
             &SocketAddrSpace::Unspecified,
+            None,
         );
 
         let new_msg = CrdsValue::new_unsigned(CrdsData::from(ContactInfo::new_localhost(
@@ -605,6 +635,7 @@ mod tests {
             &ping_cache,
             &mut Vec::new(), // pings
             &SocketAddrSpace::Unspecified,
+            None,
         );
 
         let mut ci = ContactInfo::new_localhost(&solana_pubkey::new_rand(), 0);

gossip/src/crds_gossip_push.rs

@@ -41,6 +41,7 @@ mod protocol;
 mod push_active_set;
 mod received_cache;
 pub mod restart_crds_values;
+pub mod stake_weighting_config;
 pub mod weighted_shuffle;
 
 #[macro_use]