feat: support for streamed computations and their proofs (#1209)

* feat: support for streamed computations and their proofs

Introduce the `Op::Recv` LEM operator, which awaits for data to be
received by the channel terminal at LEM interpretation time.

This operator is used to implement the logic for the `StreamIn`
continuation, which receives the argument to be applied to the
(chaining) callable object.

The result will be available once the (new) `StreamOut` continuation
is reached. When that happens, the computation can be resumed by
supplying two pieces of data through the channel:
* A flag to tell if the machine should stutter while in `StreamOut`
* The next argument to be consumed in the incoming `StreamIn` state

Note: the second message should not be be sent if the machine is set
to stutter with the first message.

There is a test to show that we're now able to construct proofs of
computations that grow incrementally by starting from where we had
stopped. We don't need to start the folding process from the
beginning.

* optimize: skip new `StreamIn` ocurrences

Prepare the next call directly from `StreamOut` without having to
go through `StreamIn` again.

Since `StreamIn` only occurs in the very first frame after this
optimization, it was renamed to `StreamStart`. And `StreamOut`,
which now serves as literal point for pause while supporting both
input and output, was renamed to `StreamPause`.

benches/common/fib.rs

@@ -90,7 +90,9 @@ fn compute_coeffs<F: LurkField>(store: &Store<F>) -> (usize, usize) {
                 }
             }
         }
-        let frame = step_func.call_simple(&input, store, &lang, 0).unwrap();
+        let frame = step_func
+            .call_simple(&input, store, &lang, 0, &dummy_terminal())
+            .unwrap();
         input = frame.output.clone();
         iteration += 1;
     }

benches/end2end.rs

@@ -80,7 +80,9 @@ fn end2end_benchmark(c: &mut Criterion) {
             let ptr = go_base::<Bn>(&store, state.clone(), s.0, s.1);
             let frames =
                 evaluate::<Bn, Coproc<Bn>>(None, ptr, &store, limit, &dummy_terminal()).unwrap();
-            let _result = prover.prove_from_frames(&pp, &frames, &store).unwrap();
+            let _result = prover
+                .prove_from_frames(&pp, &frames, &store, None)
+                .unwrap();
         })
     });
 
@@ -220,7 +222,9 @@ fn prove_benchmark(c: &mut Criterion) {
             evaluate::<Bn, Coproc<Bn>>(None, ptr, &store, limit, &dummy_terminal()).unwrap();
 
         b.iter(|| {
-            let result = prover.prove_from_frames(&pp, &frames, &store).unwrap();
+            let result = prover
+                .prove_from_frames(&pp, &frames, &store, None)
+                .unwrap();
             black_box(result);
         })
     });
