Merge pull request #444 from filecoin-project/feat/439-generate-post-to-sector-builder

feat/439 - move generate_post to SectorBuilder

Author: porcuquine

filecoin-proofs/examples/ffi/main.rs

@@ -281,7 +281,6 @@ unsafe fn sector_builder_lifecycle() -> Result<(), Box<Error>> {
             panic!("{}", c_str_to_rust_str((*resp).error_msg))
         }
 
-        // copy the bytes from C to Rust
         let data_ptr = (*resp).data_ptr as *mut u8;
         let data_len = (*resp).data_len;
         let mut bytes_out = Vec::with_capacity(data_len);

filecoin-proofs/src/api/internal.rs

@@ -17,7 +17,6 @@ use storage_proofs::circuit::zigzag::ZigZagCompound;
 use storage_proofs::compound_proof::{self, CompoundProof};
 use storage_proofs::drgporep::{self, DrgParams};
 use storage_proofs::drgraph::{new_seed, DefaultTreeHasher};
-use storage_proofs::error::Result;
 use storage_proofs::fr32::{bytes_into_fr, fr_into_bytes, Fr32Ary};
 use storage_proofs::hasher::Hasher;
 use storage_proofs::layered_drgporep;
@@ -30,6 +29,8 @@ use storage_proofs::proof::ProofScheme;
 use storage_proofs::zigzag_drgporep::ZigZagDrgPoRep;
 use storage_proofs::zigzag_graph::ZigZagBucketGraph;
 
+use crate::error;
+
 type Commitment = [u8; 32];
 
 /// FrSafe is an array of the largest whole number of bytes guaranteed not to overflow the field.
@@ -160,6 +161,34 @@ pub fn get_config(sector_config: &SectorConfig) -> (bool, Option<u32>, usize, us
     )
 }
 
