Merge pull request #28 from QED-it/streaming

streaming and unified workspace

Changelog.md

@@ -1,3 +1,13 @@
+# Version v1.3.0, 2020-10, Rust streaming
+
+Rust:
+- Unified Workspace reader with streaming mode.
+- Supports reading from stdin stream.
+- Supports reading chunk by chunk from unlimited files.
+- CLI stats, validate, and simulate work in streaming mode.
+- *(breaking)* Renamed FileSink to WorkspaceSink.
+
+
 # Version v1.2.1, 2020-10
 
 Rust:

rust/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "zkinterface"
-version = "1.2.1"
+version = "1.3.0"
 authors = ["Aurélien Nicolas <aurel@qed-it.com>"]
 license = "MIT"
 build = "build.rs"

rust/src/cli.rs

@@ -6,12 +6,7 @@ use std::io::{stdin, stdout, Read, Write, copy};
 use std::path::{Path, PathBuf};
 use structopt::StructOpt;
 
-use crate::{
-    Reader,
-    Messages,
-    consumers::stats::Stats,
-    Result,
-};
+use crate::{Reader, Workspace, Messages, consumers::stats::Stats, Result};
 use std::fs::{File, create_dir_all};
 use std::ffi::OsStr;
 
@@ -91,9 +86,9 @@ pub fn cli(options: &Options) -> Result<()> {
         "to-json" => main_json(&load_messages(options)?),
         "to-yaml" => main_yaml(&load_messages(options)?),
         "explain" => main_explain(&load_messages(options)?),
-        "validate" => main_validate(&load_messages(options)?),
-        "simulate" => main_simulate(&load_messages(options)?),
-        "stats" => main_stats(&load_messages(options)?),
+        "validate" => main_validate(&stream_messages(options)?),
+        "simulate" => main_simulate(&stream_messages(options)?),
+        "stats" => main_stats(&stream_messages(options)?),
         "fake_prove" => main_fake_prove(&load_messages(options)?),
         "fake_verify" => main_fake_verify(&load_messages(options)?),
         "help" => {
@@ -116,7 +111,7 @@ fn load_messages(opts: &Options) -> Result<Reader> {
     for path in list_files(opts)? {
         if path == Path::new("-") {
             eprintln!("Loading from stdin");
-            reader.read_from(&mut stdin())?;
+            reader.read_from(stdin())?;
         } else {
             eprintln!("Loading file {}", path.display());
             reader.read_file(path)?;
@@ -127,6 +122,11 @@ fn load_messages(opts: &Options) -> Result<Reader> {
     Ok(reader)
 }
 
+fn stream_messages(opts: &Options) -> Result<Workspace> {
+    let paths = list_files(opts)?;
+    Ok(Workspace::new(paths))
+}
+
 fn has_zkif_extension(path: &Path) -> bool {
     path.extension() == Some(OsStr::new("zkif"))
 }
@@ -176,15 +176,17 @@ fn main_example(opts: &Options) -> Result<()> {
     } else {
         create_dir_all(out_dir)?;
 
-        let path = out_dir.join("statement.zkif");
-        let mut file = File::create(&path)?;
-        example_circuit_header().write_into(&mut file)?;
-        example_constraints().write_into(&mut file)?;
+        let path = out_dir.join("header.zkif");
+        example_circuit_header().write_into(&mut File::create(&path)?)?;
         eprintln!("Written {}", path.display());
 
         let path = out_dir.join("witness.zkif");
         example_witness().write_into(&mut File::create(&path)?)?;
         eprintln!("Written {}", path.display());
+
+        let path = out_dir.join("constraints.zkif");
+        example_constraints().write_into(&mut File::create(&path)?)?;
+        eprintln!("Written {}", path.display());
     }
     Ok(())
 }
@@ -217,25 +219,21 @@ fn main_explain(reader: &Reader) -> Result<()> {
     Ok(())
 }
 
-fn main_validate(reader: &Reader) -> Result<()> {
-    let reader = Messages::from(reader);
-
+fn main_validate(ws: &Workspace) -> Result<()> {
     // Validate semantics as verifier.
     let mut validator = Validator::new_as_verifier();
-    validator.ingest_messages(&reader);
+    validator.ingest_workspace(ws);
     print_violations(&validator.get_violations())
 }
 
-fn main_simulate(reader: &Reader) -> Result<()> {
-    let reader = Messages::from(reader);
-
+fn main_simulate(ws: &Workspace) -> Result<()> {
     // Validate semantics as prover.
     let mut validator = Validator::new_as_prover();
-    validator.ingest_messages(&reader);
+    validator.ingest_workspace(ws);
     print_violations(&validator.get_violations())?;
 
     // Check whether the statement is true.
-    let ok = Simulator::default().simulate(&reader);
+    let ok = Simulator::default().simulate_workspace(ws);
     match ok {
         Err(_) => eprintln!("The statement is NOT TRUE!"),
         Ok(_) => eprintln!("The statement is TRUE!"),
@@ -254,9 +252,9 @@ fn print_violations(errors: &[String]) -> Result<()> {
     }
 }
 
-fn main_stats(reader: &Reader) -> Result<()> {
-    let mut stats = Stats::new();
-    stats.push(reader)?;
+fn main_stats(ws: &Workspace) -> Result<()> {
+    let mut stats = Stats::default();
+    stats.ingest_workspace(ws);
     serde_json::to_writer_pretty(stdout(), &stats)?;
     println!();
     Ok(())

rust/src/consumers/mod.rs

@@ -2,3 +2,4 @@ pub mod validator;
 pub mod simulator;
 pub mod stats;
 pub mod reader;
+pub mod workspace;

rust/src/consumers/reader.rs

@@ -62,7 +62,7 @@ pub fn split_messages(mut buf: &[u8]) -> Vec<&[u8]> {
     bufs
 }
 
-pub fn read_buffer(stream: &mut impl Read) -> Result<Vec<u8>> {
+pub fn read_buffer(mut stream: impl Read) -> Result<Vec<u8>> {
     let mut buffer = vec![0u8; 4];
     if stream.read_exact(&mut buffer).is_err() {
         return Ok(Vec::new()); // End of stream at the correct place.
@@ -164,9 +164,9 @@ impl Reader {
         Ok(())
     }
 
-    pub fn read_from(&mut self, reader: &mut impl Read) -> Result<()> {
+    pub fn read_from(&mut self, mut reader: impl Read) -> Result<()> {
         loop {
-            let buffer = read_buffer(reader)?;
+            let buffer = read_buffer(&mut reader)?;
             if buffer.len() == 0 {
                 return Ok(());
             }

rust/src/consumers/simulator.rs

@@ -1,4 +1,4 @@
-use crate::{Result, CircuitHeader, Witness, ConstraintSystem, Messages, Variables};
+use crate::{Result, CircuitHeader, Witness, ConstraintSystem, Messages, Variables, Workspace, Message};
 use crate::structs::constraints::BilinearConstraint;
 
 use std::collections::HashMap;
@@ -12,6 +12,8 @@ type Field = BigUint;
 pub struct Simulator {
     values: HashMap<Var, Field>,
     modulus: Field,
+
+    violations: Vec<String>,
 }
 
 impl Simulator {
@@ -28,6 +30,25 @@ impl Simulator {
         Ok(())
     }
 
+    pub fn simulate_workspace(&mut self, ws: &Workspace) -> Result<()> {
+        for msg in ws.iter_messages() {
+            match msg {
+                Message::Header(header) => {
+                    self.ingest_header(&header)?;
+                }
+                Message::ConstraintSystem(cs) => {
+                    self.ingest_constraint_system(&cs)?;
+                }
+                Message::Witness(witness) => {
+                    self.ingest_witness(&witness)?;
+                }
+                Message::Command(_) => {}
+                Message::Err(_) => {}
+            }
+        }
+        Ok(())
+    }
+
     pub fn ingest_header(&mut self, header: &CircuitHeader) -> Result<()> {
         // Set the field.
         let max = header.field_maximum.as_ref().ok_or("No field_maximum specified")?;

rust/src/consumers/stats.rs

@@ -3,10 +3,9 @@ extern crate serde_json;
 
 use serde::{Deserialize, Serialize};
 
-use crate::Reader;
-use crate::Result;
+use crate::{Workspace, Message};
 
-#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize)]
+#[derive(Clone, Default, Debug, Eq, PartialEq, Deserialize, Serialize)]
 pub struct Stats {
     num_public_inputs: u64,
     num_private_variables: u64,
@@ -18,29 +17,38 @@ pub struct Stats {
 }
 
 impl Stats {
-    pub fn new() -> Stats {
-        return Stats { num_public_inputs: 0, num_private_variables: 0, multiplications: 0, additions_a: 0, additions_b: 0, additions_c: 0, additions: 0 };
-    }
+    pub fn ingest_workspace(&mut self, ws: &Workspace) {
+        for msg in ws.iter_messages() {
+            match msg {
+                Message::Header(header) => {
+                    self.num_public_inputs = header.instance_variables.variable_ids.len() as u64;
+                    self.num_private_variables = header.free_variable_id - self.num_public_inputs - 1;
+                }
 
-    pub fn push(&mut self, reader: &Reader) -> Result<()> {
-        let header = reader.last_header().ok_or("no circuit")?;
-        self.num_public_inputs = header.instance_variables().unwrap().variable_ids().unwrap().len() as u64;
-        self.num_private_variables = header.free_variable_id() - self.num_public_inputs - 1;
+                Message::ConstraintSystem(cs) => {
+                    self.multiplications += cs.constraints.len() as u64;
 
-        for constraint in reader.iter_constraints() {
-            self.multiplications += 1;
-            if constraint.a.len() > 0 {
-                self.additions_a += (constraint.a.len() - 1) as u64;
-            }
-            if constraint.b.len() > 0 {
-                self.additions_b += (constraint.b.len() - 1) as u64;
-            }
-            if constraint.c.len() > 0 {
-                self.additions_c += (constraint.c.len() - 1) as u64;
+                    for constraint in &cs.constraints {
+                        let len_a = constraint.linear_combination_a.variable_ids.len() as u64;
+                        if len_a > 0 {
+                            self.additions_a += len_a - 1;
+                        }
+
+                        let len_b = constraint.linear_combination_b.variable_ids.len() as u64;
+                        if len_b > 0 {
+                            self.additions_b += len_b - 1;
+                        }
+
+                        let len_c = constraint.linear_combination_c.variable_ids.len() as u64;
+                        if len_c > 0 {
+                            self.additions_c += len_c - 1;
+                        }
+                    }
+                    self.additions = self.additions_a + self.additions_b + self.additions_c;
+                }
+
+                _ => {}
             }
         }
-
-        self.additions = self.additions_a + self.additions_b + self.additions_c;
-        Ok(())
     }
 }

rust/src/consumers/validator.rs

@@ -1,4 +1,4 @@
-use crate::{CircuitHeader, Witness, ConstraintSystem, Messages, Variables};
+use crate::{CircuitHeader, Witness, ConstraintSystem, Messages, Variables, Message, Workspace};
 
 use std::collections::HashMap;
 use num_bigint::BigUint;
@@ -51,8 +51,31 @@ impl Validator {
         }
     }
 
+    pub fn ingest_workspace(&mut self, ws: &Workspace) {
+        for msg in ws.iter_messages() {
+            match msg {
+                Message::Header(header) => {
+                    self.ingest_header(&header);
+                }
+                Message::ConstraintSystem(cs) => {
+                    self.ingest_constraint_system(&cs);
+                }
+                Message::Witness(witness) => {
+                    if self.as_prover {
+                        self.ingest_witness(&witness);
+                    }
+                }
+                Message::Command(_) => {}
+                Message::Err(err) => self.violate(err.to_string()),
+            }
+        }
+    }
+
     pub fn get_violations(mut self) -> Vec<String> {
         self.ensure_all_variables_used();
+        if !self.got_header {
+            self.violate("Missing header.");
+        }
         self.violations
     }
 

rust/src/consumers/workspace.rs

@@ -0,0 +1,75 @@
+use std::path::{PathBuf, Path};
+use std::fs::File;
+use std::iter;
+use std::io::{Read, stdin};
+use crate::consumers::reader::read_buffer;
+use crate::Message;
+
+pub struct Workspace {
+    paths: Vec<PathBuf>,
+    stdin: bool,
+}
+
+impl Workspace {
+    pub fn new(mut paths: Vec<PathBuf>) -> Self {
+        if paths == vec![PathBuf::from("-")] {
+            Workspace { paths: vec![], stdin: true }
+        } else {
+            paths.sort();
+            paths.sort_by_key(|path| {
+                let name = path.file_name().unwrap().to_str().unwrap();
+                match () {
+                    _ if name.contains("header") => 0,
+                    _ if name.contains("witness") => 1,
+                    _ if name.contains("constraint") => 3,
+                    _ => 4,
+                }
+            });
+            Workspace { paths, stdin: false }
+        }
+    }
+
+    pub fn iter_messages<'w>(&'w self) -> impl Iterator<Item=Message> + 'w {
+        let buffers: Box<dyn Iterator<Item=Vec<u8>>> = if self.stdin {
+            Box::new(iterate_stream(stdin()))
+        } else {
+            Box::new(iterate_files(&self.paths))
+        };
+
+        buffers.map(|buffer| Message::from(&buffer[..]))
+    }
+}
+
+pub fn iterate_files<'w>(paths: &'w [PathBuf]) -> impl Iterator<Item=Vec<u8>> + 'w {
+    paths.iter().flat_map(|path|
+        iterate_file(path))
+}
+
+pub fn iterate_file(path: &Path) -> Box<dyn Iterator<Item=Vec<u8>>> {
+    match File::open(path) {
+        Err(err) => {
+            eprintln!("Error opening workspace file {}: {}", path.display(), err);
+            Box::new(iter::empty())
+        }
+        Ok(file) => Box::new(
+            iterate_stream(file)),
+    }
+}
+
+pub fn iterate_stream<'s>(mut stream: impl Read + 's) -> impl Iterator<Item=Vec<u8>> + 's {
+    iter::from_fn(move ||
+        match read_buffer(&mut stream) {
+            Err(err) => {
+                eprintln!("Error reading: {}", err);
+                None
+            }
+            Ok(buffer) => {
+                if buffer.len() == 0 {
+                    None
+                } else {
+                    Some(buffer)
+                }
+            }
+        }
+    )
+}