Implement PieceCache::read_pieces() for ClusterPieceGetter

crates/subspace-farmer-components/src/lib.rs

@@ -54,19 +54,7 @@ pub trait PieceGetter {
         Box<dyn Stream<Item = (PieceIndex, anyhow::Result<Option<Piece>>)> + Send + Unpin + 'a>,
     >
     where
-        PieceIndices: IntoIterator<Item = PieceIndex, IntoIter: Send> + Send + 'a,
-    {
-        // TODO: Remove default impl here
-        Ok(Box::new(
-            piece_indices
-                .into_iter()
-                .map(|piece_index| async move {
-                    let result = self.get_piece(piece_index).await;
-                    (piece_index, result)
-                })
-                .collect::<FuturesUnordered<_>>(),
-        ) as Box<_>)
-    }
+        PieceIndices: IntoIterator<Item = PieceIndex, IntoIter: Send> + Send + 'a;
 }
 
 #[async_trait]

crates/subspace-farmer/src/bin/subspace-farmer/commands/cluster/plotter.rs

@@ -125,8 +125,8 @@ struct RocmPlottingOptions {
 pub(super) struct PlotterArgs {
     /// Piece getter concurrency.
     ///
-    /// Increasing this value can cause NATS communication issues if too many messages arrive via NATS, but
-    /// are not processed quickly enough.
+    /// Increasing this value can cause NATS communication issues if too many messages arrive via
+    /// NATS, but are not processed quickly enough.
     #[arg(long, default_value = "32")]
     piece_getter_concurrency: NonZeroUsize,
     /// Plotting options only used by CPU plotter

crates/subspace-farmer/src/cluster/cache.rs

@@ -90,7 +90,7 @@ impl GenericStreamRequest for ClusterCacheReadPiecesRequest {
     type Response = Result<(PieceCacheOffset, Option<(PieceIndex, Piece)>), String>;
 }
 
-/// Request plotted from farmer, request
+/// Collect plotted pieces from farmer
 #[derive(Debug, Clone, Encode, Decode)]
 struct ClusterCacheContentsRequest;