allows inserting shreds into blockstore using a reference (anza-xyz#4993)

Shreds generated during leader slots are concurrently inserted into
blockstore while also broadcasted to turbine. Currently shreds are
shared between the two paths using an Arc<Vec<Shred>> and sent to both
BroadcastRun::record and BroadcastRun::transmit:
https://github.com/anza-xyz/agave/blob/ddec7bdbc/turbine/src/broadcast_stage/standard_broadcast_run.rs#L224-L226
https://github.com/anza-xyz/agave/blob/ddec7bdbc/turbine/src/broadcast_stage.rs#L321
https://github.com/anza-xyz/agave/blob/ddec7bdbc/turbine/src/broadcast_stage.rs#L344

However, blockstore API for inserting shreds currently require an owned
shred anyways:
https://github.com/anza-xyz/agave/blob/ddec7bdbc/ledger/src/blockstore.rs#L1231

and we cannot get an owned shred out of Arc<Vec<Shred>> if there are
other references to the Arc.

As a result StandardBroadcastRun::insert is forced to do
Arc::unwrap_or_clone anyways without any guarantee that the other
reference (turbine broadcast path) is already dropped:
https://github.com/anza-xyz/agave/blob/b59c1aee7/turbine/src/broadcast_stage/standard_broadcast_run.rs#L370

This defeats the point of using Arc<Vec<Shred>> to begin with.

The commit updates blockstore insert-shreds API to instead work with
Cow<'a, Shred>. This allows to insert shreds into the blockstore with
either a reference or an owned value, and so Arc::unwrap_or_clone in
StandardBroadcastRun::insert can be avoided.

Author: behzadnouri

core/src/validator.rs

@@ -2460,7 +2460,8 @@ fn cleanup_blockstore_incorrect_shred_versions(
             let slot_meta_iterator = blockstore.slot_meta_iterator(start_slot)?;
             for (slot, _meta) in slot_meta_iterator {
                 let shreds = blockstore.get_data_shreds_for_slot(slot, 0)?;
-                let _ = backup_blockstore.insert_shreds(shreds, None, true);
+                let shreds = shreds.into_iter().map(Cow::Owned);
+                let _ = backup_blockstore.insert_cow_shreds(shreds, None, true);
                 num_slots_copied += 1;
 
                 if print_timer.elapsed() > PRINT_INTERVAL {

core/src/window_service.rs

@@ -27,6 +27,7 @@ use {
     solana_sdk::clock::{Slot, DEFAULT_MS_PER_SLOT},
     solana_turbine::cluster_nodes,
     std::{
+        borrow::Cow,
         net::UdpSocket,
         sync::{
             atomic::{AtomicBool, AtomicUsize, Ordering},
@@ -211,7 +212,7 @@ where
             debug_assert_matches!(shred, shred::Payload::Shared(_));
         }
         let shred = Shred::new_from_serialized_shred(shred).ok()?;
-        Some((shred, repair))
+        Some((Cow::Owned(shred), repair))
     };
     let now = Instant::now();
     let shreds: Vec<_> = thread_pool.install(|| {
@@ -608,9 +609,9 @@ mod test {
             };
             assert_eq!(duplicate_shred.slot(), slot);
             // Simulate storing both duplicate shreds in the same batch
-            let shreds = [original_shred.clone(), duplicate_shred.clone()]
+            let shreds = [&original_shred, &duplicate_shred]
                 .into_iter()
-                .map(|shred| (shred, /*is_repaired:*/ false));
+                .map(|shred| (Cow::Borrowed(shred), /*is_repaired:*/ false));
             blockstore
                 .insert_shreds_handle_duplicate(
                     shreds,

ledger-tool/src/blockstore.rs

@@ -31,6 +31,7 @@ use {
         hash::Hash,
     },
     std::{
+        borrow::Cow,
         collections::{BTreeMap, BTreeSet, HashMap},
         fs::File,
         io::{stdout, BufRead, BufReader, Write},
@@ -676,7 +677,8 @@ fn do_blockstore_process_command(ledger_path: &Path, matches: &ArgMatches<'_>) -
                     break;
                 }
                 let shreds = source.get_data_shreds_for_slot(slot, 0)?;
-                if target.insert_shreds(shreds, None, true).is_err() {
+                let shreds = shreds.into_iter().map(Cow::Owned);
+                if target.insert_cow_shreds(shreds, None, true).is_err() {
                     warn!("error inserting shreds for slot {}", slot);
                 }
             }