@@ -268,7 +272,9 @@ fn prove_compressed_benchmark(c: &mut Criterion) {
             evaluate::<Bn, Coproc<Bn>>(None, ptr, &store, limit, &dummy_terminal()).unwrap();
 
         b.iter(|| {
-            let (proof, _, _, _) = prover.prove_from_frames(&pp, &frames, &store).unwrap();
+            let (proof, _, _, _) = prover
+                .prove_from_frames(&pp, &frames, &store, None)
+                .unwrap();
 
             let compressed_result = proof.compress(&pp).unwrap();
             black_box(compressed_result);
@@ -313,8 +319,9 @@ fn verify_benchmark(c: &mut Criterion) {
             let prover = NovaProver::new(reduction_count, lang_rc.clone());
             let frames =
                 evaluate::<Bn, Coproc<Bn>>(None, ptr, &store, limit, &dummy_terminal()).unwrap();
-            let (proof, z0, zi, _num_steps) =
-                prover.prove_from_frames(&pp, &frames, &store).unwrap();
+            let (proof, z0, zi, _num_steps) = prover
+                .prove_from_frames(&pp, &frames, &store, None)
+                .unwrap();
 
             b.iter_batched(
                 || z0.clone(),
@@ -367,8 +374,9 @@ fn verify_compressed_benchmark(c: &mut Criterion) {
             let prover = NovaProver::new(reduction_count, lang_rc.clone());
             let frames =
                 evaluate::<Bn, Coproc<Bn>>(None, ptr, &store, limit, &dummy_terminal()).unwrap();
-            let (proof, z0, zi, _num_steps) =
-                prover.prove_from_frames(&pp, &frames, &store).unwrap();
+            let (proof, z0, zi, _num_steps) = prover
+                .prove_from_frames(&pp, &frames, &store, None)
+                .unwrap();
 
             let compressed_proof = proof.compress(&pp).unwrap();
 

benches/fibonacci.rs

@@ -114,7 +114,7 @@ fn fibonacci_prove<M: measurement::Measurement>(
             b.iter_batched(
                 || frames,
                 |frames| {
-                    let result = prover.prove_from_frames(&pp, frames, &store);
+                    let result = prover.prove_from_frames(&pp, frames, &store, None);
                     let _ = black_box(result);
                 },
                 BatchSize::LargeInput,

benches/sha256.rs

@@ -146,7 +146,7 @@ fn sha256_ivc_prove<M: measurement::Measurement>(
             b.iter_batched(
                 || frames,
                 |frames| {
-                    let result = prover.prove_from_frames(&pp, frames, store);
+                    let result = prover.prove_from_frames(&pp, frames, store, None);
                     let _ = black_box(result);
                 },
                 BatchSize::LargeInput,
@@ -234,7 +234,8 @@ fn sha256_ivc_prove_compressed<M: measurement::Measurement>(
             b.iter_batched(
                 || frames,
                 |frames| {
-                    let (proof, _, _, _) = prover.prove_from_frames(&pp, frames, store).unwrap();
+                    let (proof, _, _, _) =
+                        prover.prove_from_frames(&pp, frames, store, None).unwrap();
                     let compressed_result = proof.compress(&pp).unwrap();
 
                     let _ = black_box(compressed_result);
@@ -325,7 +326,7 @@ fn sha256_nivc_prove<M: measurement::Measurement>(
             b.iter_batched(
                 || frames,
                 |frames| {
-                    let result = prover.prove_from_frames(&pp, frames, store);
+                    let result = prover.prove_from_frames(&pp, frames, store, None);
                     let _ = black_box(result);
                 },
                 BatchSize::LargeInput,

chain-server/src/server.rs

@@ -156,7 +156,7 @@ where
                 // prove then compress the proof
                 let (proof, ..) = self
                     .prover
-                    .prove_from_frames(pp, &frames, &self.store)
+                    .prove_from_frames(pp, &frames, &self.store, None)
                     .map_err(|e| Status::internal(e.to_string()))?;
                 let proof = proof
                     .compress(pp)

examples/keccak.rs

@@ -395,7 +395,7 @@ fn main() {
     let proof_start = Instant::now();
 
     let (proof, z0, zi, _) = supernova_prover
-        .prove_from_frames(&pp, &frames, store)
+        .prove_from_frames(&pp, &frames, store, None)
         .unwrap();
 
     let proof_end = proof_start.elapsed();

examples/sha256_nivc.rs

@@ -104,7 +104,7 @@ fn main() {
     let (proof, z0, zi, _num_steps) = tracing_texray::examine(tracing::info_span!("bang!"))
         .in_scope(|| {
             supernova_prover
-                .prove_from_frames(&pp, &frames, store)
+                .prove_from_frames(&pp, &frames, store, None)
                 .unwrap()
         });
     let proof_end = proof_start.elapsed();

examples/tp_table.rs

@@ -176,7 +176,7 @@ fn main() {
                 let mut timings = Vec::with_capacity(n_samples);
                 for _ in 0..n_samples {
                     let start = Instant::now();
-                    let result = prover.prove_from_frames(&pp, frames, &store);
+                    let result = prover.prove_from_frames(&pp, frames, &store, None);
                     let _ = black_box(result);
                     let end = start.elapsed().as_secs_f64();
                     timings.push(end);

src/cli/repl/mod.rs

@@ -355,7 +355,7 @@ where
 
                     info!("Proving with NovaProver");
                     let (proof, public_inputs, public_outputs, num_steps) =
-                        prover.prove_from_frames(&pp, frames, &self.store)?;
+                        prover.prove_from_frames(&pp, frames, &self.store, None)?;
                     info!("Compressing Nova proof");
                     let proof = proof.compress(&pp)?;
                     assert_eq!(self.rc * num_steps, pad(n_frames, self.rc));
@@ -370,7 +370,7 @@ where
 
                     info!("Proving with SuperNovaProver");
                     let (proof, public_inputs, public_outputs, _num_steps) =
-                        prover.prove_from_frames(&pp, frames, &self.store)?;
+                        prover.prove_from_frames(&pp, frames, &self.store, None)?;
                     info!("Compressing SuperNova proof");
                     let proof = proof.compress(&pp)?;
                     assert!(proof.verify(&pp, &public_inputs, &public_outputs)?);

src/coroutine/memoset/prove.rs

@@ -96,7 +96,7 @@ impl<'a, F: CurveCycleEquipped, Q: Query<F> + Send + Sync>
     RecursiveSNARKTrait<F, Coroutine<'a, F, Q>> for Proof<F, Coroutine<'a, F, Q>>
 {
     type PublicParams = PublicParams<F>;
-
+    type BaseRecursiveSNARK = RecursiveSNARK<E1<F>>;
     type ErrorType = SuperNovaError;
 
     #[tracing::instrument(skip_all, name = "supernova::prove_recursively")]
@@ -105,8 +105,9 @@ impl<'a, F: CurveCycleEquipped, Q: Query<F> + Send + Sync>
         z0: &[F],
         steps: Vec<Coroutine<'a, F, Q>>,
         _store: &Store<F>,
+        init: Option<RecursiveSNARK<E1<F>>>,
     ) -> Result<Self, ProofError> {
-        let mut recursive_snark_option: Option<RecursiveSNARK<E1<F>>> = None;
+        let mut recursive_snark_option = init;
 
         let z0_primary = z0;
         let z0_secondary = Self::z0_secondary();
@@ -251,7 +252,7 @@ impl<'a, F: CurveCycleEquipped, Q: Query<F> + Send + Sync> MemosetProver<'a, F,
 
         let num_steps = steps.len();
 
-        let prove_output = Proof::prove_recursively(pp, &z0, steps, store)?;
+        let prove_output = Proof::prove_recursively(pp, &z0, steps, store, None)?;
         let zi = match prove_output {
             Proof::Recursive(ref snark, ..) => snark.zi_primary().clone(),
             Proof::Compressed(..) => unreachable!(),

src/lem/circuit.rs

@@ -1121,6 +1121,16 @@ fn synthesize_block<F: LurkField, CS: ConstraintSystem<F>, C: Coprocessor<F>>(
                 bound_allocations.insert_ptr(tgt[1].clone(), rem_ptr);
             }
             Op::Emit(_) | Op::Unit(_) => (),
+            Op::Recv(tgt) => {
+                let ptr = if let Ok(val) = ctx.bindings.get(tgt) {
+                    *val.get_ptr().expect("Received data must be a pointer")
+                } else {
+                    ctx.store.dummy()
+                };
+                let z_ptr = || ctx.store.hash_ptr(&ptr);
+                let a_ptr = AllocatedPtr::alloc_infallible(ns!(cs, format!("recv {tgt}")), z_ptr);
+                bound_allocations.insert_ptr(tgt.clone(), a_ptr);
+            }
             Op::Hide(tgt, sec, pay) => {
                 let sec = bound_allocations.get_ptr(sec)?;
                 let pay = bound_allocations.get_ptr(pay)?;
@@ -1593,7 +1603,12 @@ impl Func {
                         // three implies_u64, one sub and one linear
                         num_constraints += 197;
                     }
-                    Op::Not(..) | Op::Emit(_) | Op::Cproc(..) | Op::Copy(..) | Op::Unit(_) => (),
+                    Op::Not(..)
+                    | Op::Emit(_)
+                    | Op::Recv(_)
+                    | Op::Cproc(..)
+                    | Op::Copy(..)
+                    | Op::Unit(_) => (),
                     Op::Cons2(_, tag, _) => {
                         // tag for the image
                         globals.insert(FWrap(tag.to_field()));

src/lem/coroutine/eval.rs

@@ -200,9 +200,11 @@ fn run<F: LurkField>(
                 bindings.insert_ptr(tgt[1].clone(), c2);
             }
             Op::Emit(a) => {
+                // TODO: send `a` through a channel as in the original interpreter
                 let a = bindings.get_ptr(a)?;
                 println!("{}", a.fmt_to_string_simple(&scope.store));
             }
+            Op::Recv(_) => todo!("not supported yet"),
             Op::Cons2(img, tag, preimg) => {
                 let preimg_ptrs = bindings.get_many_ptr(preimg)?;
                 let tgt_ptr = intern_ptrs!(scope.store, *tag, preimg_ptrs[0], preimg_ptrs[1]);

src/lem/coroutine/synthesis.rs

@@ -458,6 +458,7 @@ fn synthesize_run<'a, F: LurkField, CS: ConstraintSystem<F>>(
                 bound_allocations.insert_ptr(tgt[1].clone(), rem_ptr);
             }
             Op::Emit(_) | Op::Unit(_) => (),
+            Op::Recv(_) => todo!("not supported yet"),
             Op::Hide(tgt, sec, pay) => {
                 let sec = bound_allocations.get_ptr(sec)?;
                 let pay = bound_allocations.get_ptr(pay)?;