runner & prover

.cargo/config.toml

@@ -1,2 +1,5 @@
 [env]
-RUST_LOG = "info"
+RUST_LOG = "info"
+CAIRO_PATH = "cairo"
+BOOTLOADER_PATH = "bootloader/recursive_with_poseidon/simple_bootloader.cairo"
+BOOTLOADER_OUT_NAME = "bootloader.json"

Cargo.toml

@@ -9,7 +9,7 @@ members = [
     "crates/executor",
     "crates/peer",
     "crates/prover",
-    "crates/runner",
+    "crates/runner", "crates/tests",
 ]
 exclude = []
 
@@ -44,6 +44,7 @@ libp2p = { version = "0.53.2", features = [
 ] }
 libsecp256k1 = "0.7.1"
 num-bigint = "0.4.4"
+rand = "0.8.5"
 serde = "1.0.197"
 serde_json = "1.0.115"
 starknet = "0.9.0"
@@ -56,9 +57,10 @@ tracing = "0.1.37"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 zip-extensions = "0.6.2"
 
-
 sharp-p2p-common = { path = "crates/common" }
 sharp-p2p-delegator = { path = "crates/delegator" }
 sharp-p2p-executor = { path = "crates/executor" }
 sharp-p2p-peer = { path = "crates/peer" }
 sharp-p2p-prover = { path = "crates/prover" }
+sharp-p2p-runner = { path = "crates/runner" }
+sharp-p2p-tests = { path = "crates/tests" }

crates/common/Cargo.toml

@@ -10,6 +10,7 @@ license-file.workspace = true
 [dependencies]
 bincode.workspace = true
 cairo-felt.workspace = true
+futures.workspace= true
 hex.workspace = true
 libp2p.workspace = true
 libsecp256k1.workspace = true
@@ -18,3 +19,4 @@ serde_json.workspace = true
 serde.workspace = true
 tempfile.workspace = true
 thiserror.workspace = true
+tokio.workspace = true

crates/common/src/job.rs

@@ -1,8 +1,10 @@
 use crate::hash;
