Skip to content
This repository was archived by the owner on Jan 22, 2025. It is now read-only.

Bitwise compress incomplete epoch slots #8341

Merged
pgarg66 merged 1 commit into from
Feb 20, 2020
Merged

Bitwise compress incomplete epoch slots #8341

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
156 changes: 88 additions & 68 deletions core/src/cluster_info.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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::{
Expand Down Expand Up @@ -69,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,
Expand Down Expand Up @@ -308,7 +311,7 @@ impl ClusterInfo {
)
}

pub fn compress_incomplete_slots(incomplete_slots: &BTreeSet<Slot>) -> (Slot, Vec<u8>) {
pub fn compress_incomplete_slots(incomplete_slots: &BTreeSet<Slot>) -> EpochIncompleteSlots {
if !incomplete_slots.is_empty() {
let first_slot = incomplete_slots
.iter()
Expand All @@ -318,37 +321,52 @@ impl ClusterInfo {
.iter()
.next_back()
.expect("expected to find last slot");
let mut uncompressed = vec![0u8; (last_slot.saturating_sub(*first_slot) + 1) as usize];
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| {
uncompressed[slot.saturating_sub(*first_slot) as usize] = 1;
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>, _>>()
{
(*first_slot, compressed)
} else {
(0, vec![])
return EpochIncompleteSlots {
first: *first_slot,
compressed_list: compressed,
};
}
} else {
(0, vec![])
}
EpochIncompleteSlots::default()
}

pub fn decompress_incomplete_slots(first_slot: u64, compressed: &[u8]) -> BTreeSet<Slot> {
pub fn decompress_incomplete_slots(slots: &EpochIncompleteSlots) -> BTreeSet<Slot> {
let mut old_incomplete_slots: BTreeSet<Slot> = BTreeSet::new();

if let Ok(decompressed) = compressed
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 == 1 {
old_incomplete_slots.insert(first_slot + i as u64);
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);
}
})
}
})
}
Expand All @@ -364,19 +382,13 @@ impl ClusterInfo {
slots: BTreeSet<Slot>,
incomplete_slots: &BTreeSet<Slot>,
) {
let (first_missing_slot, compressed_map) =
Self::compress_incomplete_slots(incomplete_slots);
let compressed = Self::compress_incomplete_slots(incomplete_slots);
let now = timestamp();
let entry = CrdsValue::new_signed(
CrdsData::EpochSlots(EpochSlots::new(
id,
root,
min,
slots,
first_missing_slot,
compressed_map,
now,
)),
CrdsData::EpochSlots(
0,
EpochSlots::new(id, root, min, slots, vec![compressed], now),
),
&self.keypair,
);
self.gossip
Expand Down Expand Up @@ -2189,15 +2201,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,
first_missing: 0,
stash: vec![],
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);
}

