diff --git a/Cargo.lock b/Cargo.lock index f138165765..63454ef4ab 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -34,6 +34,17 @@ dependencies = [ "const-random", ] +[[package]] +name = "ahash" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "865f8b0b3fced577b7df82e9b0eb7609595d7209c0b39e78d0646672e244b1b1" +dependencies = [ + "getrandom 0.2.0", + "lazy_static", + "version_check 0.9.2", +] + [[package]] name = "aho-corasick" version = "0.7.10" @@ -607,7 +618,7 @@ version = "0.1.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "25e4c606eb459dd29f7c57b2e0879f2b6f14ee130918c2b78ccb58a9624e6c7a" dependencies = [ - "getrandom", + "getrandom 0.1.14", "proc-macro-hack", ] @@ -815,7 +826,7 @@ version = "3.11.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0f260e2fc850179ef410018660006951c1b55b79e8087e87111a2c388994b9b5" dependencies = [ - "ahash", + "ahash 0.3.8", "cfg-if 0.1.10", "num_cpus", ] @@ -1293,6 +1304,17 @@ dependencies = [ "wasi", ] +[[package]] +name = "getrandom" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee8025cf36f917e6a52cce185b7c7177689b838b7ec138364e50cc2277a56cf4" +dependencies = [ + "cfg-if 0.1.10", + "libc", + "wasi", +] + [[package]] name = "gimli" version = "0.21.0" @@ -1405,7 +1427,7 @@ version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e91b62f79061a0bc2e046024cb7ba44b08419ed238ecbd9adbd787434b9e8c25" dependencies = [ - "ahash", + "ahash 0.3.8", "autocfg 1.0.0", ] @@ -2832,7 +2854,7 @@ version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6a6b1679d49b24bbfe0c803429aa1874472f50d9b363131f0e89fc356b544d03" dependencies = [ - "getrandom", + "getrandom 0.1.14", "libc", "rand_chacha 0.2.2", "rand_core 0.5.1", @@ -2881,7 +2903,7 @@ version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "90bde5296fc891b0cef12a6d03ddccc162ce7b2aff54160af9338f8d40df6d19" dependencies = [ - "getrandom", + "getrandom 0.1.14", ] [[package]] @@ -3019,7 +3041,7 @@ version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09b23093265f8d200fa7b4c2c76297f47e681c655f6f1285a8780d6a022f7431" dependencies = [ - "getrandom", + "getrandom 0.1.14", "redox_syscall", "rust-argon2", ] @@ -3911,6 +3933,7 @@ dependencies = [ name = "solana-core" version = "1.4.15" dependencies = [ + "ahash 0.6.1", "base64 0.12.3", "bincode", "bs58", diff --git a/core/Cargo.toml b/core/Cargo.toml index 610c36faff..f5c684cd82 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -14,6 +14,7 @@ edition = "2018" codecov = { repository = "solana-labs/solana", branch = "master", service = "github" } [dependencies] +ahash = "0.6.1" base64 = "0.12.3" bincode = "1.3.1" bv = { version = "0.11.1", features = ["serde"] } diff --git a/core/src/shred_fetch_stage.rs b/core/src/shred_fetch_stage.rs index 7eace281e5..f4cc2cb173 100644 --- a/core/src/shred_fetch_stage.rs +++ b/core/src/shred_fetch_stage.rs @@ -1,6 +1,10 @@ //! The `shred_fetch_stage` pulls shreds from UDP sockets and sends it to a channel. -use bv::BitVec; +use ahash::AHasher; +use lru::LruCache; +use rand::{thread_rng, Rng}; +use std::hash::Hasher; + use solana_ledger::blockstore::MAX_DATA_SHREDS_PER_SLOT; use solana_ledger::shred::{ CODING_SHRED, DATA_SHRED, OFFSET_OF_SHRED_INDEX, OFFSET_OF_SHRED_SLOT, OFFSET_OF_SHRED_TYPE, @@ -10,9 +14,8 @@ use solana_perf::cuda_runtime::PinnedVec; use solana_perf::packet::{limited_deserialize, Packet, PacketsRecycler}; use solana_perf::recycler::Recycler; use solana_runtime::bank_forks::BankForks; -use solana_sdk::clock::Slot; +use solana_sdk::clock::{Slot, DEFAULT_MS_PER_SLOT}; use solana_streamer::streamer::{self, PacketReceiver, PacketSender}; -use std::collections::HashMap; use std::net::UdpSocket; use std::sync::atomic::AtomicBool; use std::sync::mpsc::channel; @@ -21,7 +24,8 @@ use std::sync::RwLock; use std::thread::{self, Builder, JoinHandle}; use std::time::Instant; -pub type ShredsReceived = HashMap<(Slot, u8), BitVec>; +const DEFAULT_LRU_SIZE: usize = 10_000; +pub type ShredsReceived = LruCache; #[derive(Default)] struct ShredFetchStats { @@ -73,11 +77,12 @@ impl ShredFetchStage { last_slot: Slot, slots_per_epoch: u64, modify: &F, + seeds: (u128, u128), ) where F: Fn(&mut Packet), { p.meta.discard = true; - if let Some((slot, index)) = Self::get_slot_index(p, stats) { + if let Some((slot, _index)) = Self::get_slot_index(p, stats) { // Seems reasonable to limit shreds to 2 epochs away if slot > last_root && slot < (last_slot + 2 * slots_per_epoch) @@ -86,16 +91,15 @@ impl ShredFetchStage { let shred_type = p.data[OFFSET_OF_SHRED_TYPE]; if shred_type == DATA_SHRED || shred_type == CODING_SHRED { // Shred filter - let slot_received = - shreds_received - .entry((slot, shred_type)) - .or_insert_with(|| { - BitVec::new_fill(false, MAX_DATA_SHREDS_PER_SLOT as u64) - }); - if !slot_received.get(index.into()) { + + let mut hasher = AHasher::new_with_keys(seeds.0, seeds.1); + hasher.write(&p.data[0..p.meta.size]); + let hash = hasher.finish(); + + if shreds_received.get(&hash).is_none() { + shreds_received.put(hash, ()); p.meta.discard = false; modify(p); - slot_received.set(index.into(), true); } else { stats.duplicate_shred += 1; } @@ -116,8 +120,8 @@ impl ShredFetchStage { ) where F: Fn(&mut Packet), { - let mut shreds_received = ShredsReceived::default(); - let mut last_cleared = Instant::now(); + let mut shreds_received = LruCache::new(DEFAULT_LRU_SIZE); + let mut last_updated = Instant::now(); // In the case of bank_forks=None, setup to accept any slot range let mut last_root = 0; @@ -126,11 +130,13 @@ impl ShredFetchStage { let mut last_stats = Instant::now(); let mut stats = ShredFetchStats::default(); + let mut seeds = (thread_rng().gen::(), thread_rng().gen::()); while let Some(mut p) = recvr.iter().next() { - if last_cleared.elapsed().as_millis() > 200 { + if last_updated.elapsed().as_millis() as u64 > DEFAULT_MS_PER_SLOT { + last_updated = Instant::now(); + seeds = (thread_rng().gen::(), thread_rng().gen::()); shreds_received.clear(); - last_cleared = Instant::now(); if let Some(bank_forks) = bank_forks.as_ref() { let bank_forks_r = bank_forks.read().unwrap(); last_root = bank_forks_r.root(); @@ -150,6 +156,7 @@ impl ShredFetchStage { last_slot, slots_per_epoch, &modify, + seeds, ); }); if last_stats.elapsed().as_millis() > 1000 { @@ -215,15 +222,15 @@ impl ShredFetchStage { ) -> Self { let recycler: PacketsRecycler = Recycler::warmed(100, 1024); - let tvu_threads = sockets.into_iter().map(|socket| { - streamer::receiver( - socket, - &exit, - sender.clone(), - recycler.clone(), - "shred_fetch_stage", - ) - }); + let (mut tvu_threads, tvu_filter) = Self::packet_modifier( + sockets, + &exit, + sender.clone(), + recycler.clone(), + bank_forks.clone(), + "shred_fetch", + |_| {}, + ); let (tvu_forwards_threads, fwd_thread_hdl) = Self::packet_modifier( forward_sockets, @@ -239,20 +246,21 @@ impl ShredFetchStage { vec![repair_socket], &exit, sender.clone(), - recycler.clone(), + recycler, bank_forks, "shred_fetch_repair", |p| p.meta.repair = true, ); - let mut thread_hdls: Vec<_> = tvu_threads - .chain(tvu_forwards_threads.into_iter()) - .collect(); - thread_hdls.extend(repair_receiver.into_iter()); - thread_hdls.push(fwd_thread_hdl); - thread_hdls.push(repair_handler); + tvu_threads.extend(tvu_forwards_threads.into_iter()); + tvu_threads.extend(repair_receiver.into_iter()); + tvu_threads.push(tvu_filter); + tvu_threads.push(fwd_thread_hdl); + tvu_threads.push(repair_handler); - Self { thread_hdls } + Self { + thread_hdls: tvu_threads, + } } pub fn join(self) -> thread::Result<()> { @@ -271,7 +279,7 @@ mod tests { #[test] fn test_data_code_same_index() { solana_logger::setup(); - let mut shreds_received = ShredsReceived::default(); + let mut shreds_received = LruCache::new(DEFAULT_LRU_SIZE); let mut packet = Packet::default(); let mut stats = ShredFetchStats::default(); @@ -279,6 +287,8 @@ mod tests { let shred = Shred::new_from_data(slot, 3, 0, None, true, true, 0, 0, 0); shred.copy_to_packet(&mut packet); + let seeds = (thread_rng().gen::(), thread_rng().gen::()); + let last_root = 0; let last_slot = 100; let slots_per_epoch = 10; @@ -290,6 +300,7 @@ mod tests { last_slot, slots_per_epoch, &|_p| {}, + seeds, ); assert!(!packet.meta.discard); @@ -304,6 +315,7 @@ mod tests { last_slot, slots_per_epoch, &|_p| {}, + seeds, ); assert!(!packet.meta.discard); } @@ -311,12 +323,13 @@ mod tests { #[test] fn test_shred_filter() { solana_logger::setup(); - let mut shreds_received = ShredsReceived::default(); + let mut shreds_received = LruCache::new(DEFAULT_LRU_SIZE); let mut packet = Packet::default(); let mut stats = ShredFetchStats::default(); let last_root = 0; let last_slot = 100; let slots_per_epoch = 10; + let seeds = (thread_rng().gen::(), thread_rng().gen::()); // packet size is 0, so cannot get index ShredFetchStage::process_packet( &mut packet, @@ -326,6 +339,7 @@ mod tests { last_slot, slots_per_epoch, &|_p| {}, + seeds, ); assert_eq!(stats.index_overrun, 1); assert!(packet.meta.discard); @@ -341,6 +355,7 @@ mod tests { last_slot, slots_per_epoch, &|_p| {}, + seeds, ); assert!(packet.meta.discard); @@ -353,6 +368,7 @@ mod tests { last_slot, slots_per_epoch, &|_p| {}, + seeds, ); assert!(!packet.meta.discard); @@ -365,6 +381,7 @@ mod tests { last_slot, slots_per_epoch, &|_p| {}, + seeds, ); assert!(packet.meta.discard); @@ -380,6 +397,7 @@ mod tests { last_slot, slots_per_epoch, &|_p| {}, + seeds, ); assert!(packet.meta.discard); @@ -394,6 +412,7 @@ mod tests { last_slot, slots_per_epoch, &|_p| {}, + seeds, ); assert!(packet.meta.discard); }