-use libsecp256k1::{curve::Scalar, sign, PublicKey, SecretKey, Signature};
+use libsecp256k1::{curve::Scalar, sign, Message, PublicKey, SecretKey, Signature};
 use std::{
     fmt::Display,
+    fs,
     hash::{DefaultHasher, Hash, Hasher},
+    path::PathBuf,
 };
 
 /*
@@ -24,6 +26,33 @@ pub struct Job {
     pub signature: Signature, // The signature of the delegator, used in the bootloader stage to confirm authenticity of the Job<->Delegator relationship
 }
 
+impl Job {
+    pub fn new(
+        reward: u32,
+        num_of_steps: u32,
+        cairo_pie_file: PathBuf,
+        registry_address: &str,
+        secret_key: SecretKey,
+    ) -> Self {
+        Self {
+            reward,
+            num_of_steps,
+            cairo_pie: fs::read(cairo_pie_file).unwrap(),
+            registry_address: registry_address.to_string(),
+            public_key: PublicKey::from_secret_key(&secret_key),
+            signature: libsecp256k1::sign(
+                // TODO proper impl just mocked rn for tests
+                &Message::parse(&[
+                    0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0,
+                    0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0, 0x0,
+                ]),
+                &secret_key,
+            )
+            .0,
+        }
+    }
+}
+
 impl Default for Job {
     fn default() -> Self {
         let secret_key = &SecretKey::default();

crates/common/src/lib.rs

@@ -3,5 +3,6 @@ pub mod job;
 pub mod job_trace;
 pub mod job_witness;
 pub mod network;
+pub mod process;
 pub mod topic;
 pub mod vec252;

crates/common/src/process.rs

@@ -0,0 +1,35 @@
+use futures::{Future, FutureExt};
+use std::pin::Pin;
+use tokio::sync::mpsc;
+
+pub struct Process<'future, PR> {
+    future: Pin<Box<dyn Future<Output = PR> + 'future>>,
+    abort: mpsc::Sender<()>,
+}
+
+impl<'future, PR> Process<'future, PR> {
+    pub fn new(
+        future: Pin<Box<dyn Future<Output = PR> + 'future>>,
+        abort: mpsc::Sender<()>,
+    ) -> Self {
+        Self { future, abort }
+    }
+
+    pub async fn abort(&self) -> Result<(), mpsc::error::SendError<()>> {
+        self.abort.send(()).await
+    }
+
+    pub fn into_parts(self) -> (Pin<Box<dyn Future<Output = PR> + 'future>>, mpsc::Sender<()>) {
+        (self.future, self.abort)
+    }
+}
+
+impl<'future, PR> Future for Process<'future, PR> {
+    type Output = PR;
+    fn poll(
+        mut self: Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Self::Output> {
+        self.future.poll_unpin(cx)
+    }
+}

crates/prover/Cargo.toml

@@ -13,7 +13,9 @@ cairo-proof-parser.workspace = true
 futures.workspace= true
 itertools.workspace = true
 serde_json.workspace = true
+serde.workspace = true
 sharp-p2p-common.workspace = true
+strum.workspace = true
 tempfile.workspace = true
 thiserror.workspace = true
 tokio.workspace = true

crates/prover/build.rs

@@ -1,9 +1,9 @@
-// use std::process::Command;
+use std::process::Command;
 
 fn main() {
     // Check if stone-prover command is present
-    // Command::new("cpu_air_prover")
-    //     .arg("--help")
-    //     .output()
-    //     .expect("Failed to execute cpu_air_prover command");
+    Command::new("cpu_air_prover")
+        .arg("--help")
+        .output()
+        .expect("Failed to execute cpu_air_prover command");
 }

crates/prover/src/stone_prover/mod.rs

@@ -1,121 +1,113 @@
-use crate::{
-    errors::ProverControllerError,
-    traits::{Prover, ProverController},
-};
+use self::types::{config::Config, params::Params};
+use crate::{errors::ProverControllerError, traits::ProverController};
+use async_process::Stdio;
+use futures::Future;
 use itertools::{chain, Itertools};
-use sharp_p2p_common::{hash, job_trace::JobTrace, job_witness::JobWitness, vec252::VecFelt252};
+use sharp_p2p_common::{
+    hash, job_trace::JobTrace, job_witness::JobWitness, process::Process, vec252::VecFelt252,
+};
 use std::{
-    collections::HashMap,
     hash::{DefaultHasher, Hash, Hasher},
-    io::Read,
+    io::{Read, Write},
+    pin::Pin,
 };
 use tempfile::NamedTempFile;
-use tokio::process::{Child, Command};
-use tracing::{debug, trace};
+use tokio::{process::Command, select, sync::mpsc};
+use tracing::debug;
+
+pub mod tests;
+pub mod types;
 
 pub struct StoneProver {
-    tasks: HashMap<u64, Child>,
+    cpu_air_prover_config: Config,
+    cpu_air_params: Params,
 }
 
-impl Prover for StoneProver {
-    fn init() -> impl ProverController {
-        Self { tasks: HashMap::new() }
+impl StoneProver {
+    pub fn new(cpu_air_prover_config: Config, cpu_air_params: Params) -> Self {
+        Self { cpu_air_prover_config, cpu_air_params }
     }
 }
 
 impl ProverController for StoneProver {
-    async fn prove(&mut self, job_trace: JobTrace) -> Result<JobWitness, ProverControllerError> {
-        let mut out_file = NamedTempFile::new()?;
-
-        let cpu_air_prover_config = NamedTempFile::new()?; // TODO implement default config and getting info from integrity verifier
-        let cpu_air_params = NamedTempFile::new()?; // TODO implement default config and getting info from integrity verifier
-
-        let task = Command::new("cpu_air_prover")
-            .args(["--out_file", out_file.path().to_string_lossy().as_ref()])
-            .args([
-                "--air_private_input",
-                job_trace.air_private_input.path().to_string_lossy().as_ref(),
-            ])
-            .args([
-                "--air_public_input",
-                job_trace.air_public_input.path().to_string_lossy().as_ref(),
-            ])
-            .args([
-                "--cpu_air_prover_config",
-                cpu_air_prover_config.path().to_string_lossy().as_ref(),
-            ])
-            .args(["--cpu_air_params", cpu_air_params.path().to_string_lossy().as_ref()])
-            .arg("--generate_annotations")
-            .spawn()?;
-
-        let job_trace_hash = hash!(job_trace);
-
-        debug!("task {} spawned", job_trace_hash);
-        self.tasks.insert(job_trace_hash.to_owned(), task);
-
-        let task_status = self
-            .tasks
-            .get_mut(&job_trace_hash)
-            .ok_or(ProverControllerError::TaskNotFound)?
-            .wait()
-            .await?;
-
-        trace!("task {} woke up", job_trace_hash);
-        if !task_status.success() {
-            debug!("task terminated {}", job_trace_hash);
-            return Err(ProverControllerError::TaskTerminated);
-        }
-
-        let task_output = self
-            .tasks
-            .remove(&job_trace_hash)
-            .ok_or(ProverControllerError::TaskNotFound)?
-            .wait_with_output()
-            .await?;
-        trace!("task {} output {:?}", job_trace_hash, task_output);
-
-        let mut raw_proof = String::new();
-        out_file.read_to_string(&mut raw_proof)?;
-
-        let parsed_proof = cairo_proof_parser::parse(raw_proof)
-            .map_err(|e| ProverControllerError::ProofParseError(e.to_string()))?;
-
-        let config: VecFelt252 = serde_json::from_str(&parsed_proof.config.to_string())?;
-        let public_input: VecFelt252 =
-            serde_json::from_str(&parsed_proof.public_input.to_string())?;
-        let unsent_commitment: VecFelt252 =
-            serde_json::from_str(&parsed_proof.unsent_commitment.to_string())?;
-        let witness: VecFelt252 = serde_json::from_str(&parsed_proof.witness.to_string())?;
-
-        let proof = chain!(
-            config.into_iter(),
-            public_input.into_iter(),
-            unsent_commitment.into_iter(),
-            witness.into_iter()
-        )
-        .collect_vec();
-
-        Ok(JobWitness { proof })
-    }
-
-    fn terminate(&mut self, job_trace_hash: u64) -> Result<(), ProverControllerError> {
-        self.tasks
-            .get_mut(&job_trace_hash)
-            .ok_or(ProverControllerError::TaskNotFound)?
-            .start_kill()?;
-        trace!("task scheduled for termination {}", job_trace_hash);
-        Ok(())
-    }
-
-    fn drop(mut self) -> Result<(), ProverControllerError> {
-        let keys: Vec<u64> = self.tasks.keys().cloned().collect();
-        for job_trace_hash in keys.iter() {
-            self.tasks
-                .get_mut(job_trace_hash)
-                .ok_or(ProverControllerError::TaskNotFound)?
-                .start_kill()?;
-            trace!("task scheduled for termination {}", job_trace_hash);
-        }
-        Ok(())
+    type ProcessResult = Result<JobWitness, ProverControllerError>;
+    fn run(
+        &self,
+        job_trace: JobTrace,
+    ) -> Result<Process<Self::ProcessResult>, ProverControllerError> {
+        let (terminate_tx, mut terminate_rx) = mpsc::channel::<()>(10);
+        let future: Pin<Box<dyn Future<Output = Self::ProcessResult> + '_>> =
+            Box::pin(async move {
+                let mut out_file = NamedTempFile::new()?;
+
+                let mut cpu_air_prover_config = NamedTempFile::new()?;
+                let mut cpu_air_params = NamedTempFile::new()?;
+
+                cpu_air_prover_config
+                    .write_all(&serde_json::to_string(&self.cpu_air_prover_config)?.into_bytes())?;
+                cpu_air_params
+                    .write_all(&serde_json::to_string(&self.cpu_air_params)?.into_bytes())?;
+
+                let mut task = Command::new("cpu_air_prover")
+                    .arg("--out_file")
+                    .arg(out_file.path())
+                    .arg("--private_input_file")
+                    .arg(job_trace.air_private_input.path())
+                    .arg("--public_input_file")
+                    .arg(job_trace.air_public_input.path())
+                    .arg("--prover_config_file")
+                    .arg(cpu_air_prover_config.path())
+                    .arg("--parameter_file")
+                    .arg(cpu_air_params.path())
+                    .arg("--generate_annotations")
+                    .stdout(Stdio::null())
+                    .spawn()?;
+
+                let job_trace_hash = hash!(job_trace);
+
+                debug!("task {} spawned", job_trace_hash);
+
+                loop {
+                    select! {
+                        output = task.wait() => {
+                            debug!("{:?}", output);
+                            if !output?.success() {
+                                return Err(ProverControllerError::TaskTerminated);
+                            }
+                            let output = task.wait_with_output().await?;
+                            debug!("{:?}", output);
+                            break;
+                        }
+                        Some(()) = terminate_rx.recv() => {
+                            task.start_kill()?;
+                        }
+                    }
+                }
+
+                let mut raw_proof = String::new();
+                out_file.read_to_string(&mut raw_proof)?;
+
+                let parsed_proof = cairo_proof_parser::parse(raw_proof)
+                    .map_err(|e| ProverControllerError::ProofParseError(e.to_string()))?;
+
+                let config: VecFelt252 = serde_json::from_str(&parsed_proof.config.to_string())?;
+                let public_input: VecFelt252 =
+                    serde_json::from_str(&parsed_proof.public_input.to_string())?;
+                let unsent_commitment: VecFelt252 =
+                    serde_json::from_str(&parsed_proof.unsent_commitment.to_string())?;
+                let witness: VecFelt252 = serde_json::from_str(&parsed_proof.witness.to_string())?;
+
+                let proof = chain!(
+                    config.into_iter(),
+                    public_input.into_iter(),
+                    unsent_commitment.into_iter(),
+                    witness.into_iter()
+                )
+                .collect_vec();
+
+                Ok(JobWitness { proof })
+            });
+
+        Ok(Process::new(future, terminate_tx))
     }
 }

crates/prover/src/stone_prover/tests/mod.rs

@@ -0,0 +1,6 @@
+pub mod models;
+
+#[cfg(test)]
+pub mod multiple_job;
+#[cfg(test)]
+pub mod single_job;