Skip to content

Commit

Permalink
Revert "Add limit and shrink policy for recycler (solana-labs#15320)"
Browse files Browse the repository at this point in the history
This reverts commit c2e8814.
  • Loading branch information
behzadnouri committed Apr 7, 2021
1 parent c5c3ae0 commit a1f5055
Show file tree
Hide file tree
Showing 24 changed files with 129 additions and 513 deletions.
1 change: 0 additions & 1 deletion Cargo.lock

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

2 changes: 1 addition & 1 deletion bench-streamer/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@ fn main() -> Result<()> {

let mut read_channels = Vec::new();
let mut read_threads = Vec::new();
let recycler = PacketsRecycler::new_without_limit("bench-streamer-recycler-shrink-stats");
let recycler = PacketsRecycler::default();
for _ in 0..num_sockets {
let read = solana_net_utils::bind_to(ip_addr, port, false).unwrap();
read.set_read_timeout(Some(Duration::new(1, 0))).unwrap();
Expand Down
12 changes: 6 additions & 6 deletions core/src/cluster_info.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1926,7 +1926,7 @@ impl ClusterInfo {
let mut last_contact_info_trace = timestamp();
let mut last_contact_info_save = timestamp();
let mut entrypoints_processed = false;
let recycler = PacketsRecycler::new_without_limit("gossip-recycler-shrink-stats");
let recycler = PacketsRecycler::default();
let crds_data = vec![
CrdsData::Version(Version::new(self.id())),
CrdsData::NodeInstance(self.instance.with_wallclock(timestamp())),
Expand Down Expand Up @@ -2191,7 +2191,7 @@ impl ClusterInfo {
.process_pull_requests(callers.cloned(), timestamp());
let output_size_limit =
self.update_data_budget(stakes.len()) / PULL_RESPONSE_MIN_SERIALIZED_SIZE;
let mut packets = Packets::new_with_recycler(recycler.clone(), 64).unwrap();
let mut packets = Packets::new_with_recycler(recycler.clone(), 64, "handle_pull_requests");
let (caller_and_filters, addrs): (Vec<_>, Vec<_>) = {
let mut rng = rand::thread_rng();
let check_pull_request =
Expand Down Expand Up @@ -2480,7 +2480,8 @@ impl ClusterInfo {
if packets.is_empty() {
None
} else {
let packets = Packets::new_with_recycler_data(recycler, packets).unwrap();
let packets =
Packets::new_with_recycler_data(recycler, "handle_ping_messages", packets);
Some(packets)
}
}
Expand Down Expand Up @@ -3172,8 +3173,7 @@ impl ClusterInfo {
exit: &Arc<AtomicBool>,
) -> JoinHandle<()> {
let exit = exit.clone();
let recycler =
PacketsRecycler::new_without_limit("cluster-info-listen-recycler-shrink-stats");
let recycler = PacketsRecycler::default();
Builder::new()
.name("solana-listen".to_string())
.spawn(move || {
Expand Down Expand Up @@ -3619,7 +3619,7 @@ mod tests {
.iter()
.map(|ping| Pong::new(ping, &this_node).unwrap())
.collect();
let recycler = PacketsRecycler::new_without_limit("");
let recycler = PacketsRecycler::default();
let packets = cluster_info
.handle_ping_messages(
remote_nodes
Expand Down
7 changes: 1 addition & 6 deletions core/src/fetch_stage.rs
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,6 @@ impl FetchStage {
exit,
&sender,
&poh_recorder,
None,
coalesce_ms,
),
receiver,
Expand All @@ -48,7 +47,6 @@ impl FetchStage {
exit: &Arc<AtomicBool>,
sender: &PacketSender,
poh_recorder: &Arc<Mutex<PohRecorder>>,
allocated_packet_limit: Option<u32>,
coalesce_ms: u64,
) -> Self {
let tx_sockets = sockets.into_iter().map(Arc::new).collect();
Expand All @@ -59,7 +57,6 @@ impl FetchStage {
exit,
&sender,
&poh_recorder,
allocated_packet_limit,
coalesce_ms,
)
}
Expand Down Expand Up @@ -105,11 +102,9 @@ impl FetchStage {
exit: &Arc<AtomicBool>,
sender: &PacketSender,
poh_recorder: &Arc<Mutex<PohRecorder>>,
limit: Option<u32>,
coalesce_ms: u64,
) -> Self {
let recycler: PacketsRecycler =
Recycler::warmed(1000, 1024, limit, "fetch_stage_recycler_shrink");
let recycler: PacketsRecycler = Recycler::warmed(1000, 1024);

let tpu_threads = sockets.into_iter().map(|socket| {
streamer::receiver(
Expand Down
2 changes: 1 addition & 1 deletion core/src/gossip_service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,7 @@ impl GossipService {
gossip_socket.clone(),
&exit,
request_sender,
Recycler::new_without_limit("gossip-receiver-recycler-shrink-stats"),
Recycler::default(),
"gossip_receiver",
1,
);
Expand Down
24 changes: 16 additions & 8 deletions core/src/serve_repair.rs
Original file line number Diff line number Diff line change
Expand Up @@ -279,7 +279,7 @@ impl ServeRepair {
exit: &Arc<AtomicBool>,
) -> JoinHandle<()> {
let exit = exit.clone();
let recycler = PacketsRecycler::new_without_limit("serve-repair-recycler-shrink-stats");
let recycler = PacketsRecycler::default();
Builder::new()
.name("solana-repair-listen".to_string())
.spawn(move || {
Expand Down Expand Up @@ -498,7 +498,11 @@ impl ServeRepair {

if let Some(packet) = packet {
inc_new_counter_debug!("serve_repair-window-request-ledger", 1);
return Some(Packets::new_with_recycler_data(recycler, vec![packet])).unwrap();
return Some(Packets::new_with_recycler_data(
recycler,
"run_window_request",
vec![packet],
));
}
}

Expand Down Expand Up @@ -534,7 +538,11 @@ impl ServeRepair {
from_addr,
nonce,
)?;
return Packets::new_with_recycler_data(recycler, vec![packet]);
return Some(Packets::new_with_recycler_data(
recycler,
"run_highest_window_request",
vec![packet],
));
}
None
}
Expand All @@ -547,7 +555,7 @@ impl ServeRepair {
max_responses: usize,
nonce: Nonce,
) -> Option<Packets> {
let mut res = Packets::new_with_recycler(recycler.clone(), 64).unwrap();
let mut res = Packets::new_with_recycler(recycler.clone(), 64, "run_orphan");
if let Some(blockstore) = blockstore {
// Try to find the next "n" parent slots of the input slot
while let Ok(Some(meta)) = blockstore.meta(slot) {
Expand Down Expand Up @@ -601,7 +609,7 @@ mod tests {

/// test run_window_request responds with the right shred, and do not overrun
fn run_highest_window_request(slot: Slot, num_slots: u64, nonce: Nonce) {
let recycler = PacketsRecycler::new_without_limit("");
let recycler = PacketsRecycler::default();
solana_logger::setup();
let ledger_path = get_tmp_ledger_path!();
{
Expand Down Expand Up @@ -669,7 +677,7 @@ mod tests {

/// test window requests respond with the right shred, and do not overrun
fn run_window_request(slot: Slot, nonce: Nonce) {
let recycler = PacketsRecycler::new_without_limit("");
let recycler = PacketsRecycler::default();
solana_logger::setup();
let ledger_path = get_tmp_ledger_path!();
{
Expand Down Expand Up @@ -837,7 +845,7 @@ mod tests {

fn run_orphan(slot: Slot, num_slots: u64, nonce: Nonce) {
solana_logger::setup();
let recycler = PacketsRecycler::new_without_limit("");
let recycler = PacketsRecycler::default();
let ledger_path = get_tmp_ledger_path!();
{
let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
Expand Down Expand Up @@ -908,7 +916,7 @@ mod tests {
#[test]
fn run_orphan_corrupted_shred_size() {
solana_logger::setup();
let recycler = PacketsRecycler::new_without_limit("");
let recycler = PacketsRecycler::default();
let ledger_path = get_tmp_ledger_path!();
{
let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
Expand Down
2 changes: 1 addition & 1 deletion core/src/serve_repair_service.rs
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ impl ServeRepairService {
serve_repair_socket.clone(),
&exit,
request_sender,
Recycler::new_without_limit("serve-repair-receiver-recycler-shrink-stats"),
Recycler::default(),
"serve_repair_receiver",
1,
);
Expand Down
4 changes: 1 addition & 3 deletions core/src/shred_fetch_stage.rs
Original file line number Diff line number Diff line change
Expand Up @@ -168,10 +168,8 @@ impl ShredFetchStage {
sender: &PacketSender,
bank_forks: Option<Arc<RwLock<BankForks>>>,
exit: &Arc<AtomicBool>,
limit: Option<u32>,
) -> Self {
let recycler: PacketsRecycler =
Recycler::warmed(100, 1024, limit, "shred_fetch_stage_recycler_shrink");
let recycler: PacketsRecycler = Recycler::warmed(100, 1024);

let (mut tvu_threads, tvu_filter) = Self::packet_modifier(
sockets,
Expand Down
4 changes: 2 additions & 2 deletions core/src/sigverify.rs
Original file line number Diff line number Diff line change
Expand Up @@ -23,8 +23,8 @@ impl Default for TransactionSigVerifier {
fn default() -> Self {
init();
Self {
recycler: Recycler::warmed(50, 4096, None, ""),
recycler_out: Recycler::warmed(50, 4096, None, ""),
recycler: Recycler::warmed(50, 4096),
recycler_out: Recycler::warmed(50, 4096),
}
}
}
Expand Down
5 changes: 1 addition & 4 deletions core/src/sigverify_shreds.rs
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,7 @@ impl ShredSigVerifier {
Self {
bank_forks,
leader_schedule_cache,
recycler_cache: RecyclerCache::warmed(
"shred-sig-verifier-offsets-recycler-shrink-stats",
"shred-sig-verifier-buffer-recycler-shrink-stats",
),
recycler_cache: RecyclerCache::warmed(),
}
}
fn read_slots(batches: &[Packets]) -> HashSet<u64> {
Expand Down
3 changes: 0 additions & 3 deletions core/src/tpu.rs
Original file line number Diff line number Diff line change
Expand Up @@ -74,9 +74,6 @@ impl Tpu {
&exit,
&packet_sender,
&poh_recorder,
// At 1024 packets per `Packet`, each packet about MTU size ~1k, this is roughly
// 20GB
Some(20_000),
tpu_coalesce_ms,
);
let (verified_sender, verified_receiver) = unbounded();
Expand Down
1 change: 0 additions & 1 deletion core/src/tvu.rs
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,6 @@ impl Tvu {
&fetch_sender,
Some(bank_forks.clone()),
&exit,
None,
);

let (verified_sender, verified_receiver) = unbounded();
Expand Down
2 changes: 1 addition & 1 deletion ledger/benches/sigverify_shreds.rs
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ const NUM_PACKETS: usize = 256;
const NUM_BATCHES: usize = 1;
#[bench]
fn bench_sigverify_shreds_sign_gpu(bencher: &mut Bencher) {
let recycler_cache = RecyclerCache::new("", "");
let recycler_cache = RecyclerCache::default();

let mut packets = Packets::default();
packets.packets.set_pinnable();
Expand Down
17 changes: 5 additions & 12 deletions ledger/src/entry.rs
Original file line number Diff line number Diff line change
Expand Up @@ -225,21 +225,12 @@ pub struct EntryVerificationState {
device_verification_data: DeviceVerificationData,
}

#[derive(Clone)]
#[derive(Default, Clone)]
pub struct VerifyRecyclers {
hash_recycler: Recycler<PinnedVec<Hash>>,
tick_count_recycler: Recycler<PinnedVec<u64>>,
}

impl Default for VerifyRecyclers {
fn default() -> Self {
Self {
hash_recycler: Recycler::new_without_limit("hash_recycler_shrink_stats"),
tick_count_recycler: Recycler::new_without_limit("tick_count_recycler_shrink_stats"),
}
}
}

#[derive(PartialEq, Clone, Copy, Debug)]
pub enum EntryVerificationStatus {
Failure,
Expand Down Expand Up @@ -563,12 +554,14 @@ impl EntrySlice for [Entry] {
.take(self.len())
.collect();

let mut hashes_pinned = recyclers.hash_recycler.allocate().unwrap();
let mut hashes_pinned = recyclers.hash_recycler.allocate("poh_verify_hash");
hashes_pinned.set_pinnable();
hashes_pinned.resize(hashes.len(), Hash::default());
hashes_pinned.copy_from_slice(&hashes);

let mut num_hashes_vec = recyclers.tick_count_recycler.allocate().unwrap();
let mut num_hashes_vec = recyclers
.tick_count_recycler
.allocate("poh_verify_num_hashes");
num_hashes_vec.reserve_and_pin(cmp::max(1, self.len()));
for entry in self {
num_hashes_vec.push(entry.num_hashes.saturating_sub(1));
Expand Down
24 changes: 12 additions & 12 deletions ledger/src/sigverify_shreds.rs
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,7 @@ fn slot_key_data_for_gpu<
.push(*slot);
}
}
let mut keyvec = recycler_cache.buffer().allocate().unwrap();
let mut keyvec = recycler_cache.buffer().allocate("shred_gpu_pubkeys");
keyvec.set_pinnable();
let mut slot_to_key_ix = HashMap::new();

Expand All @@ -152,7 +152,7 @@ fn slot_key_data_for_gpu<
slot_to_key_ix.insert(s, i);
}
}
let mut offsets = recycler_cache.offsets().allocate().unwrap();
let mut offsets = recycler_cache.offsets().allocate("shred_offsets");
offsets.set_pinnable();
slots.iter().for_each(|packet_slots| {
packet_slots.iter().for_each(|slot| {
Expand Down Expand Up @@ -185,11 +185,11 @@ fn shred_gpu_offsets(
batches: &[Packets],
recycler_cache: &RecyclerCache,
) -> (TxOffset, TxOffset, TxOffset, Vec<Vec<u32>>) {
let mut signature_offsets = recycler_cache.offsets().allocate().unwrap();
let mut signature_offsets = recycler_cache.offsets().allocate("shred_signatures");
signature_offsets.set_pinnable();
let mut msg_start_offsets = recycler_cache.offsets().allocate().unwrap();
let mut msg_start_offsets = recycler_cache.offsets().allocate("shred_msg_starts");
msg_start_offsets.set_pinnable();
let mut msg_sizes = recycler_cache.offsets().allocate().unwrap();
let mut msg_sizes = recycler_cache.offsets().allocate("shred_msg_sizes");
msg_sizes.set_pinnable();
let mut v_sig_lens = vec![];
for batch in batches.iter() {
Expand Down Expand Up @@ -242,7 +242,7 @@ pub fn verify_shreds_gpu(
trace!("pubkeys_len: {}", pubkeys_len);
let (signature_offsets, msg_start_offsets, msg_sizes, v_sig_lens) =
shred_gpu_offsets(pubkeys_len, batches, recycler_cache);
let mut out = recycler_cache.buffer().allocate().unwrap();
let mut out = recycler_cache.buffer().allocate("out_buffer");
out.set_pinnable();
elems.push(
perf_libs::Elems {
Expand Down Expand Up @@ -332,7 +332,7 @@ pub fn sign_shreds_cpu(keypair: &Keypair, batches: &mut [Packets]) {
}

pub fn sign_shreds_gpu_pinned_keypair(keypair: &Keypair, cache: &RecyclerCache) -> PinnedVec<u8> {
let mut vec = cache.buffer().allocate().unwrap();
let mut vec = cache.buffer().allocate("pinned_keypair");
let pubkey = keypair.pubkey().to_bytes();
let secret = keypair.secret().to_bytes();
let mut hasher = Sha512::default();
Expand Down Expand Up @@ -370,17 +370,17 @@ pub fn sign_shreds_gpu(
let mut num_packets = num_keypair_packets;

//should be zero
let mut pubkey_offsets = recycler_cache.offsets().allocate().unwrap();
let mut pubkey_offsets = recycler_cache.offsets().allocate("pubkey offsets");
pubkey_offsets.resize(count, 0);

let mut secret_offsets = recycler_cache.offsets().allocate().unwrap();
let mut secret_offsets = recycler_cache.offsets().allocate("secret_offsets");
secret_offsets.resize(count, pubkey_size as u32);

trace!("offset: {}", offset);
let (signature_offsets, msg_start_offsets, msg_sizes, _v_sig_lens) =
shred_gpu_offsets(offset, batches, recycler_cache);
let total_sigs = signature_offsets.len();
let mut signatures_out = recycler_cache.buffer().allocate().unwrap();
let mut signatures_out = recycler_cache.buffer().allocate("ed25519 signatures");
signatures_out.set_pinnable();
signatures_out.resize(total_sigs * sig_size, 0);
elems.push(
Expand Down Expand Up @@ -560,7 +560,7 @@ pub mod tests {

fn run_test_sigverify_shreds_gpu(slot: Slot) {
solana_logger::setup();
let recycler_cache = RecyclerCache::new("", "");
let recycler_cache = RecyclerCache::default();

let mut batch = [Packets::default()];
let mut shred = Shred::new_from_data(
Expand Down Expand Up @@ -624,7 +624,7 @@ pub mod tests {

fn run_test_sigverify_shreds_sign_gpu(slot: Slot) {
solana_logger::setup();
let recycler_cache = RecyclerCache::new("", "");
let recycler_cache = RecyclerCache::default();

let mut packets = Packets::default();
let num_packets = 32;
Expand Down
1 change: 0 additions & 1 deletion perf/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@ solana-sdk = { path = "../sdk", version = "=1.7.0" }
solana-rayon-threadlimit = { path = "../rayon-threadlimit", version = "=1.7.0" }
solana-budget-program = { path = "../programs/budget", version = "=1.7.0" }
solana-logger = { path = "../logger", version = "=1.7.0" }
solana-measure = { path = "../measure", version = "=1.7.0" }
solana-metrics = { path = "../metrics", version = "=1.7.0" }
curve25519-dalek = { version = "2" }

Expand Down
Loading

0 comments on commit a1f5055

Please sign in to comment.