From 6686e09c98976cc8bf0b7e80dab33db31e0e1e5b Mon Sep 17 00:00:00 2001 From: behzad nouri Date: Thu, 19 Dec 2024 19:50:57 -0600 Subject: [PATCH] dedups shreds by common-header instead of entire payload Shreds in the retransmit stage: * don't have repair nonce (repaired shreds are not retransmitted). * are already resigned by this node as the retransmitter. * have their leader's signature verified. Therefore in order to dedup shreds, it suffices to compare: (signature, slot, shred-index, shred-type) Because ShredCommonHeader already includes all of the above tuple, the rest of the payload can be skipped. --- ledger/src/shred.rs | 5 +++++ turbine/src/retransmit_stage.rs | 14 +++++++++++++- 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/ledger/src/shred.rs b/ledger/src/shred.rs index c9039091813575..7b721c9407483f 100644 --- a/ledger/src/shred.rs +++ b/ledger/src/shred.rs @@ -638,6 +638,11 @@ pub mod layout { packet.buffer_mut().get_mut(..size) } + #[inline] + pub fn get_common_header_bytes(shred: &[u8]) -> Option<&[u8]> { + shred.get(..SIZE_OF_COMMON_SHRED_HEADER) + } + pub(crate) fn get_signature(shred: &[u8]) -> Option { shred .get(..SIZE_OF_SIGNATURE) diff --git a/turbine/src/retransmit_stage.rs b/turbine/src/retransmit_stage.rs index db9d1f96662a16..3ee4662e5d7e21 100644 --- a/turbine/src/retransmit_stage.rs +++ b/turbine/src/retransmit_stage.rs @@ -165,10 +165,22 @@ impl ShredDeduper { .maybe_reset(rng, false_positive_rate, reset_cycle); } + // Returns true if the shred is duplicate and should be discarded. + #[must_use] fn dedup(&self, key: ShredId, shred: &[u8], max_duplicate_count: usize) -> bool { + // Shreds in the retransmit stage: + // * don't have repair nonce (repaired shreds are not retransmitted). + // * are already resigned by this node as the retransmitter. + // * have their leader's signature verified. + // Therefore in order to dedup shreds, it suffices to compare: + // (signature, slot, shred-index, shred-type) + // Because ShredCommonHeader already includes all of the above tuple, + // the rest of the payload can be skipped. // In order to detect duplicate blocks across cluster, we retransmit // max_duplicate_count different shreds for each ShredId. - self.deduper.dedup(shred) + shred::layout::get_common_header_bytes(shred) + .map(|header| self.deduper.dedup(header)) + .unwrap_or(true) || (0..max_duplicate_count).all(|i| self.shred_id_filter.dedup(&(key, i))) } }