Expand All @@ -2208,15 +2222,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(),
first_missing: 0,
stash: vec![],
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 {
Expand All @@ -2228,15 +2244,17 @@ mod tests {
desired_size
);
}
value.data = CrdsData::EpochSlots(EpochSlots {
from: Pubkey::default(),
root: 0,
lowest: 0,
slots,
first_missing: 0,
stash: vec![],
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()]);
Expand Down Expand Up @@ -2376,15 +2394,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(),
let value = CrdsValue::new_unsigned(CrdsData::EpochSlots(
0,
vec![],
timestamp(),
)));
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
Expand Down Expand Up @@ -2450,26 +2470,26 @@ mod tests {
let mut incomplete_slots: BTreeSet<Slot> = BTreeSet::new();

assert_eq!(
(0, vec![]),
EpochIncompleteSlots::default(),
ClusterInfo::compress_incomplete_slots(&incomplete_slots)
);

incomplete_slots.insert(100);
let (first, compressed) = ClusterInfo::compress_incomplete_slots(&incomplete_slots);
assert_eq!(100, first);
let decompressed = ClusterInfo::decompress_incomplete_slots(first, &compressed);
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 (first, compressed) = ClusterInfo::compress_incomplete_slots(&incomplete_slots);
assert_eq!(100, first);
let decompressed = ClusterInfo::decompress_incomplete_slots(first, &compressed);
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 (first, compressed) = ClusterInfo::compress_incomplete_slots(&incomplete_slots);
assert_eq!(80, first);
let decompressed = ClusterInfo::decompress_incomplete_slots(first, &compressed);
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);
}
}
47 changes: 21 additions & 26 deletions core/src/crds_value.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@ use std::{
pub type VoteIndex = u8;
pub const MAX_VOTES: VoteIndex = 32;

pub type EpochSlotIndex = u8;

/// CrdsValue that is replicated across the cluster
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub struct CrdsValue {
Expand Down Expand Up @@ -58,7 +60,13 @@ impl Signable for CrdsValue {
pub enum CrdsData {
ContactInfo(ContactInfo),
Vote(VoteIndex, Vote),
EpochSlots(EpochSlots),
EpochSlots(EpochSlotIndex, EpochSlots),
}

#[derive(Serialize, Deserialize, Clone, Debug, Default, PartialEq)]
pub struct EpochIncompleteSlots {
pub first: Slot,
pub compressed_list: Vec<u8>,
}

#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
Expand All @@ -67,8 +75,7 @@ pub struct EpochSlots {
pub root: Slot,
pub lowest: Slot,
pub slots: BTreeSet<Slot>,
pub first_missing: Slot,
pub stash: Vec<u8>,
pub stash: Vec<EpochIncompleteSlots>,
pub wallclock: u64,
}

Expand All @@ -78,16 +85,14 @@ impl EpochSlots {
root: Slot,
lowest: Slot,
slots: BTreeSet<Slot>,
first_missing: Slot,
stash: Vec<u8>,
stash: Vec<EpochIncompleteSlots>,
wallclock: u64,
) -> Self {
Self {
from,
root,
lowest,
slots,
first_missing,
stash,
wallclock,
}
Expand Down Expand Up @@ -160,21 +165,21 @@ impl CrdsValue {
match &self.data {
CrdsData::ContactInfo(contact_info) => contact_info.wallclock,
CrdsData::Vote(_, vote) => vote.wallclock,
CrdsData::EpochSlots(vote) => vote.wallclock,
CrdsData::EpochSlots(_, vote) => vote.wallclock,
}
}
pub fn pubkey(&self) -> Pubkey {
match &self.data {
CrdsData::ContactInfo(contact_info) => contact_info.id,
CrdsData::Vote(_, vote) => vote.from,
CrdsData::EpochSlots(slots) => slots.from,
CrdsData::EpochSlots(_, slots) => slots.from,
}
}
pub fn label(&self) -> CrdsValueLabel {
match &self.data {
CrdsData::ContactInfo(_) => CrdsValueLabel::ContactInfo(self.pubkey()),
CrdsData::Vote(ix, _) => CrdsValueLabel::Vote(*ix, self.pubkey()),
CrdsData::EpochSlots(_) => CrdsValueLabel::EpochSlots(self.pubkey()),
CrdsData::EpochSlots(_, _) => CrdsValueLabel::EpochSlots(self.pubkey()),
}
}
pub fn contact_info(&self) -> Option<&ContactInfo> {
Expand All @@ -199,7 +204,7 @@ impl CrdsValue {

pub fn epoch_slots(&self) -> Option<&EpochSlots> {
match &self.data {
CrdsData::EpochSlots(slots) => Some(slots),
CrdsData::EpochSlots(_, slots) => Some(slots),
_ => None,
}
}
Expand Down Expand Up @@ -283,15 +288,10 @@ mod test {
let key = v.clone().vote().unwrap().from;
assert_eq!(v.label(), CrdsValueLabel::Vote(0, key));

let v = CrdsValue::new_unsigned(CrdsData::EpochSlots(EpochSlots::new(
Pubkey::default(),
0,
let v = CrdsValue::new_unsigned(CrdsData::EpochSlots(
0,
BTreeSet::new(),
0,
vec![],
0,
)));
EpochSlots::new(Pubkey::default(), 0, 0, BTreeSet::new(), vec![], 0),
));
assert_eq!(v.wallclock(), 0);
let key = v.clone().epoch_slots().unwrap().from;
assert_eq!(v.label(), CrdsValueLabel::EpochSlots(key));
Expand All @@ -312,15 +312,10 @@ mod test {
));
verify_signatures(&mut v, &keypair, &wrong_keypair);
let btreeset: BTreeSet<Slot> = vec![1, 2, 3, 6, 8].into_iter().collect();
v = CrdsValue::new_unsigned(CrdsData::EpochSlots(EpochSlots::new(
keypair.pubkey(),
0,
0,
btreeset,
v = CrdsValue::new_unsigned(CrdsData::EpochSlots(
0,
vec![],
timestamp(),
)));
EpochSlots::new(keypair.pubkey(), 0, 0, btreeset, vec![], timestamp()),
));
verify_signatures(&mut v, &keypair, &wrong_keypair);
}

Expand Down