Fixes for serializing entries over blobs and reorg into ledger

src/accountant.rs

@@ -6,6 +6,7 @@
 extern crate libc;
 
 use chrono::prelude::*;
+use entry::Entry;
 use event::Event;
 use hash::Hash;
 use mint::Mint;
@@ -15,8 +16,8 @@ use signature::{KeyPair, PublicKey, Signature};
 use std::collections::hash_map::Entry::Occupied;
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::result;
-use std::sync::atomic::{AtomicIsize, Ordering};
 use std::sync::RwLock;
+use std::sync::atomic::{AtomicIsize, Ordering};
 use transaction::Transaction;
 
 pub const MAX_ENTRY_IDS: usize = 1024 * 4;
@@ -232,6 +233,16 @@ impl Accountant {
         results
     }
 
+    pub fn process_verified_entries(&self, entries: Vec<Entry>) -> Result<()> {
+        for entry in entries {
+            self.register_entry_id(&entry.id);
+            for result in self.process_verified_events(entry.events) {
+                result?;
+            }
+        }
+        Ok(())
+    }
+
     /// Process a Witness Signature that has already been verified.
     fn process_verified_sig(&self, from: PublicKey, tx_sig: Signature) -> Result<()> {
         if let Occupied(mut e) = self.pending.write().unwrap().entry(tx_sig) {

src/bin/testnode.rs

@@ -17,8 +17,8 @@ use std::env;
 use std::io::{stdin, stdout, Read};
 use std::net::UdpSocket;
 use std::process::exit;
-use std::sync::atomic::AtomicBool;
 use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
 
 fn print_usage(program: &str, opts: Options) {
     let mut brief = format!("Usage: cat <transaction.log> | {} [options]\n\n", program);

src/ecdsa.rs

@@ -135,8 +135,8 @@ mod tests {
     use packet::{Packet, Packets, SharedPackets};
     use std::sync::RwLock;
     use thin_client_service::Request;
-    use transaction::test_tx;
     use transaction::Transaction;
+    use transaction::test_tx;
 
     fn make_packet_from_transaction(tr: Transaction) -> Packet {
         let tx = serialize(&Request::Transaction(tr)).unwrap();

src/hash.rs

@@ -1,7 +1,7 @@
 //! The `hash` module provides functions for creating SHA-256 hashes.
 
-use generic_array::typenum::U32;
 use generic_array::GenericArray;
+use generic_array::typenum::U32;
 use sha2::{Digest, Sha256};
 
 pub type Hash = GenericArray<u8, U32>;

src/historian.rs

@@ -4,8 +4,8 @@
 use entry::Entry;
 use hash::Hash;
 use recorder::{ExitReason, Recorder, Signal};
-use std::sync::mpsc::{channel, Receiver, Sender, TryRecvError};
 use std::sync::Mutex;
+use std::sync::mpsc::{channel, Receiver, Sender, TryRecvError};
 use std::thread::{spawn, JoinHandle};
 use std::time::Instant;
 

src/ledger.rs

@@ -1,9 +1,17 @@
 //! The `ledger` module provides functions for parallel verification of the
 //! Proof of History ledger.
 
+use bincode::{deserialize, serialize_into};
 use entry::{next_tick, Entry};
+use event::Event;
 use hash::Hash;
+use packet;
+use packet::{SharedBlob, BLOB_DATA_SIZE, BLOB_SIZE};
 use rayon::prelude::*;
+use std::cmp::min;
+use std::collections::VecDeque;
+use std::io::Cursor;
+use std::mem::size_of;
 
 pub trait Block {
     /// Verifies the hashes and counts of a slice of events are all consistent.
@@ -30,10 +38,104 @@ pub fn next_ticks(start_hash: &Hash, num_hashes: u64, len: usize) -> Vec<Entry>
     ticks
 }
 
+pub fn process_entry_list_into_blobs(
+    list: &Vec<Entry>,
+    blob_recycler: &packet::BlobRecycler,
+    q: &mut VecDeque<SharedBlob>,
+) {
+    let mut start = 0;
+    let mut end = 0;
+    while start < list.len() {
+        let mut entries: Vec<Vec<Entry>> = Vec::new();
+        let mut total = 0;
+        for i in &list[start..] {
+            total += size_of::<Event>() * i.events.len();
+            total += size_of::<Entry>();
+            if total >= BLOB_DATA_SIZE {
+                break;
+            }
+            end += 1;
+        }
+        // See if we need to split the events
+        if end <= start {
+            let mut event_start = 0;
+            let num_events_per_blob = BLOB_DATA_SIZE / size_of::<Event>();
+            let total_entry_chunks =
+                (list[end].events.len() + num_events_per_blob - 1) / num_events_per_blob;
+            trace!(
+                "splitting events end: {} total_chunks: {}",
+                end,
+                total_entry_chunks
+            );
+            for _ in 0..total_entry_chunks {
+                let event_end = min(event_start + num_events_per_blob, list[end].events.len());
+                trace!("event start: {} end: {}", event_start, event_end);
+                let mut entry = Entry {
+                    num_hashes: list[end].num_hashes,
+                    id: list[end].id,
+                    events: list[end].events[event_start..event_end].to_vec(),
+                };
+                trace!("event[0]: {:?}", entry.events[0]);
+                entries.push(vec![entry]);
+                event_start = event_end;
+            }
+            end += 1;
+        } else {
+            entries.push(list[start..end].to_vec());
+        }
+
+        for entry in entries {
+            trace!(
+                "entry.id: {:?} num_hashes: {} num events: {}",
+                entry[0].id,
+                entry[0].num_hashes,
+                entry[0].events.len()
+            );
+            let b = blob_recycler.allocate();
+            let pos = {
+                let mut bd = b.write().unwrap();
+                let mut out = Cursor::new(bd.data_mut());
+                serialize_into(&mut out, &entry).expect("failed to serialize output");
+                out.position() as usize
+            };
+            assert!(pos < BLOB_SIZE);
+            b.write().unwrap().set_size(pos);
+            q.push_back(b);
+        }
+        start = end;
+    }
+}
+
+pub fn reconstruct_entries_from_blobs(blobs: &VecDeque<SharedBlob>) -> Vec<Entry> {
+    let mut entries_to_apply: Vec<Entry> = Vec::new();
+    let mut last_id = Hash::default();
+    for msgs in blobs {
+        let blob = msgs.read().unwrap();
+        let entries: Vec<Entry> = deserialize(&blob.data()[..blob.meta.size]).unwrap();
+        for entry in entries {
+            if entry.id == last_id {
+                trace!("same id!");
+                if let Some(last_entry) = entries_to_apply.last_mut() {
+                    last_entry.events.extend(entry.events);
+                }
+            } else {
+                last_id = entry.id;
+                entries_to_apply.push(entry);
+            }
+        }
+        //TODO respond back to leader with hash of the state
+    }
+    entries_to_apply
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
+    use entry;
     use hash::hash;
+    use packet::BlobRecycler;
+    use signature::{KeyPair, KeyPairUtil};
+    use transaction::Transaction;
 
     #[test]
     fn test_verify_slice() {
@@ -48,6 +150,24 @@ mod tests {
         bad_ticks[1].id = one;
         assert!(!bad_ticks.verify(&zero)); // inductive step, bad
     }
+
+    #[test]
+    fn test_entry_to_blobs() {
+        let zero = Hash::default();
+        let one = hash(&zero);
+        let keypair = KeyPair::new();
+        let tr0 = Event::Transaction(Transaction::new(&keypair, keypair.pubkey(), 1, one));
+        let events = vec![tr0.clone(); 10000];
+        let e0 = entry::create_entry(&zero, 0, events);
+
+        let entry_list = vec![e0.clone(); 1];
+        let blob_recycler = BlobRecycler::default();
+        let mut blob_q = VecDeque::new();
+        process_entry_list_into_blobs(&entry_list, &blob_recycler, &mut blob_q);
+        let entries = reconstruct_entries_from_blobs(&blob_q);
+
+        assert_eq!(entry_list, entries);
+    }
 }
 
 #[cfg(all(feature = "unstable", test))]

src/mint.rs

@@ -1,7 +1,7 @@
 //! The `mint` module is a library for generating the chain's genesis block.
 
-use entry::create_entry;
 use entry::Entry;
+use entry::create_entry;
 use event::Event;
 use hash::{hash, Hash};
 use ring::rand::SystemRandom;

src/result.rs

@@ -78,9 +78,9 @@ mod tests {
     use std::io;
     use std::io::Write;
     use std::net::SocketAddr;
-    use std::sync::mpsc::channel;
     use std::sync::mpsc::RecvError;
     use std::sync::mpsc::RecvTimeoutError;
+    use std::sync::mpsc::channel;
     use std::thread;
 
     fn addr_parse_error() -> Result<SocketAddr> {

src/signature.rs

@@ -1,7 +1,7 @@
 //! The `signature` module provides functionality for public, and private keys.
 
-use generic_array::typenum::{U32, U64};
 use generic_array::GenericArray;
+use generic_array::typenum::{U32, U64};
 use ring::signature::Ed25519KeyPair;
 use ring::{rand, signature};
 use untrusted;

src/streamer.rs

@@ -438,8 +438,8 @@ mod test {
     use std::sync::{Arc, RwLock};
     use std::thread::sleep;
     use std::time::Duration;
-    use streamer::{blob_receiver, receiver, responder, retransmitter, window};
     use streamer::{BlobReceiver, PacketReceiver};
+    use streamer::{blob_receiver, receiver, responder, retransmitter, window};
 
     fn get_msgs(r: PacketReceiver, num: &mut usize) {
         for _t in 0..5 {