Merge pull request #1026 from filecoin-project/feat/pad-reader

feat: improved approach to do bit padding/calculate piece commitments

Author: dignifiedquire

fil-proofs-tooling/src/bin/benchy/flarp.rs

@@ -89,6 +89,10 @@ pub struct FlarpOutputs {
     tree_r_last_wall_time_ms: u64,
     window_comm_leaves_time_cpu_time_ms: u64,
     window_comm_leaves_time_wall_time_ms: u64,
+    add_piece_cpu_time_ms: u64,
+    add_piece_wall_time_ms: u64,
+    generate_piece_commitment_cpu_time_ms: u64,
+    generate_piece_commitment_wall_time_ms: u64,
     #[serde(flatten)]
     circuits: CircuitOutputs,
 }
@@ -157,6 +161,14 @@ fn augment_with_op_measurements(mut output: &mut FlarpOutputs) {
                 output.post_partial_ticket_hash_cpu_time_ms = cpu_time;
                 output.post_partial_ticket_hash_time_ms = wall_time;
             }
+            AddPiece => {
+                output.add_piece_cpu_time_ms = cpu_time;
+                output.add_piece_wall_time_ms = wall_time;
+            }
+            GeneratePieceCommitment => {
+                output.generate_piece_commitment_cpu_time_ms = cpu_time;
+                output.generate_piece_commitment_wall_time_ms = wall_time;
+            }
         }
     }
 }
