Skip to content

Commit

Permalink
Faster dedup v1.8 (#22619)
Browse files Browse the repository at this point in the history 
* Faster dedup
  • Loading branch information
@aeyakovenko
aeyakovenko committed Jan 21, 2022
1 parent 59eee75 commit edf1954
Show file tree
Hide file tree
Showing 6 changed files with 197 additions and 93 deletions.
31 changes: 19 additions & 12 deletions Cargo.lock
View file

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

33 changes: 11 additions & 22 deletions core/src/sigverify_stage.rs
View file
Expand Up @@ -7,13 +7,11 @@

use {
crate::sigverify,
core::time::Duration,
crossbeam_channel::{SendError, Sender as CrossbeamSender},
itertools::Itertools,
solana_bloom::bloom::{AtomicBloom, Bloom},
solana_measure::measure::Measure,
solana_perf::packet::PacketBatch,
solana_perf::sigverify::dedup_packets,
solana_perf::sigverify::Deduper,
solana_sdk::timing,
solana_streamer::streamer::{self, PacketBatchReceiver, StreamerError},
std::{
Expand Down Expand Up @@ -215,7 +213,7 @@ impl SigVerifyStage {
}

fn verifier<T: SigVerifier>(
bloom: &AtomicBloom<&[u8]>,
deduper: &Deduper,
recvr: &PacketBatchReceiver,
sendr: &CrossbeamSender<Vec<PacketBatch>>,
verifier: &T,
Expand All @@ -231,15 +229,15 @@ impl SigVerifyStage {
);

let mut dedup_time = Measure::start("sigverify_dedup_time");
let dedup_fail = dedup_packets(bloom, &mut batches) as usize;
let dedup_fail = deduper.dedup_packets(&mut batches) as usize;
dedup_time.stop();
let valid_packets = num_packets.saturating_sub(dedup_fail);
let num_unique = num_packets.saturating_sub(dedup_fail);

let mut discard_time = Measure::start("sigverify_discard_time");
if valid_packets > MAX_SIGVERIFY_BATCH {
if num_unique > MAX_SIGVERIFY_BATCH {
Self::discard_excess_packets(&mut batches, MAX_SIGVERIFY_BATCH)
};
let excess_fail = valid_packets.saturating_sub(MAX_SIGVERIFY_BATCH);
let excess_fail = num_unique.saturating_sub(MAX_SIGVERIFY_BATCH);
discard_time.stop();

let mut verify_batch_time = Measure::start("sigverify_batch_time");
Expand Down Expand Up @@ -298,25 +296,16 @@ impl SigVerifyStage {
let verifier = verifier.clone();
let mut stats = SigVerifierStats::default();
let mut last_print = Instant::now();
const MAX_BLOOM_AGE: Duration = Duration::from_millis(2_000);
const MAX_BLOOM_ITEMS: usize = 1_000_000;
const MAX_BLOOM_FAIL: f64 = 0.0001;
const MAX_BLOOM_BITS: usize = 8 << 22;
const MAX_DEDUPER_AGE_MS: u64 = 2_000;
const MAX_DEDUPER_ITEMS: u32 = 1_000_000;
Builder::new()
.name("solana-verifier".to_string())
.spawn(move || {
let mut bloom =
Bloom::random(MAX_BLOOM_ITEMS, MAX_BLOOM_FAIL, MAX_BLOOM_BITS).into();
let mut bloom_age = Instant::now();
let mut deduper = Deduper::new(MAX_DEDUPER_ITEMS, MAX_DEDUPER_AGE_MS);
loop {
let now = Instant::now();
if now.duration_since(bloom_age) > MAX_BLOOM_AGE {
bloom =
Bloom::random(MAX_BLOOM_ITEMS, MAX_BLOOM_FAIL, MAX_BLOOM_BITS).into();
bloom_age = now;
}
deduper.reset();
if let Err(e) = Self::verifier(
&bloom,
&deduper,
&packet_receiver,
&verified_sender,
&verifier,
Expand Down
1 change: 1 addition & 0 deletions perf/Cargo.toml
View file
Expand Up @@ -12,6 +12,7 @@ edition = "2018"
[dependencies]
bincode = "1.3.1"
curve25519-dalek = { version = "2" }
ahash = "0.7.6"
dlopen = "0.1.8"
dlopen_derive = "0.1.4"
lazy_static = "1.4.0"
Expand Down
54 changes: 37 additions & 17 deletions perf/benches/dedup.rs
View file
Expand Up @@ -5,14 +5,15 @@ extern crate test;

use {
rand::prelude::*,
solana_bloom::bloom::{AtomicBloom, Bloom},
solana_perf::{
packet::{to_packet_batches, PacketBatch},
sigverify,
},
test::Bencher,
};

const NUM: usize = 4096;

fn test_packet_with_size(size: usize, rng: &mut ThreadRng) -> Vec<u8> {
// subtract 8 bytes because the length will get serialized as well
(0..size.checked_sub(8).unwrap())
Expand All @@ -22,19 +23,14 @@ fn test_packet_with_size(size: usize, rng: &mut ThreadRng) -> Vec<u8> {

fn do_bench_dedup_packets(bencher: &mut Bencher, mut batches: Vec<PacketBatch>) {
// verify packets
let mut bloom: AtomicBloom<&[u8]> = Bloom::random(1_000_000, 0.0001, 8 << 22).into();
let mut deduper = sigverify::Deduper::new(1_000_000, 2_000);
bencher.iter(|| {
// bench
sigverify::dedup_packets(&bloom, &mut batches);

// reset
bloom.clear_for_tests();
batches.iter_mut().for_each(|batch| {
batch.packets.iter_mut().for_each(|p| {
p.meta.discard = false;
})
});
})
let _ans = deduper.dedup_packets(&mut batches);
deduper.reset();
batches
.iter_mut()
.for_each(|b| b.packets.iter_mut().for_each(|p| p.meta.discard = false));
});
}

#[bench]
Expand All @@ -45,7 +41,7 @@ fn bench_dedup_same_small_packets(bencher: &mut Bencher) {

let batches = to_packet_batches(
&std::iter::repeat(small_packet)
.take(4096)
.take(NUM)
.collect::<Vec<_>>(),
128,
);
Expand All @@ -60,7 +56,7 @@ fn bench_dedup_same_big_packets(bencher: &mut Bencher) {
let big_packet = test_packet_with_size(1024, &mut rng);

let batches = to_packet_batches(
&std::iter::repeat(big_packet).take(4096).collect::<Vec<_>>(),
&std::iter::repeat(big_packet).take(NUM).collect::<Vec<_>>(),
128,
);

Expand All @@ -73,7 +69,7 @@ fn bench_dedup_diff_small_packets(bencher: &mut Bencher) {
let mut rng = rand::thread_rng();

let batches = to_packet_batches(
&(0..4096)
&(0..NUM)
.map(|_| test_packet_with_size(128, &mut rng))
.collect::<Vec<_>>(),
128,
Expand All @@ -88,11 +84,35 @@ fn bench_dedup_diff_big_packets(bencher: &mut Bencher) {
let mut rng = rand::thread_rng();

let batches = to_packet_batches(
&(0..4096)
&(0..NUM)
.map(|_| test_packet_with_size(1024, &mut rng))
.collect::<Vec<_>>(),
128,
);

do_bench_dedup_packets(bencher, batches);
}

#[bench]
#[ignore]
fn bench_dedup_baseline(bencher: &mut Bencher) {
let mut rng = rand::thread_rng();

let batches = to_packet_batches(
&(0..0)
.map(|_| test_packet_with_size(128, &mut rng))
.collect::<Vec<_>>(),
128,
);

do_bench_dedup_packets(bencher, batches);
}

#[bench]
#[ignore]
fn bench_dedup_reset(bencher: &mut Bencher) {
let mut deduper = sigverify::Deduper::new(1_000_000, 0);
bencher.iter(|| {
deduper.reset();
});
}

0 comments on commit edf1954

Please sign in to comment.