Erasure fixes and add erasure to ci

ci/buildkite.yml

@@ -11,6 +11,8 @@ steps:
     name: "cuda"
   - command: "ci/shellcheck.sh"
     name: "shellcheck [public]"
+  - command: "ci/test-erasure.sh"
+    name: "erasure"
   - wait
   - command: "ci/publish.sh"
     name: "publish release artifacts"

ci/test-erasure.sh

@@ -0,0 +1,29 @@
+#!/bin/bash -e
+
+set -o xtrace
+
+cd "$(dirname "$0")/.."
+
+if [[ -z "${libgf_complete_URL:-}" ]]; then
+  echo libgf_complete_URL undefined
+  exit 1
+fi
+
+if [[ -z "${libJerasure_URL:-}" ]]; then
+  echo libJerasure_URL undefined
+  exit 1
+fi
+
+curl -X GET -o libJerasure.so "$libJerasure_URL"
+curl -X GET -o libgf_complete.so "$libgf_complete_URL"
+
+ln -s libJerasure.so libJerasure.so.2
+ln -s libJerasure.so libJerasure.so.2.0.0
+ln -s libgf_complete.so libgf_complete.so.1.0.0
+export LD_LIBRARY_PATH=$PWD:$LD_LIBRARY_PATH
+
+# shellcheck disable=SC1090    # <-- shellcheck can't follow ~
+source ~/.cargo/env
+cargo test --features="erasure"
+
+exit 0

src/bin/client-demo.rs

@@ -1,3 +1,4 @@
+extern crate env_logger;
 extern crate getopts;
 extern crate isatty;
 extern crate pnet;
@@ -49,6 +50,7 @@ fn get_ip_addr() -> Option<IpAddr> {
 }
 
 fn main() {
+    env_logger::init().unwrap();
     let mut threads = 4usize;
     let mut num_nodes = 1usize;
 

src/crdt.rs

@@ -229,14 +229,38 @@ impl Crdt {
         }
     }
 
+    pub fn index_blobs(
+        obj: &Arc<RwLock<Self>>,
+        blobs: &Vec<SharedBlob>,
+        receive_index: &mut u64,
+    ) -> Result<()> {
+        let me: ReplicatedData = {
+            let robj = obj.read().expect("'obj' read lock in crdt::index_blobs");
+            debug!("broadcast table {}", robj.table.len());
+            robj.table[&robj.me].clone()
+        };
+
+        // enumerate all the blobs, those are the indices
+        for (i, b) in blobs.iter().enumerate() {
+            // only leader should be broadcasting
+            let mut blob = b.write().expect("'blob' write lock in crdt::index_blobs");
+            blob.set_id(me.id).expect("set_id in pub fn broadcast");
+            blob.set_index(*receive_index + i as u64)
+                .expect("set_index in pub fn broadcast");
+        }
+
+        Ok(())
+    }
+
     /// broadcast messages from the leader to layer 1 nodes
     /// # Remarks
     /// We need to avoid having obj locked while doing any io, such as the `send_to`
     pub fn broadcast(
         obj: &Arc<RwLock<Self>>,
-        blobs: &Vec<SharedBlob>,
+        window: &Arc<RwLock<Vec<Option<SharedBlob>>>>,
         s: &UdpSocket,
         transmit_index: &mut u64,
+        received_index: u64,
     ) -> Result<()> {
         let (me, table): (ReplicatedData, Vec<ReplicatedData>) = {
             // copy to avoid locking during IO
@@ -266,31 +290,35 @@ impl Crdt {
             return Err(Error::CrdtTooSmall);
         }
         trace!("nodes table {}", nodes.len());
-        trace!("blobs table {}", blobs.len());
-        // enumerate all the blobs, those are the indices
+
+        // enumerate all the blobs in the window, those are the indices
         // transmit them to nodes, starting from a different node
-        let orders: Vec<_> = blobs
-            .iter()
-            .enumerate()
-            .zip(
-                nodes
-                    .iter()
-                    .cycle()
-                    .skip((*transmit_index as usize) % nodes.len()),
-            )
-            .collect();
+        let mut orders = Vec::new();
+        let window_l = window.write().unwrap();
+        for i in *transmit_index..received_index {
+            let is = i as usize;
+            let k = is % window_l.len();
+            assert!(window_l[k].is_some());
+
+            orders.push((window_l[k].clone(), nodes[is % nodes.len()]));
+        }
+
         trace!("orders table {}", orders.len());
         let errs: Vec<_> = orders
             .into_iter()
-            .map(|((i, b), v)| {
+            .map(|(b, v)| {
                 // only leader should be broadcasting
                 assert!(me.current_leader_id != v.id);
-                let mut blob = b.write().expect("'b' write lock in pub fn broadcast");
-                blob.set_id(me.id).expect("set_id in pub fn broadcast");
-                blob.set_index(*transmit_index + i as u64)
-                    .expect("set_index in pub fn broadcast");
+                let bl = b.unwrap();
+                let blob = bl.read().expect("blob read lock in streamer::broadcast");
                 //TODO profile this, may need multiple sockets for par_iter
-                trace!("broadcast {} to {}", blob.meta.size, v.replicate_addr);
+                trace!(
+                    "broadcast idx: {} sz: {} to {} coding: {}",
+                    blob.get_index().unwrap(),
+                    blob.meta.size,
+                    v.replicate_addr,
+                    blob.is_coding()
+                );
                 assert!(blob.meta.size < BLOB_SIZE);
                 let e = s.send_to(&blob.data[..blob.meta.size], &v.replicate_addr);
                 trace!("done broadcast {} to {}", blob.meta.size, v.replicate_addr);