Merge pull request #977 from subspace/fix-block-production

Fix block production

Author: nazar-pc

Cargo.lock

@@ -31,6 +31,7 @@ parity-scale-codec = { version = "3.2.1", default-features = false, features = [
 rand = { version = "0.8.5", features = ["min_const_gen"], optional = true }
 rand_chacha = { version = "0.3.1", default-features = false }
 rand_core = { version = "0.6.4", default-features = false, features = ["alloc"] }
+rayon = { version = "1.6.0", optional = true }
 scale-info = { version = "2.3.0", default-features = false, features = ["derive"] }
 serde = { version = "1.0.147", optional = true, features = ["derive"] }
 serde_arrays = "0.1.0"
@@ -41,7 +42,13 @@ uint = { version = "0.9.4", default-features = false }
 criterion = "0.4.0"
 
 [features]
-default = ["std"]
+default = [
+    "parallel-decoding",
+    "std",
+]
+# Parallel decoding will use all CPUs available, but will allocate a memory of a size of a sector instead of square root
+# of that
+parallel-decoding = ["rayon"]
 std = [
     "ark-bls12-381/std",
     "ark-ff/std",

crates/subspace-core-primitives/Cargo.toml

@@ -6,10 +6,10 @@ mod tests;
 use crate::Scalar;
 use alloc::vec::Vec;
 use ark_bls12_381::Fr;
-use ark_poly::univariate::DensePolynomial;
-use ark_poly::{EvaluationDomain, GeneralEvaluationDomain, Polynomial, UVPolynomial};
-use core::mem;
+use ark_poly::{EvaluationDomain, GeneralEvaluationDomain};
 use num_integer::Roots;
+#[cfg(feature = "parallel-decoding")]
+use rayon::prelude::*;
 
 #[derive(Debug, Copy, Clone)]
 #[cfg_attr(feature = "thiserror", derive(thiserror::Error))]
@@ -127,34 +127,66 @@ impl SectorCodec {
             return Err(SectorCodecError::FailedToInstantiateDomain);
         };
 
-        let mut row = Vec::with_capacity(self.sector_side_size_in_scalars);
+        #[cfg(not(feature = "parallel-decoding"))]
+        {
+            let mut row = Vec::with_capacity(self.sector_side_size_in_scalars);
+
+            for row_index in 0..self.sector_side_size_in_scalars {
+                row.extend(
+                    sector
+                        .iter()
+                        .skip(row_index)
+                        .step_by(self.sector_side_size_in_scalars)
+                        .map(|scalar| scalar.0),
+                );
+
+                domain.coset_ifft_in_place(&mut row);
+                domain.fft_in_place(&mut row);
 
-        for row_index in 0..self.sector_side_size_in_scalars {
-            row.extend(
                 sector
-                    .iter()
+                    .iter_mut()
                     .skip(row_index)
                     .step_by(self.sector_side_size_in_scalars)
-                    .map(|scalar| scalar.0),
-            );
-
-            domain.coset_ifft_in_place(&mut row);
-
-            let polynomial = DensePolynomial::from_coefficients_vec(mem::take(&mut row));
+                    .zip(row.iter())
+                    .for_each(|(output, input)| *output = Scalar(*input));
 
-            sector
-                .iter_mut()
-                .skip(row_index)
-                .step_by(self.sector_side_size_in_scalars)
-                .zip(domain.elements())
-                .for_each(|(output, column)| {
-                    *output = Scalar(polynomial.evaluate(&column));
-                });
+                // Clear for next iteration of the loop
+                row.clear();
+            }
+        }
+        #[cfg(feature = "parallel-decoding")]
+        {
+            // Transform sector grid from columns to rows
+            let mut rows = vec![
+                vec![Fr::default(); self.sector_side_size_in_scalars];
+                self.sector_side_size_in_scalars
+            ];
+            for (row_index, row) in rows.iter_mut().enumerate() {
+                sector
+                    .iter()
+                    .skip(row_index)
+                    .step_by(self.sector_side_size_in_scalars)
+                    .zip(row)
+                    .for_each(|(input, output)| *output = input.0);
+            }
+
+            // Decode rows in parallel
+            rows.par_iter_mut().for_each(|row| {
+                domain.coset_ifft_in_place(row);
+                domain.fft_in_place(row);
+            });
 
-            // Reuse allocation and clear for next iteration of the loop
-            row = polynomial.coeffs;
-            row.clear();
+            // Store result back into inout sector
+            for (row_index, row) in rows.into_iter().enumerate() {
+                sector
+                    .iter_mut()
+                    .skip(row_index)
+                    .step_by(self.sector_side_size_in_scalars)
+                    .zip(row)
+                    .for_each(|(output, input)| *output = Scalar(input));
+            }
         }
+
         Ok(())
     }
 }

crates/subspace-core-primitives/src/sector_codec.rs

@@ -34,7 +34,7 @@ tracing = "0.1.37"
 criterion = "0.4.0"
 futures = "0.3.25"
 memmap2 = "0.5.8"
-rayon = "1.5.3"
+rayon = "1.6.0"
 subspace-archiving = { version = "0.1.0", path = "../subspace-archiving" }
 
 [[bench]]

crates/subspace-farmer-components/Cargo.toml

@@ -58,4 +58,4 @@ zeroize = "1.5.7"
 jemallocator = "0.5.0"
 
 [dev-dependencies]
-rayon = "1.5.3"
+rayon = "1.6.0"

crates/subspace-farmer/Cargo.toml

@@ -46,7 +46,7 @@ use subspace_rpc_primitives::{SlotInfo, SolutionResponse};
 use thiserror::Error;
 use tokio::runtime::Handle;
 use tokio::sync::broadcast;
-use tracing::{debug, error, info, info_span, trace, Instrument, Span};
+use tracing::{debug, error, info, info_span, trace, warn, Instrument, Span};
 use ulid::Ulid;
 
 // Refuse to compile on non-64-bit platforms, offsets may fail on those when converting from u64 to
@@ -715,23 +715,29 @@ impl SingleDiskPlot {
                                 Err(error) => Err(PlottingError::LowLevel(error))?,
                             };
 
-                            let mut metadata_header = metadata_header.lock();
-                            metadata_header.sector_count += 1;
-                            metadata_header_mmap
-                                .copy_from_slice(metadata_header.encode().as_slice());
+                            {
+                                let mut metadata_header = metadata_header.lock();
+                                metadata_header.sector_count += 1;
+                                metadata_header_mmap
+                                    .copy_from_slice(metadata_header.encode().as_slice());
+                            }
 
                             handlers.sector_plotted.call_simple(&plotted_sector);
 
                             // TODO: Migrate this over to using `on_sector_plotted` instead
                             // Publish pieces-by-sector if we use DSN
-                            let publishing_result = handle.block_on(
-                                piece_publisher.publish_pieces(plotted_sector.piece_indexes),
-                            );
-
-                            // cancelled
-                            if publishing_result.is_err() {
-                                return;
-                            }
+                            tokio::spawn({
+                                let piece_publisher = piece_publisher.clone();
+
+                                async move {
+                                    if let Err(error) = piece_publisher
+                                        .publish_pieces(plotted_sector.piece_indexes)
+                                        .await
+                                    {
+                                        warn!(%sector_index, %error, "Failed to publish pieces to DSN");
+                                    }
+                                }
+                            });
                         }
                     };
 
@@ -915,6 +921,13 @@ impl SingleDiskPlot {
                                 if solutions.len() >= SOLUTIONS_LIMIT {
                                     break;
                                 }
+                                // TODO: It is known that decoding is slow now and we'll only be
+                                //  able to decode a single sector within time slot reliably, in the
+                                //  future we may want allow more than one sector to be valid within
+                                //  the same disk plot.
+                                if !solutions.is_empty() {
+                                    break;
+                                }
                             }
 
                             let response = SolutionResponse {

crates/subspace-farmer/src/single_disk_plot.rs

@@ -14,6 +14,7 @@ use tracing::{debug, error, trace};
 const PUBLISH_PIECE_BY_SECTOR_WAITING_DURATION_IN_SECS: u64 = 1;
 
 // Piece-by-sector DSN publishing helper.
+#[derive(Clone)]
 pub(crate) struct PieceSectorPublisher {
     dsn_node: Node,
     cancelled: Arc<AtomicBool>,