@@ -183,6 +195,7 @@ pub fn run(
     skip_seal_proof: bool,
     skip_post_proof: bool,
     only_replicate: bool,
+    only_add_piece: bool,
 ) -> Metadata<FlarpReport> {
     configure_global_config(&inputs);
 
@@ -192,15 +205,15 @@ pub fn run(
 
     assert!(inputs.num_sectors > 0, "Missing num_sectors");
 
-    let (cfg, created, replica_measurement) =
-        create_replicas(sector_size, inputs.num_sectors as usize);
+    let (cfg, repls) = create_replicas(sector_size, inputs.num_sectors as usize, only_add_piece);
 
-    if only_replicate {
+    if only_add_piece || only_replicate {
         augment_with_op_measurements(&mut outputs);
         return Metadata::wrap(FlarpReport { inputs, outputs })
             .expect("failed to retrieve metadata");
     }
 
+    let (created, replica_measurement) = repls.unwrap();
     generate_params(&inputs);
 
     if !skip_seal_proof {

fil-proofs-tooling/src/bin/benchy/main.rs

@@ -134,6 +134,12 @@ fn main() -> Result<()> {
                 .long("only-replicate")
                 .takes_value(false)
                 .help("only run replication"),
+        )
+        .arg(
+            Arg::with_name("only-add-piece")
+                .long("only-add-piece")
+                .takes_value(false)
+                .help("only run piece addition"),
         );
 
     let merkleproof_cmd = SubCommand::with_name("merkleproofs")
@@ -215,6 +221,7 @@ fn main() -> Result<()> {
                 m.is_present("skip-seal-proof"),
                 m.is_present("skip-post-proof"),
                 m.is_present("only-replicate"),
+                m.is_present("only-add-piece"),
             );
 
             serde_json::to_writer(stdout(), &outputs)

fil-proofs-tooling/src/bin/benchy/shared.rs

@@ -9,9 +9,8 @@ use fil_proofs_tooling::{measure, FuncMeasurement};
 use filecoin_proofs::constants::POREP_PARTITIONS;
 use filecoin_proofs::types::{PaddedBytesAmount, PoRepConfig, SectorSize, UnpaddedBytesAmount};
 use filecoin_proofs::{
-    add_piece, generate_piece_commitment, seal_pre_commit_phase1, seal_pre_commit_phase2,
-    validate_cache_for_precommit_phase2, PieceInfo, PoRepProofPartitions, PrivateReplicaInfo,
-    PublicReplicaInfo, SealPreCommitOutput,
+    add_piece, seal_pre_commit_phase1, seal_pre_commit_phase2, validate_cache_for_precommit_phase2,
+    PieceInfo, PoRepProofPartitions, PrivateReplicaInfo, PublicReplicaInfo, SealPreCommitOutput,
 };
 use storage_proofs::sector::SectorId;
 
@@ -26,7 +25,7 @@ pub struct PreCommitReplicaOutput {
     pub public_replica_info: PublicReplicaInfo,
 }
 
-pub fn create_piece(piece_bytes: UnpaddedBytesAmount) -> (NamedTempFile, PieceInfo) {
+pub fn create_piece(piece_bytes: UnpaddedBytesAmount) -> NamedTempFile {
     info!("create_piece");
     let mut file = NamedTempFile::new().expect("failed to create piece file");
     {
@@ -57,24 +56,19 @@ pub fn create_piece(piece_bytes: UnpaddedBytesAmount) -> (NamedTempFile, PieceIn
         .seek(SeekFrom::Start(0))
         .expect("failed to seek to beginning of piece file");
 
-    info!("generating piece commitment");
-    let info = generate_piece_commitment(file.as_file_mut(), piece_bytes)
-        .expect("failed to generate piece commitment");
-
-    file.as_file_mut()
-        .seek(SeekFrom::Start(0))
-        .expect("failed to seek to beginning of piece file");
-
-    (file, info)
+    file
 }
 
 pub fn create_replicas(
     sector_size: SectorSize,
     qty_sectors: usize,
+    only_add: bool,
 ) -> (
     PoRepConfig,
-    Vec<(SectorId, PreCommitReplicaOutput)>,
-    FuncMeasurement<Vec<SealPreCommitOutput>>,
+    Option<(
+        Vec<(SectorId, PreCommitReplicaOutput)>,
+        FuncMeasurement<Vec<SealPreCommitOutput>>,
+    )>,
 ) {
     info!("creating replicas: {:?} - {}", sector_size, qty_sectors);
     let sector_size_unpadded_bytes_ammount =
@@ -115,30 +109,35 @@ pub fn create_replicas(
         staged_files.push(staged_file);
     }
 
-    let (piece_files, piece_infos): (Vec<_>, Vec<_>) = (0..qty_sectors)
+    let piece_files: Vec<_> = (0..qty_sectors)
         .into_par_iter()
         .map(|_i| {
-            let (piece_file, piece_info) = create_piece(UnpaddedBytesAmount::from(
-                PaddedBytesAmount::from(sector_size),
-            ));
-            (piece_file, vec![piece_info])
+            create_piece(UnpaddedBytesAmount::from(PaddedBytesAmount::from(
+                sector_size,
+            )))
         })
-        .unzip();
+        .collect();
 
     info!("adding pieces");
+    let mut piece_infos = Vec::new();
     for (i, (mut piece_file, mut staged_file)) in piece_files
         .into_iter()
         .zip(staged_files.iter_mut())
         .enumerate()
     {
         info!("add piece {}", i);
-        add_piece(
+        let info = add_piece(
             &mut piece_file,
             &mut staged_file,
             sector_size_unpadded_bytes_ammount,
             &[],
         )
         .unwrap();
+        piece_infos.push(vec![info]);
+    }
+
+    if only_add {
+        return (porep_config, None);
     }
 
     let seal_pre_commit_outputs = measure(|| {
@@ -213,5 +212,5 @@ pub fn create_replicas(
         ));
     }
 
-    (porep_config, out, seal_pre_commit_outputs)
+    (porep_config, Some((out, seal_pre_commit_outputs)))
 }

filecoin-proofs/Cargo.toml

@@ -17,7 +17,6 @@ lazy_static = "1.2"
 memmap = "0.7"
 colored = "1.6"
 pbr = "1.0"
-tempfile = "3"
 byteorder = "1"
 itertools = "0.8"
 serde_cbor = "0.10.2"
@@ -46,22 +45,24 @@ bincode = "1.1.2"
 anyhow = "1.0.23"
 rand_xorshift = "0.2.0"
 typenum = "1.11.2"
+bitintr = "0.3.0"
+gperftools = { version = "0.2", optional = true }
 
 [dependencies.reqwest]
 version = "0.9"
 default-features = false
 features = ["default-tls-vendored"]
 
 [dev-dependencies]
-gperftools = "0.2"
 criterion = "0.3"
 rexpect = "0.3.0"
 pretty_assertions = "0.6.1"
 failure = "0.1"
+tempfile = "3"
 
 [features]
 default = ["gpu"]
-cpu-profile = []
+cpu-profile = ["gperftools"]
 heap-profile = ["gperftools/heap"]
 simd = ["storage-proofs/simd"]
 asm = ["storage-proofs/asm"]

filecoin-proofs/benches/preprocessing.rs

@@ -1,18 +1,40 @@
-use std::fs::File;
-use std::io::{Read, Seek, SeekFrom};
+use std::io::{self, Read};
 use std::time::Duration;
 
 use criterion::{criterion_group, criterion_main, Criterion, ParameterizedBenchmark, Throughput};
-use filecoin_proofs::fr32::{write_padded, write_unpadded};
 use rand::{thread_rng, Rng};
 
+#[cfg(feature = "cpu-profile")]
+#[inline(always)]
+fn start_profile(stage: &str) {
+    gperftools::profiler::PROFILER
+        .lock()
+        .unwrap()
+        .start(format!("./{}.profile", stage))
+        .unwrap();
+}
+
+#[cfg(not(feature = "cpu-profile"))]
+#[inline(always)]
+fn start_profile(_stage: &str) {}
+
+#[cfg(feature = "cpu-profile")]
+#[inline(always)]
+fn stop_profile() {
+    gperftools::profiler::PROFILER
+        .lock()
+        .unwrap()
+        .stop()
+        .unwrap();
+}
+
+#[cfg(not(feature = "cpu-profile"))]
+#[inline(always)]
+fn stop_profile() {}
+
 fn random_data(size: usize) -> Vec<u8> {
     let mut rng = thread_rng();
-    let mut data = vec![0u8; size as usize];
-    for i in 0..data.len() {
-        data[i] = rng.gen();
-    }
-    data
+    (0..size).map(|_| rng.gen()).collect()
 }
 
 fn preprocessing_benchmark(c: &mut Criterion) {
@@ -22,56 +44,23 @@ fn preprocessing_benchmark(c: &mut Criterion) {
             "write_padded",
             |b, size| {
                 let data = random_data(*size);
+                let mut buf = Vec::with_capacity(*size);
 
+                start_profile(&format!("write_padded_{}", *size));
                 b.iter(|| {
-                    let tmpfile: File = tempfile::tempfile().unwrap();
-
-                    write_padded_bench(tmpfile, data.clone());
-                })
+                    let mut reader = filecoin_proofs::PadReader::new(io::Cursor::new(&data));
+                    reader.read_to_end(&mut buf).unwrap();
+                    assert!(buf.len() >= data.len());
+                });
+                stop_profile();
             },
             vec![128, 256, 512, 256_000, 512_000, 1024_000, 2048_000],
         )
-        .with_function("write_padded + unpadded", |b, size| {
-            let data = random_data(*size);
-
-            b.iter(|| {
-                let tmpfile: File = tempfile::tempfile().unwrap();
-
-                write_padded_unpadded_bench(tmpfile, data.clone());
-            })
-        })
         .sample_size(10)
         .throughput(|s| Throughput::Bytes(*s as u64))
         .warm_up_time(Duration::from_secs(1)),
     );
 }
 
-fn write_padded_bench(mut file: File, data: Vec<u8>) {
-    let _ = write_padded(&mut &data[..], &mut file).unwrap();
-    let padded_written = file.seek(SeekFrom::End(0)).unwrap() as usize;
-
-    assert!(padded_written > data.len());
-}
-
-fn write_padded_unpadded_bench(mut file: File, data: Vec<u8>) {
-    write_padded(&mut &data[..], &mut file).unwrap();
-
-    let padded_written = file.seek(SeekFrom::End(0)).unwrap() as usize;
-
-    assert!(padded_written > data.len());
-
-    let mut buf = Vec::with_capacity(padded_written);
-    file.seek(SeekFrom::Start(0)).unwrap();
-    file.read_to_end(&mut buf).unwrap();
-
-    let mut unpadded_file: File = tempfile::tempfile().unwrap();
-
-    write_unpadded(&buf, &mut unpadded_file, 0, data.len()).unwrap();
-
-    let unpadded_written = unpadded_file.seek(SeekFrom::End(0)).unwrap() as usize;
-
-    assert!(unpadded_written == data.len());
-}
-
 criterion_group!(benches, preprocessing_benchmark);
 criterion_main!(benches);