moves gossip CrdsValue hash calculations to CrdsValue::deserialize (#3593)

CrdsValue hashes are calculated in Crds::insert while holding an
exclusive lock on gossip CRDS table:
https://github.com/anza-xyz/agave/blob/d2e7488c1/gossip/src/crds.rs#L252

Additionally in order to calculate the hash, the CrdsValue needs to be
bincode serialized, which can exacerbate lock contention on gossip CRDS
table:
https://github.com/anza-xyz/agave/blob/d2e7488c1/gossip/src/crds.rs#L159

This commit instead moves hash calculation to CrdsValue::deserialize,
which is done in a thread-pool and without acquiring any locks:
https://github.com/anza-xyz/agave/blob/d2e7488c1/gossip/src/cluster_info.rs#L2317-L2326

Author: behzadnouri

gossip/src/crds.rs

@@ -35,19 +35,13 @@ use {
         crds_value::{CrdsValue, CrdsValueLabel},
     },
     assert_matches::debug_assert_matches,
-    bincode::serialize,
     indexmap::{
         map::{rayon::ParValues, Entry, IndexMap},
         set::IndexSet,
     },
     lru::LruCache,
     rayon::{prelude::*, ThreadPool},
-    solana_sdk::{
-        clock::Slot,
-        hash::{hash, Hash},
-        pubkey::Pubkey,
-        signature::Signature,
-    },
+    solana_sdk::{clock::Slot, hash::Hash, pubkey::Pubkey, signature::Signature},
     std::{
         cmp::Ordering,
         collections::{hash_map, BTreeMap, HashMap, VecDeque},
@@ -131,8 +125,6 @@ pub struct VersionedCrdsValue {
     pub value: CrdsValue,
     /// local time when updated
     pub(crate) local_timestamp: u64,
-    /// value hash
-    pub(crate) value_hash: Hash,
     /// None -> value upserted by GossipRoute::{LocalMessage,PullRequest}
     /// Some(0) -> value upserted by GossipRoute::PullResponse
     /// Some(k) if k > 0 -> value upserted by GossipRoute::PushMessage w/ k - 1 push duplicates
@@ -156,7 +148,6 @@ impl Cursor {
 
 impl VersionedCrdsValue {
     fn new(value: CrdsValue, cursor: Cursor, local_timestamp: u64, route: GossipRoute) -> Self {
-        let value_hash = hash(&serialize(&value).unwrap());
         let num_push_recv = match route {
             GossipRoute::LocalMessage => None,
             GossipRoute::PullRequest => None,
@@ -168,7 +159,6 @@ impl VersionedCrdsValue {
             ordinal: cursor.ordinal(),
             value,
             local_timestamp,
-            value_hash,
             num_push_recv,
         }
     }
@@ -223,10 +213,7 @@ fn overrides(value: &CrdsValue, other: &VersionedCrdsValue) -> bool {
         // Ties should be broken in a deterministic way across the cluster.
         // For backward compatibility this is done by comparing hash of
         // serialized values.
-        Ordering::Equal => {
-            let value_hash = hash(&serialize(&value).unwrap());
-            other.value_hash < value_hash
-        }
+        Ordering::Equal => other.value.hash() < value.hash(),
     }
 }
 
@@ -310,7 +297,7 @@ impl Crds {
                 // does not need to be updated.
                 debug_assert_eq!(entry.get().value.pubkey(), pubkey);
                 self.cursor.consume(value.ordinal);
-                self.purged.push_back((entry.get().value_hash, now));
+                self.purged.push_back((*entry.get().value.hash(), now));
                 entry.insert(value);
                 Ok(())
             }
@@ -323,8 +310,8 @@ impl Crds {
                 );
                 // Identify if the message is outdated (as opposed to
                 // duplicate) by comparing value hashes.
-                if entry.get().value_hash != value.value_hash {
-                    self.purged.push_back((value.value_hash, now));
+                if entry.get().value.hash() != value.value.hash() {
+                    self.purged.push_back((*value.value.hash(), now));
                     Err(CrdsError::InsertFailed)
                 } else if matches!(route, GossipRoute::PushMessage(_)) {
                     let entry = entry.get_mut();
@@ -560,7 +547,7 @@ impl Crds {
         let Some((index, _ /*label*/, value)) = self.table.swap_remove_full(key) else {
             return;
         };
-        self.purged.push_back((value.value_hash, now));
+        self.purged.push_back((*value.value.hash(), now));
         self.shards.remove(index, &value);
         match value.value.data() {
             CrdsData::ContactInfo(_) => {
@@ -837,7 +824,7 @@ mod tests {
             &Pubkey::default(),
             0,
         )));
-        let value_hash = hash(&serialize(&original).unwrap());
+        let value_hash = *original.hash();
         assert_matches!(crds.insert(original, 0, GossipRoute::LocalMessage), Ok(()));
         let val = CrdsValue::new_unsigned(CrdsData::ContactInfo(ContactInfo::new_localhost(
             &Pubkey::default(),
@@ -857,7 +844,7 @@ mod tests {
             &Pubkey::default(),
             0,
         )));
-        let val1_hash = hash(&serialize(&val1).unwrap());
+        let val1_hash = *val1.hash();
         assert_eq!(
             crds.insert(val1.clone(), 0, GossipRoute::LocalMessage),
             Ok(())
@@ -912,7 +899,7 @@ mod tests {
         let other = NodeInstance::new(&mut rng, pubkey, now - 1);
         let other = other.with_wallclock(now + 1);
         let other = make_crds_value(other);
-        let value_hash = hash(&serialize(&other).unwrap());
+        let value_hash = *other.hash();
         assert_eq!(
             crds.insert(other, now, GossipRoute::LocalMessage),
             Err(CrdsError::InsertFailed)
@@ -924,7 +911,7 @@ mod tests {
         for _ in 0..100 {
             let other = NodeInstance::new(&mut rng, pubkey, now);
             let other = make_crds_value(other);
-            let value_hash = hash(&serialize(&other).unwrap());
+            let value_hash = *other.hash();
             match crds.insert(other, now, GossipRoute::LocalMessage) {
                 Ok(()) => num_overrides += 1,
                 Err(CrdsError::InsertFailed) => {
@@ -1433,7 +1420,7 @@ mod tests {
             let purged: HashSet<_> = crds.purged.iter().map(|(hash, _)| hash).copied().collect();
             values
                 .into_iter()
-                .filter(|v| purged.contains(&v.value_hash))
+                .filter(|v| purged.contains(v.value.hash()))
                 .collect()
         };
         assert_eq!(purged.len() + crds.table.len(), num_values);
@@ -1515,10 +1502,10 @@ mod tests {
 
         assert_eq!(v1.value.label(), v2.value.label());
         assert_eq!(v1.value.wallclock(), v2.value.wallclock());
-        assert_ne!(v1.value_hash, v2.value_hash);
+        assert_ne!(v1.value.hash(), v2.value.hash());
         assert!(v1 != v2);
         assert!(!(v1 == v2));
-        if v1.value_hash > v2.value_hash {
+        if v1.value.hash() > v2.value.hash() {
             assert!(overrides(&v1.value, &v2));
             assert!(!overrides(&v2.value, &v1));
         } else {

gossip/src/crds_gossip_pull.rs

@@ -30,7 +30,7 @@ use {
     rayon::{prelude::*, ThreadPool},
     solana_bloom::bloom::{Bloom, ConcurrentBloom},
     solana_sdk::{
-        hash::{hash, Hash},
+        hash::Hash,
         native_token::LAMPORTS_PER_SOL,
         packet::PACKET_DATA_SIZE,
         pubkey::Pubkey,
@@ -349,7 +349,7 @@ impl CrdsGossipPull {
         let failed_inserts = responses
             .into_iter()
             .filter_map(upsert)
-            .map(|resp| hash(&bincode::serialize(&resp).unwrap()))
+            .map(|resp| *resp.hash())
             .collect();
         (active_values, expired_values, failed_inserts)
     }
@@ -427,7 +427,7 @@ impl CrdsGossipPull {
         thread_pool.install(|| {
             crds.par_values()
                 .with_min_len(PAR_MIN_LENGTH)
-                .map(|v| v.value_hash)
+                .map(|v| *v.value.hash())
                 .chain(crds.purged().with_min_len(PAR_MIN_LENGTH))
                 .chain(
                     failed_inserts
@@ -472,13 +472,13 @@ impl CrdsGossipPull {
             }
             let caller_wallclock = caller_wallclock.checked_add(jitter).unwrap_or(0);
             let pred = |entry: &&VersionedCrdsValue| {
-                debug_assert!(filter.test_mask(&entry.value_hash));
+                debug_assert!(filter.test_mask(entry.value.hash()));
                 // Skip values that are too new.
                 if entry.value.wallclock() > caller_wallclock {
                     total_skipped.fetch_add(1, Ordering::Relaxed);
                     false
                 } else {
-                    !filter.filter_contains(&entry.value_hash)
+                    !filter.filter_contains(entry.value.hash())
                 }
             };
             let out: Vec<_> = crds
@@ -811,7 +811,11 @@ pub(crate) mod tests {
         assert_eq!(filters.len(), MIN_NUM_BLOOM_FILTERS.max(4));
         let crds = crds.read().unwrap();
         let purged: Vec<_> = thread_pool.install(|| crds.purged().collect());
-        let hash_values: Vec<_> = crds.values().map(|v| v.value_hash).chain(purged).collect();
+        let hash_values: Vec<_> = crds
+            .values()
+            .map(|v| *v.value.hash())
+            .chain(purged)
+            .collect();
         // CrdsValue::new_rand may generate exact same value twice in which
         // case its hash-value is not added to purged values.
         assert!(
@@ -1246,10 +1250,7 @@ pub(crate) mod tests {
         node_crds
             .insert(old.clone(), 0, GossipRoute::LocalMessage)
             .unwrap();
-        let value_hash = {
-            let entry: &VersionedCrdsValue = node_crds.get(&old.label()).unwrap();
-            entry.value_hash
-        };
+        let value_hash = *node_crds.get::<&CrdsValue>(&old.label()).unwrap().hash();
         //verify self is valid
         assert_eq!(
             node_crds.get::<&CrdsValue>(&node_label).unwrap().label(),

gossip/src/crds_shards.rs

@@ -25,12 +25,12 @@ impl CrdsShards {
     }
 
     pub fn insert(&mut self, index: usize, value: &VersionedCrdsValue) -> bool {
-        let hash = CrdsFilter::hash_as_u64(&value.value_hash);
+        let hash = CrdsFilter::hash_as_u64(value.value.hash());
         self.shard_mut(hash).insert(index, hash).is_none()
     }
 
     pub fn remove(&mut self, index: usize, value: &VersionedCrdsValue) -> bool {
-        let hash = CrdsFilter::hash_as_u64(&value.value_hash);
+        let hash = CrdsFilter::hash_as_u64(value.value.hash());
         self.shard_mut(hash).swap_remove(&index).is_some()
     }
 
@@ -94,7 +94,7 @@ impl CrdsShards {
         assert_eq!(indices, (0..crds.len()).collect::<Vec<_>>());
         for (shard_index, shard) in self.shards.iter().enumerate() {
             for (&index, &hash) in shard {
-                assert_eq!(hash, CrdsFilter::hash_as_u64(&crds[index].value_hash));
+                assert_eq!(hash, CrdsFilter::hash_as_u64(crds[index].value.hash()));
                 assert_eq!(
                     shard_index as u64,
                     hash.checked_shr(64 - self.shard_bits).unwrap_or(0)
@@ -155,7 +155,7 @@ mod test {
     // Returns true if the first mask_bits most significant bits of hash is the
     // same as the given bit mask.
     fn check_mask(value: &VersionedCrdsValue, mask: u64, mask_bits: u32) -> bool {
-        let hash = CrdsFilter::hash_as_u64(&value.value_hash);
+        let hash = CrdsFilter::hash_as_u64(value.value.hash());
         let ones = (!0u64).checked_shr(mask_bits).unwrap_or(0u64);
         (hash | ones) == (mask | ones)
     }
@@ -217,7 +217,7 @@ mod test {
         }
         // Existing hash values.
         for (index, value) in values.iter().enumerate() {
-            let mask = CrdsFilter::hash_as_u64(&value.value_hash);
+            let mask = CrdsFilter::hash_as_u64(value.value.hash());
             let hits: Vec<_> = shards.find(mask, 64).collect();
             assert_eq!(hits, vec![index]);
         }

gossip/src/crds_value.rs

@@ -7,8 +7,10 @@ use {
     },
     bincode::serialize,
     rand::Rng,
+    serde::de::{Deserialize, Deserializer},
     solana_sanitize::{Sanitize, SanitizeError},
     solana_sdk::{
+        hash::Hash,
         pubkey::Pubkey,
         signature::{Keypair, Signable, Signature, Signer},
     },
@@ -17,10 +19,12 @@ use {
 
 /// CrdsValue that is replicated across the cluster
 #[cfg_attr(feature = "frozen-abi", derive(AbiExample))]
-#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
+#[derive(Serialize, Clone, Debug, PartialEq, Eq)]
 pub struct CrdsValue {
     signature: Signature,
     data: CrdsData,
+    #[serde(skip_serializing)]
+    hash: Hash, // Sha256 hash of [signature, data].
 }
 
 impl Sanitize for CrdsValue {
@@ -98,14 +102,23 @@ impl CrdsValue {
     pub fn new(data: CrdsData, keypair: &Keypair) -> Self {
         let bincode_serialized_data = bincode::serialize(&data).unwrap();
         let signature = keypair.sign_message(&bincode_serialized_data);
-        Self { signature, data }
+        let hash = solana_sdk::hash::hashv(&[signature.as_ref(), &bincode_serialized_data]);
+        Self {
+            signature,
+            data,
+            hash,
+        }
     }
 
     #[cfg(test)]
     pub(crate) fn new_unsigned(data: CrdsData) -> Self {
+        let bincode_serialized_data = bincode::serialize(&data).unwrap();
+        let signature = Signature::default();
+        let hash = solana_sdk::hash::hashv(&[signature.as_ref(), &bincode_serialized_data]);
         Self {
-            signature: Signature::default(),
+            signature,
             data,
+            hash,
         }
     }
 
@@ -134,6 +147,11 @@ impl CrdsValue {
         &self.data
     }
 
+    #[inline]
+    pub(crate) fn hash(&self) -> &Hash {
+        &self.hash
+    }
+
     /// Totally unsecure unverifiable wallclock of the node that generated this message
     /// Latest wallclock is always picked.
     /// This is used to time out push messages.
@@ -194,6 +212,29 @@ impl CrdsValue {
     }
 }
 
+// Manual implementation of Deserialize for CrdsValue in order to populate
+// CrdsValue.hash which is skipped in serialization.
+impl<'de> Deserialize<'de> for CrdsValue {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        #[derive(Deserialize)]
+        struct CrdsValue {
+            signature: Signature,
+            data: CrdsData,
+        }
+        let CrdsValue { signature, data } = CrdsValue::deserialize(deserializer)?;
+        let bincode_serialized_data = bincode::serialize(&data).unwrap();
+        let hash = solana_sdk::hash::hashv(&[signature.as_ref(), &bincode_serialized_data]);
+        Ok(Self {
+            signature,
+            data,
+            hash,
+        })
+    }
+}
+
 #[cfg(test)]
 mod test {
     use {