+pub struct PoStOutput {
+    pub snark_proof: [u8; 192],
+    pub faults: Vec<u64>,
+}
+
+pub struct PoStInputPart {
+    pub sealed_sector_access: Option<String>,
+    pub comm_r: [u8; 32],
+}
+
+pub struct PoStInput {
+    pub challenge_seed: [u8; 32],
+    pub input_parts: Vec<PoStInputPart>,
+}
+
+pub fn generate_post(input: PoStInput) -> error::Result<PoStOutput> {
+    let faults: Vec<u64> = if !input.input_parts.is_empty() {
+        vec![0]
+    } else {
+        Default::default()
+    };
+
+    Ok(PoStOutput {
+        snark_proof: [42; 192],
+        faults,
+    })
+}
+
 pub struct SealOutput {
     pub comm_r: Commitment,
     pub comm_r_star: Commitment,
@@ -173,7 +202,7 @@ pub fn seal<T: Into<PathBuf> + AsRef<Path>>(
     out_path: T,
     prover_id_in: &FrSafe,
     sector_id_in: &FrSafe,
-) -> Result<SealOutput> {
+) -> error::Result<SealOutput> {
     let (fake, delay_seconds, sector_bytes, proof_sector_bytes, uses_official_circuit) =
         get_config(sector_config);
 
@@ -313,7 +342,7 @@ fn perform_replication<T: AsRef<Path>>(
     data: &mut [u8],
     fake: bool,
     proof_sector_bytes: usize,
-) -> Result<(
+) -> error::Result<(
     layered_drgporep::Tau<<DefaultTreeHasher as Hasher>::Domain>,
     Vec<MerkleTree<<DefaultTreeHasher as Hasher>::Domain, <DefaultTreeHasher as Hasher>::Function>>,
 )> {
@@ -343,7 +372,7 @@ fn perform_replication<T: AsRef<Path>>(
     }
 }
 
-fn write_data<T: AsRef<Path>>(out_path: T, data: &[u8]) -> Result<()> {
+fn write_data<T: AsRef<Path>>(out_path: T, data: &[u8]) -> error::Result<()> {
     // Write replicated data to out_path.
     let f_out = File::create(out_path)?;
     let mut buf_writer = BufWriter::new(f_out);
@@ -359,7 +388,7 @@ pub fn get_unsealed_range<T: Into<PathBuf> + AsRef<Path>>(
     sector_id_in: &FrSafe,
     offset: u64,
     num_bytes: u64,
-) -> Result<(u64)> {
+) -> error::Result<(u64)> {
     let (fake, delay_seconds, sector_bytes, proof_sector_bytes, _uses_official_circuit) =
         get_config(sector_config);
     if let Some(delay) = delay_seconds {
@@ -401,7 +430,7 @@ pub fn verify_seal(
     prover_id_in: &FrSafe,
     sector_id_in: &FrSafe,
     proof_vec: &[u8],
-) -> Result<bool> {
+) -> error::Result<bool> {
     let (_fake, _delay_seconds, _sector_bytes, proof_sector_bytes, uses_official_circuit) =
         get_config(sector_config);
 
@@ -455,7 +484,7 @@ pub fn verify_seal(
 
     let proof = MultiProof::new_from_reader(Some(POREP_PARTITIONS), proof_vec, groth_params)?;
 
-    ZigZagCompound::verify(&compound_public_params, &public_inputs, &proof)
+    ZigZagCompound::verify(&compound_public_params, &public_inputs, &proof).map_err(|e| e.into())
 }
 
 #[cfg(test)]

filecoin-proofs/src/api/mod.rs

@@ -1,3 +1,4 @@
+use crate::api::internal::PoStOutput;
 use crate::api::responses::err_code_and_msg;
 use crate::api::responses::FCPResponseStatus;
 use crate::api::responses::FFIPieceMetadata;
@@ -69,7 +70,7 @@ pub unsafe extern "C" fn verify_seal(
                 response.is_valid = false;
             }
             Err(err) => {
-                let (code, ptr) = err_code_and_msg(&err.into());
+                let (code, ptr) = err_code_and_msg(&err);
                 response.status_code = code;
                 response.error_msg = ptr;
             }
@@ -87,18 +88,12 @@ pub unsafe extern "C" fn verify_seal(
 
 /// Generates a proof-of-spacetime for the given replica commitments.
 ///
-/// # Arguments
-///
-/// * `flattened_comm_rs_ptr` - pointer to the first cell in an array containing replica commitment
-///                             bytes
-/// * `flattened_comm_rs_len` - number of bytes in the flattened_comm_rs_ptr array (must be a
-///                             multiple of 32)
-/// * `_challenge_seed`       - currently unused
 #[no_mangle]
 pub unsafe extern "C" fn generate_post(
+    ptr: *mut SectorBuilder,
     flattened_comm_rs_ptr: *const u8,
     flattened_comm_rs_len: libc::size_t,
-    _challenge_seed: &[u8; 32],
+    challenge_seed: &[u8; 32],
 ) -> *mut responses::GeneratePoSTResponse {
     let comm_rs = from_raw_parts(flattened_comm_rs_ptr, flattened_comm_rs_len)
         .iter()
@@ -111,24 +106,30 @@ pub unsafe extern "C" fn generate_post(
             acc
         });
 
-    // if more than one comm_r was provided, pretend like the first was faulty
-    let fault_idxs: Vec<u64> = vec![0]
-        .into_iter()
-        .take(if comm_rs.len() > 1 { 1 } else { 0 })
-        .collect();
+    let mut response: responses::GeneratePoSTResponse = Default::default();
 
-    let mut result: responses::GeneratePoSTResponse = Default::default();
-
-    result.faults_len = fault_idxs.len();
-    result.faults_ptr = fault_idxs.as_ptr();
+    match (*ptr).generate_post(&comm_rs, challenge_seed) {
+        Ok(PoStOutput {
+            snark_proof,
+            faults,
+        }) => {
+            response.status_code = FCPResponseStatus::FCPNoError;
+            response.proof = snark_proof;
 
-    // tell Rust to forget about the Vec; we'll free it when we free the GeneratePoSTResult
-    mem::forget(fault_idxs);
+            response.faults_len = faults.len();
+            response.faults_ptr = faults.as_ptr();
 
-    // write some fake proof
-    result.proof = [42; API_POST_PROOF_BYTES];
+            // we'll free this stuff when we free the GeneratePoSTResponse
+            mem::forget(faults);
+        }
+        Err(err) => {
+            let (code, ptr) = err_code_and_msg(&err);
+            response.status_code = code;
+            response.error_msg = ptr;
+        }
+    }
 
-    Box::into_raw(Box::new(result))
+    raw_ptr(response)
 }
 
 /// Verifies that a proof-of-spacetime is valid.
@@ -482,43 +483,3 @@ pub unsafe extern "C" fn get_staged_sectors(
 
     raw_ptr(response)
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use rand::{thread_rng, Rng};
-
-    #[test]
-    fn generate_verify_post_roundtrip_test() {
-        unsafe {
-            let comm_rs: [u8; 32] = thread_rng().gen();
-            let challenge_seed: [u8; 32] = thread_rng().gen();
-            let generate_post_res = generate_post(&comm_rs[0], 32, &challenge_seed);
-
-            assert_eq!(
-                FCPResponseStatus::FCPNoError,
-                (*generate_post_res).status_code,
-                "generate_post failed"
-            );
-
-            let verify_post_res = verify_post(
-                &comm_rs[0],
-                32,
-                &challenge_seed,
-                &(*generate_post_res).proof,
-                (*generate_post_res).faults_ptr,
-                (*generate_post_res).faults_len,
-            );
-
-            assert_eq!(
-                FCPResponseStatus::FCPNoError,
-                (*verify_post_res).status_code,
-                "error verifying PoSt"
-            );
-            assert_eq!(true, (*verify_post_res).is_valid, "invalid PoSt");
-
-            responses::destroy_generate_post_response(generate_post_res);
-            responses::destroy_verify_post_response(verify_post_res);
-        }
-    }
-}

filecoin-proofs/src/api/sector_builder/mod.rs

@@ -1,3 +1,4 @@
+use crate::api::internal::PoStOutput;
 use crate::api::sector_builder::errors::SectorBuilderErr;
 use crate::api::sector_builder::kv_store::fs::FileSystemKvs;
 use crate::api::sector_builder::kv_store::KeyValueStore;
@@ -143,7 +144,18 @@ impl SectorBuilder {
 
     // Returns all staged sector metadata.
     pub fn get_staged_sectors(&self) -> Result<Vec<StagedSectorMetadata>> {
-        self.run_blocking(Request::GetStagedSectors)
+        log_unrecov(self.run_blocking(Request::GetStagedSectors))
+    }
+
+    // Generates a proof-of-spacetime. Blocks the calling thread.
+    pub fn generate_post(
+        &self,
+        comm_rs: &[[u8; 32]],
+        challenge_seed: &[u8; 32],
+    ) -> Result<PoStOutput> {
+        log_unrecov(
+            self.run_blocking(|tx| Request::GeneratePoSt(Vec::from(comm_rs), *challenge_seed, tx)),
+        )
     }
 
     // Run a task, blocking on the return channel.

filecoin-proofs/src/api/sector_builder/scheduler.rs

@@ -1,3 +1,7 @@
+use crate::api::internal;
+use crate::api::internal::PoStInput;
+use crate::api::internal::PoStInputPart;
+use crate::api::internal::PoStOutput;
 use crate::api::sector_builder::errors::err_piecenotfound;
 use crate::api::sector_builder::errors::err_unrecov;
 use crate::api::sector_builder::helpers::add_piece::add_piece;
@@ -17,6 +21,7 @@ use crate::api::sector_builder::WrappedKeyValueStore;
 use crate::api::sector_builder::WrappedSectorStore;
 use crate::error::ExpectWithBacktrace;
 use crate::error::Result;
+use std::collections::HashMap;
 use std::sync::mpsc;
 use std::sync::Arc;
 use std::thread;
@@ -40,6 +45,11 @@ pub enum Request {
     GetSealedSectors(mpsc::SyncSender<Result<Vec<SealedSectorMetadata>>>),
     GetStagedSectors(mpsc::SyncSender<Result<Vec<StagedSectorMetadata>>>),
     GetSealStatus(SectorId, mpsc::SyncSender<Result<SealStatus>>),
+    GeneratePoSt(
+        Vec<[u8; 32]>,
+        [u8; 32],
+        mpsc::SyncSender<Result<PoStOutput>>,
+    ),
     RetrievePiece(String, mpsc::SyncSender<Result<Vec<u8>>>),
     SealAllStagedSectors(mpsc::SyncSender<Result<()>>),
     GetMaxUserBytesPerStagedSector(mpsc::SyncSender<u64>),
@@ -117,6 +127,9 @@ impl Scheduler {
                     Request::HandleSealResult(sector_id, result) => {
                         m.handle_seal_result(sector_id, *result);
                     }
+                    Request::GeneratePoSt(comm_rs, chg_seed, tx) => {
+                        m.generate_post(&comm_rs, &chg_seed, tx)
+                    }
                     Request::Shutdown => break,
                 }
             }
@@ -143,6 +156,46 @@ pub struct SectorMetadataManager {
 }
 
 impl SectorMetadataManager {
+    pub fn generate_post(
+        &self,
+        comm_rs: &[[u8; 32]],
+        challenge_seed: &[u8; 32],
+        return_channel: mpsc::SyncSender<Result<PoStOutput>>,
+    ) {
+        // reduce our sealed sector state-map to a mapping of comm_r to sealed
+        // sector access (AKA path to sealed sector file)
+        let comm_r_to_sector_access: HashMap<[u8; 32], String> = self
+            .state
+            .sealed
+            .sectors
+            .values()
+            .fold(HashMap::new(), |mut acc, item| {
+                let v = item.sector_access.clone();
+                let k = item.comm_r;
+                acc.entry(k).or_insert(v);
+                acc
+            });
+
+        let mut input_parts: Vec<PoStInputPart> = Default::default();
+
+        // eject from this loop with an error if we've been provided a comm_r
+        // which does not correspond to any sealed sector metadata
+        for comm_r in comm_rs {
+            input_parts.push(PoStInputPart {
+                sealed_sector_access: comm_r_to_sector_access.get(comm_r).cloned(),
+                comm_r: *comm_r,
+            });
+        }
+
+        let output = internal::generate_post(PoStInput {
+            challenge_seed: *challenge_seed,
+            input_parts,
+        });
+
+        // TODO: Where should this work be scheduled? New worker type?
+        return_channel.send(output).expects(FATAL_HUNGUP);
+    }
+
     // Unseals the sector containing the referenced piece and returns its
     // bytes. Produces an error if this sector builder does not have a sealed
     // sector containing the referenced piece.