Revert "Add limit and shrink policy for recycler (solana-labs#15320)"

This reverts commit c2e8814.

bench-streamer/src/main.rs

@@ -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();

core/src/cluster_info.rs

@@ -1925,7 +1925,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())),
@@ -2189,7 +2189,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 =
@@ -2478,7 +2478,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)
         }
     }
@@ -3170,8 +3171,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 || {
@@ -3616,7 +3616,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

core/src/fetch_stage.rs

@@ -35,7 +35,6 @@ impl FetchStage {
                 exit,
                 &sender,
                 &poh_recorder,
-                None,
                 coalesce_ms,
             ),
             receiver,
@@ -47,7 +46,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();
@@ -58,7 +56,6 @@ impl FetchStage {
             exit,
             &sender,
             &poh_recorder,
-            allocated_packet_limit,
             coalesce_ms,
         )
     }
@@ -104,11 +101,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(

core/src/gossip_service.rs

@@ -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,
         );

core/src/serve_repair.rs

@@ -278,7 +278,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 || {
@@ -490,7 +490,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],
+                ));
             }
         }
 
@@ -526,7 +530,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
     }
@@ -539,7 +547,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) {
@@ -593,7 +601,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!();
         {
@@ -661,7 +669,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!();
         {
@@ -829,7 +837,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());
@@ -900,7 +908,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());

core/src/serve_repair_service.rs

@@ -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,
         );

core/src/shred_fetch_stage.rs

@@ -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,

core/src/sigverify.rs

@@ -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),
         }
     }
 }

core/src/sigverify_shreds.rs

@@ -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> {

core/src/tpu.rs

@@ -75,9 +75,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();

core/src/tvu.rs

@@ -151,7 +151,6 @@ impl Tvu {
             &fetch_sender,
             Some(bank_forks.clone()),
             &exit,
-            None,
         );
 
         let (verified_sender, verified_receiver) = unbounded();

ledger/benches/sigverify_shreds.rs

@@ -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();

ledger/src/entry.rs

@@ -258,21 +258,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,
@@ -590,12 +581,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));

ledger/src/sigverify_shreds.rs

@@ -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();
 
@@ -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| {
@@ -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() {
@@ -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 {
@@ -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();
@@ -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(
@@ -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(
@@ -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;

perf/Cargo.toml

@@ -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" }