Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

try bounded channel for sigverify-retransmit #5091

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

try bounded channel for sigverify-retransmit #5091

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f612dd9
try bounded channel for sigverify-retransmit
Feb 27, 2025
67c5bbb
bump to 2048 slots to avoid any backpressure
Mar 2, 2025
847265e
formatting
Mar 7, 2025
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
7 changes: 5 additions & 2 deletions core/src/tvu.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ use {
window_service::{WindowService, WindowServiceChannels},
},
bytes::Bytes,
crossbeam_channel::{unbounded, Receiver, Sender},
crossbeam_channel::{bounded, unbounded, Receiver, Sender},
solana_client::connection_cache::ConnectionCache,
solana_geyser_plugin_manager::block_metadata_notifier_interface::BlockMetadataNotifierArc,
solana_gossip::{
Expand Down Expand Up @@ -191,7 +191,10 @@ impl Tvu {
);

let (verified_sender, verified_receiver) = unbounded();
let (retransmit_sender, retransmit_receiver) = unbounded();
// Allow for a max of 2048 batches of up to 1024 packets each (according to MAX_IOV).
// In reality this holds about 20K shreds since most batches are not full.
let (retransmit_sender, retransmit_receiver) = bounded(2048);

let shred_sigverify = solana_turbine::sigverify_shreds::spawn_shred_sigverify(
cluster_info.clone(),
bank_forks.clone(),
Expand Down
18 changes: 17 additions & 1 deletion turbine/src/sigverify_shreds.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -265,7 +265,16 @@ fn run_shred_sigverify<const K: usize>(
});
// Repaired shreds are not retransmitted.
stats.num_retransmit_shreds += shreds.len();
retransmit_sender.send(shreds.clone())?;
if let Err(send_err) = retransmit_sender.try_send(shreds.clone()) {
match send_err {
crossbeam_channel::TrySendError::Full(v) => {
stats.num_retransmit_stage_overflow += v.len();
}
crossbeam_channel::TrySendError::Disconnected(_) => {
return Err(Error::RecvDisconnected);
}
}
}
// Send all shreds to window service to be inserted into blockstore.
let shreds = shreds
.into_iter()
Expand Down Expand Up @@ -412,6 +421,7 @@ struct ShredSigVerifyStats {
num_discards_pre: usize,
num_duplicates: usize,
num_invalid_retransmitter: AtomicUsize,
num_retransmit_stage_overflow: usize,
num_retransmit_shreds: usize,
num_unknown_slot_leader: AtomicUsize,
num_unknown_turbine_parent: AtomicUsize,
Expand All @@ -433,6 +443,7 @@ impl ShredSigVerifyStats {
num_discards_post: 0usize,
num_duplicates: 0usize,
num_invalid_retransmitter: AtomicUsize::default(),
num_retransmit_stage_overflow: 0,
num_retransmit_shreds: 0usize,
num_unknown_slot_leader: AtomicUsize::default(),
num_unknown_turbine_parent: AtomicUsize::default(),
Expand All @@ -459,6 +470,11 @@ impl ShredSigVerifyStats {
self.num_invalid_retransmitter.load(Ordering::Relaxed),
i64
),
(
"num_retransmit_stage_overflow",
self.num_retransmit_stage_overflow,
i64
),
("num_retransmit_shreds", self.num_retransmit_shreds, i64),
(
"num_unknown_slot_leader",
Expand Down
Loading