[V0.23] Cherry picks from master

core/Cargo.toml

@@ -18,6 +18,7 @@ bincode = "1.2.1"
 bs58 = "0.3.0"
 byteorder = "1.3.2"
 chrono = { version = "0.4.10", features = ["serde"] }
+compression = "0.1.5"
 core_affinity = "0.5.10"
 crc = { version = "1.8.1", optional = true }
 crossbeam-channel = "0.3"

core/src/cluster_info.rs

@@ -12,6 +12,7 @@
 //! * layer 2 - Everyone else, if layer 1 is `2^10`, layer 2 should be able to fit `2^20` number of nodes.
 //!
 //! Bank needs to provide an interface for us to query the stake weight
+use crate::crds_value::EpochIncompleteSlots;
 use crate::packet::limited_deserialize;
 use crate::streamer::{PacketReceiver, PacketSender};
 use crate::{
@@ -26,6 +27,7 @@ use crate::{
     weighted_shuffle::{weighted_best, weighted_shuffle},
 };
 use bincode::{serialize, serialized_size};
+use compression::prelude::*;
 use core::cmp;
 use itertools::Itertools;
 use solana_ledger::{bank_forks::BankForks, staking_utils};
@@ -68,6 +70,8 @@ const MAX_PROTOCOL_PAYLOAD_SIZE: u64 = PACKET_DATA_SIZE as u64 - MAX_PROTOCOL_HE
 /// The largest protocol header size
 const MAX_PROTOCOL_HEADER_SIZE: u64 = 214;
 
+const NUM_BITS_PER_BYTE: u64 = 8;
+
 #[derive(Debug, PartialEq, Eq)]
 pub enum ClusterInfoError {
     NoPeers,
@@ -307,10 +311,84 @@ impl ClusterInfo {
         )
     }
 
-    pub fn push_epoch_slots(&mut self, id: Pubkey, root: Slot, min: Slot, slots: BTreeSet<Slot>) {
+    pub fn compress_incomplete_slots(incomplete_slots: &BTreeSet<Slot>) -> EpochIncompleteSlots {
+        if !incomplete_slots.is_empty() {
+            let first_slot = incomplete_slots
+                .iter()
+                .next()
+                .expect("expected to find at least one slot");
+            let last_slot = incomplete_slots
+                .iter()
+                .next_back()
+                .expect("expected to find last slot");
+            let num_uncompressed_bits = last_slot.saturating_sub(*first_slot) + 1;
+            let num_uncompressed_bytes = if num_uncompressed_bits % NUM_BITS_PER_BYTE > 0 {
+                1
+            } else {
+                0
+            } + num_uncompressed_bits / NUM_BITS_PER_BYTE;
+            let mut uncompressed = vec![0u8; num_uncompressed_bytes as usize];
+            incomplete_slots.iter().for_each(|slot| {
+                let offset_from_first_slot = slot.saturating_sub(*first_slot);
+                let index = offset_from_first_slot / NUM_BITS_PER_BYTE;
+                let bit_index = offset_from_first_slot % NUM_BITS_PER_BYTE;
+                uncompressed[index as usize] |= 1 << bit_index;
+            });
+            if let Ok(compressed) = uncompressed
+                .iter()
+                .cloned()
+                .encode(&mut GZipEncoder::new(), Action::Finish)
+                .collect::<std::result::Result<Vec<u8>, _>>()
+            {
+                return EpochIncompleteSlots {
+                    first: *first_slot,
+                    compressed_list: compressed,
+                };
+            }
+        }
+        EpochIncompleteSlots::default()
+    }
+
+    pub fn decompress_incomplete_slots(slots: &EpochIncompleteSlots) -> BTreeSet<Slot> {
+        let mut old_incomplete_slots: BTreeSet<Slot> = BTreeSet::new();
+
+        if let Ok(decompressed) = slots
+            .compressed_list
+            .iter()
+            .cloned()
+            .decode(&mut GZipDecoder::new())
+            .collect::<std::result::Result<Vec<u8>, _>>()
+        {
+            decompressed.iter().enumerate().for_each(|(i, val)| {
+                if *val != 0 {
+                    (0..8).for_each(|bit_index| {
+                        if (1 << bit_index & *val) != 0 {
+                            let slot = slots.first + i as u64 * NUM_BITS_PER_BYTE + bit_index;
+                            old_incomplete_slots.insert(slot as u64);
+                        }
+                    })
+                }
+            })
+        }
+
+        old_incomplete_slots
+    }
+
+    pub fn push_epoch_slots(
+        &mut self,
+        id: Pubkey,
+        root: Slot,
+        min: Slot,
+        slots: BTreeSet<Slot>,
+        incomplete_slots: &BTreeSet<Slot>,
+    ) {
+        let compressed = Self::compress_incomplete_slots(incomplete_slots);
         let now = timestamp();
         let entry = CrdsValue::new_signed(
-            CrdsData::EpochSlots(EpochSlots::new(id, root, min, slots, now)),
+            CrdsData::EpochSlots(
+                0,
+                EpochSlots::new(id, root, min, slots, vec![compressed], now),
+            ),
             &self.keypair,
         );
         self.gossip
@@ -2120,13 +2198,17 @@ mod tests {
         for i in 0..128 {
             btree_slots.insert(i);
         }
-        let value = CrdsValue::new_unsigned(CrdsData::EpochSlots(EpochSlots {
-            from: Pubkey::default(),
-            root: 0,
-            lowest: 0,
-            slots: btree_slots,
-            wallclock: 0,
-        }));
+        let value = CrdsValue::new_unsigned(CrdsData::EpochSlots(
+            0,
+            EpochSlots {
+                from: Pubkey::default(),
+                root: 0,
+                lowest: 0,
+                slots: btree_slots,
+                stash: vec![],
+                wallclock: 0,
+            },
+        ));
         test_split_messages(value);
     }
 
@@ -2137,13 +2219,17 @@ mod tests {
         let payload: Vec<CrdsValue> = vec![];
         let vec_size = serialized_size(&payload).unwrap();
         let desired_size = MAX_PROTOCOL_PAYLOAD_SIZE - vec_size;
-        let mut value = CrdsValue::new_unsigned(CrdsData::EpochSlots(EpochSlots {
-            from: Pubkey::default(),
-            root: 0,
-            lowest: 0,
-            slots: BTreeSet::new(),
-            wallclock: 0,
-        }));
+        let mut value = CrdsValue::new_unsigned(CrdsData::EpochSlots(
+            0,
+            EpochSlots {
+                from: Pubkey::default(),
+                root: 0,
+                lowest: 0,
+                slots: BTreeSet::new(),
+                stash: vec![],
+                wallclock: 0,
+            },
+        ));
 
         let mut i = 0;
         while value.size() <= desired_size {
@@ -2155,13 +2241,17 @@ mod tests {
                     desired_size
                 );
             }
-            value.data = CrdsData::EpochSlots(EpochSlots {
-                from: Pubkey::default(),
-                root: 0,
-                lowest: 0,
-                slots,
-                wallclock: 0,
-            });
+            value.data = CrdsData::EpochSlots(
+                0,
+                EpochSlots {
+                    from: Pubkey::default(),
+                    root: 0,
+                    lowest: 0,
+                    slots,
+                    stash: vec![],
+                    wallclock: 0,
+                },
+            );
             i += 1;
         }
         let split = ClusterInfo::split_gossip_messages(vec![value.clone()]);
@@ -2301,13 +2391,17 @@ mod tests {
             let other_node_pubkey = Pubkey::new_rand();
             let other_node = ContactInfo::new_localhost(&other_node_pubkey, timestamp());
             cluster_info.insert_info(other_node.clone());
-            let value = CrdsValue::new_unsigned(CrdsData::EpochSlots(EpochSlots::new(
-                other_node_pubkey,
-                peer_root,
-                peer_lowest,
-                BTreeSet::new(),
-                timestamp(),
-            )));
+            let value = CrdsValue::new_unsigned(CrdsData::EpochSlots(
+                0,
+                EpochSlots::new(
+                    other_node_pubkey,
+                    peer_root,
+                    peer_lowest,
+                    BTreeSet::new(),
+                    vec![],
+                    timestamp(),
+                ),
+            ));
             let _ = cluster_info.gossip.crds.insert(value, timestamp());
         }
         // only half the visible peers should be eligible to serve this repair
@@ -2367,4 +2461,32 @@ mod tests {
             serialized_size(&protocol).expect("unable to serialize gossip protocol") as usize;
         PACKET_DATA_SIZE - (protocol_size - filter_size)
     }
+
+    #[test]
+    fn test_compress_incomplete_slots() {
+        let mut incomplete_slots: BTreeSet<Slot> = BTreeSet::new();
+
+        assert_eq!(
+            EpochIncompleteSlots::default(),
+            ClusterInfo::compress_incomplete_slots(&incomplete_slots)
+        );
+
+        incomplete_slots.insert(100);
+        let compressed = ClusterInfo::compress_incomplete_slots(&incomplete_slots);
+        assert_eq!(100, compressed.first);
+        let decompressed = ClusterInfo::decompress_incomplete_slots(&compressed);
+        assert_eq!(incomplete_slots, decompressed);
+
+        incomplete_slots.insert(104);
+        let compressed = ClusterInfo::compress_incomplete_slots(&incomplete_slots);
+        assert_eq!(100, compressed.first);
+        let decompressed = ClusterInfo::decompress_incomplete_slots(&compressed);
+        assert_eq!(incomplete_slots, decompressed);
+
+        incomplete_slots.insert(80);
+        let compressed = ClusterInfo::compress_incomplete_slots(&incomplete_slots);
+        assert_eq!(80, compressed.first);
+        let decompressed = ClusterInfo::decompress_incomplete_slots(&compressed);
+        assert_eq!(incomplete_slots, decompressed);
+    }
 }