From 4944c965e4378bd87261c0538679b87f24c5ebdf Mon Sep 17 00:00:00 2001
From: Anatoly Yakovenko <aeyakovenko@gmail.com>
Date: Thu, 12 Apr 2018 10:26:32 -0700
Subject: [PATCH 01/10] update

heap

update

update

wip

use a vec and sort

builds

update

tests

update

fmt

update

progress

fmt

passes needs retransmit test

tests

cleanup

update

update

update

update

fmt
---
 src/lib.rs         |   1 +
 src/streamer.rs    | 192 ++++++++++++++++++++++++++++++++++++++++-----
 src/subscribers.rs | 136 ++++++++++++++++++++++++++++++++
 3 files changed, 309 insertions(+), 20 deletions(-)
 create mode 100644 src/subscribers.rs

diff --git a/src/lib.rs b/src/lib.rs
index 706e61fb87..54ae1656b8 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -15,6 +15,7 @@ pub mod recorder;
 pub mod result;
 pub mod signature;
 pub mod streamer;
+pub mod subscribers;
 pub mod transaction;
 extern crate bincode;
 extern crate byteorder;
diff --git a/src/streamer.rs b/src/streamer.rs
index 9ed4e2d800..6ada2c4e00 100644
--- a/src/streamer.rs
+++ b/src/streamer.rs
@@ -4,9 +4,10 @@ use std::collections::VecDeque;
 use std::net::UdpSocket;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::mpsc;
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
 use std::thread::{spawn, JoinHandle};
 use std::time::Duration;
+use subscribers;
 
 pub type PacketReceiver = mpsc::Receiver<SharedPackets>;
 pub type PacketSender = mpsc::Sender<SharedPackets>;
@@ -75,14 +76,77 @@ pub fn responder(
 
 //TODO, we would need to stick block authentication before we create the
 //window.
+fn recv_blobs(recycler: &BlobRecycler, sock: &UdpSocket, s: &BlobSender) -> Result<()> {
+    let dq = Blob::recv_from(recycler, sock)?;
+    if !dq.is_empty() {
+        s.send(dq)?;
+    }
+    Ok(())
+}
+
+pub fn blob_receiver(
+    exit: Arc<AtomicBool>,
+    recycler: BlobRecycler,
+    sock: UdpSocket,
+    s: BlobSender,
+) -> Result<JoinHandle<()>> {
+    //DOCUMENTED SIDE-EFFECT
+    //1 second timeout on socket read
+    let timer = Duration::new(1, 0);
+    sock.set_read_timeout(Some(timer))?;
+    let t = spawn(move || loop {
+        if exit.load(Ordering::Relaxed) {
+            break;
+        }
+        let _ = recv_blobs(&recycler, &sock, &s);
+    });
+    Ok(t)
+}
+
 fn recv_window(
     window: &mut Vec<Option<SharedBlob>>,
+    subs: &Arc<RwLock<subscribers::Subscribers>>,
     recycler: &BlobRecycler,
     consumed: &mut usize,
-    socket: &UdpSocket,
+    r: &BlobReceiver,
     s: &BlobSender,
+    cast: &BlobSender,
 ) -> Result<()> {
-    let mut dq = Blob::recv_from(recycler, socket)?;
+    let timer = Duration::new(1, 0);
+    let mut dq = r.recv_timeout(timer)?;
+    while let Ok(mut nq) = r.try_recv() {
+        dq.append(&mut nq)
+    }
+    {
+        //retransmit all leader blocks
+        let mut castq = VecDeque::new();
+        let rsubs = subs.read().unwrap();
+        for b in &dq {
+            let p = b.read().unwrap();
+            if p.meta.addr() == rsubs.leader.addr {
+                //TODO
+                //need to copy the retransmited blob
+                //otherwise we get into races with which thread
+                //should do the recycling
+                //
+                //a better absraction would be to recycle when the blob
+                //is dropped via a weakref to the recycler
+                let nv = recycler.allocate();
+                {
+                    let mut mnv = nv.write().unwrap();
+                    let sz = p.meta.size;
+                    mnv.meta.size = sz;
+                    mnv.data[..sz].copy_from_slice(&p.data[..sz]);
+                }
+                castq.push_back(nv);
+            }
+        }
+        if !castq.is_empty() {
+            cast.send(castq)?;
+        }
+    }
+    //send a contiguous set of blocks
+    let mut contq = VecDeque::new();
     while let Some(b) = dq.pop_front() {
         let b_ = b.clone();
         let mut p = b.write().unwrap();
@@ -97,46 +161,82 @@ fn recv_window(
             } else {
                 debug!("duplicate blob at index {:}", w);
             }
-            //send a contiguous set of blocks
-            let mut dq = VecDeque::new();
             loop {
                 let k = *consumed % NUM_BLOBS;
                 if window[k].is_none() {
                     break;
                 }
-                dq.push_back(window[k].clone().unwrap());
+                contq.push_back(window[k].clone().unwrap());
                 window[k] = None;
                 *consumed += 1;
             }
-            if !dq.is_empty() {
-                s.send(dq)?;
-            }
         }
     }
+    if !contq.is_empty() {
+        s.send(contq)?;
+    }
     Ok(())
 }
 
 pub fn window(
-    sock: UdpSocket,
     exit: Arc<AtomicBool>,
-    r: BlobRecycler,
+    subs: Arc<RwLock<subscribers::Subscribers>>,
+    recycler: BlobRecycler,
+    r: BlobReceiver,
     s: BlobSender,
+    cast: BlobSender,
 ) -> JoinHandle<()> {
     spawn(move || {
         let mut window = vec![None; NUM_BLOBS];
         let mut consumed = 0;
-        let timer = Duration::new(1, 0);
-        sock.set_read_timeout(Some(timer)).unwrap();
         loop {
-            if recv_window(&mut window, &r, &mut consumed, &sock, &s).is_err()
-                || exit.load(Ordering::Relaxed)
-            {
+            if exit.load(Ordering::Relaxed) {
                 break;
             }
+            let _ = recv_window(&mut window, &subs, &recycler, &mut consumed, &r, &s, &cast);
         }
     })
 }
 
+fn retransmit(
+    subs: &Arc<RwLock<subscribers::Subscribers>>,
+    recycler: &BlobRecycler,
+    r: &BlobReceiver,
+    sock: &UdpSocket,
+) -> Result<()> {
+    let timer = Duration::new(1, 0);
+    let mut dq = r.recv_timeout(timer)?;
+    while let Ok(mut nq) = r.try_recv() {
+        dq.append(&mut nq);
+    }
+    {
+        let wsubs = subs.read().unwrap();
+        for b in &dq {
+            let mut mb = b.write().unwrap();
+            wsubs.retransmit(&mut mb, sock)?;
+        }
+    }
+    while let Some(b) = dq.pop_front() {
+        recycler.recycle(b);
+    }
+    Ok(())
+}
+
+pub fn retransmitter(
+    sock: UdpSocket,
+    exit: Arc<AtomicBool>,
+    subs: Arc<RwLock<subscribers::Subscribers>>,
+    recycler: BlobRecycler,
+    r: BlobReceiver,
+) -> JoinHandle<()> {
+    spawn(move || loop {
+        if exit.load(Ordering::Relaxed) {
+            break;
+        }
+        let _ = retransmit(&subs, &recycler, &r, &sock);
+    })
+}
+
 #[cfg(all(feature = "unstable", test))]
 mod bench {
     extern crate test;
@@ -248,9 +348,11 @@ mod test {
     use std::net::UdpSocket;
     use std::sync::atomic::{AtomicBool, Ordering};
     use std::sync::mpsc::channel;
-    use std::sync::Arc;
+    use std::sync::{Arc, RwLock};
     use std::time::Duration;
-    use streamer::{receiver, responder, window, BlobReceiver, PacketReceiver};
+    use streamer::{blob_receiver, receiver, responder, retransmitter, window, BlobReceiver,
+                   PacketReceiver};
+    use subscribers::{Node, Subscribers};
 
     fn get_msgs(r: PacketReceiver, num: &mut usize) {
         for _t in 0..5 {
@@ -325,9 +427,24 @@ mod test {
         let addr = read.local_addr().unwrap();
         let send = UdpSocket::bind("127.0.0.1:0").expect("bind");
         let exit = Arc::new(AtomicBool::new(false));
+        let subs = Arc::new(RwLock::new(Subscribers::new(
+            Node::default(),
+            Node::default(),
+        )));
         let resp_recycler = BlobRecycler::default();
         let (s_reader, r_reader) = channel();
-        let t_receiver = window(read, exit.clone(), resp_recycler.clone(), s_reader);
+        let t_receiver =
+            blob_receiver(exit.clone(), resp_recycler.clone(), read, s_reader).unwrap();
+        let (s_window, r_window) = channel();
+        let (s_cast, _r_cast) = channel();
+        let t_window = window(
+            exit.clone(),
+            subs,
+            resp_recycler.clone(),
+            r_reader,
+            s_window,
+            s_cast,
+        );
         let (s_responder, r_responder) = channel();
         let t_responder = responder(send, exit.clone(), resp_recycler.clone(), r_responder);
         let mut msgs = VecDeque::new();
@@ -344,10 +461,45 @@ mod test {
         }
         s_responder.send(msgs).expect("send");
         let mut num = 0;
-        get_blobs(r_reader, &mut num);
+        get_blobs(r_window, &mut num);
         assert_eq!(num, 10);
         exit.store(true, Ordering::Relaxed);
         t_receiver.join().expect("join");
         t_responder.join().expect("join");
+        t_window.join().expect("join");
     }
+
+    #[test]
+    pub fn retransmit() {
+        let read = UdpSocket::bind("127.0.0.1:0").expect("bind");
+        let send = UdpSocket::bind("127.0.0.1:0").expect("bind");
+        let exit = Arc::new(AtomicBool::new(false));
+        let subs = Arc::new(RwLock::new(Subscribers::new(
+            Node::default(),
+            Node::default(),
+        )));
+        let n3 = Node::new([0; 8], 1, read.local_addr().unwrap());
+        subs.write().unwrap().insert(&[n3]);
+        let (s_cast, r_cast) = channel();
+        let re = BlobRecycler::default();
+        let saddr = send.local_addr().unwrap();
+        let t_retransmit = retransmitter(send, exit.clone(), subs, re.clone(), r_cast);
+        let mut bq = VecDeque::new();
+        let b = re.allocate();
+        b.write().unwrap().meta.size = 10;
+        bq.push_back(b);
+        s_cast.send(bq).unwrap();
+        let (s_recv, r_recv) = channel();
+        let t_receiver = blob_receiver(exit.clone(), re.clone(), read, s_recv).unwrap();
+        let mut oq = r_recv.recv().unwrap();
+        assert_eq!(oq.len(), 1);
+        let o = oq.pop_front().unwrap();
+        let ro = o.read().unwrap();
+        assert_eq!(ro.meta.size, 10);
+        assert_eq!(ro.meta.addr(), saddr);
+        exit.store(true, Ordering::Relaxed);
+        t_receiver.join().expect("join");
+        t_retransmit.join().expect("join");
+    }
+
 }
diff --git a/src/subscribers.rs b/src/subscribers.rs
new file mode 100644
index 0000000000..0a123540b3
--- /dev/null
+++ b/src/subscribers.rs
@@ -0,0 +1,136 @@
+use packet::Blob;
+use rayon::prelude::*;
+use result::{Error, Result};
+use std::net::{SocketAddr, UdpSocket};
+
+#[derive(Clone, PartialEq)]
+pub struct Node {
+    pub id: [u64; 8],
+    pub weight: u64,
+    pub addr: SocketAddr,
+}
+
+//sockaddr doesn't implement default
+impl Default for Node {
+    fn default() -> Node {
+        Node {
+            id: [0; 8],
+            weight: 0,
+            addr: "0.0.0.0:0".parse().unwrap(),
+        }
+    }
+}
+
+impl Node {
+    pub fn new(id: [u64; 8], weight: u64, addr: SocketAddr) -> Node {
+        Node {
+            id: id,
+            weight: weight,
+            addr: addr,
+        }
+    }
+    fn key(&self) -> i64 {
+        (self.weight as i64).checked_neg().unwrap()
+    }
+}
+
+pub struct Subscribers {
+    pub data: Vec<Node>,
+    pub me: Node,
+    pub leader: Node,
+}
+
+impl Subscribers {
+    pub fn new(me: Node, leader: Node) -> Subscribers {
+        let mut h = Subscribers {
+            data: vec![],
+            me: me.clone(),
+            leader: leader.clone(),
+        };
+        h.insert(&[me, leader]);
+        h
+    }
+    pub fn retransmit(&self, blob: &mut Blob, s: &UdpSocket) -> Result<()> {
+        let errs: Vec<_> = self.data
+            .par_iter()
+            .map(|i| {
+                if self.me == *i {
+                    return Ok(0);
+                }
+                if self.leader == *i {
+                    return Ok(0);
+                }
+                trace!("retransmit blob to {}", i.addr);
+                s.send_to(&blob.data[..blob.meta.size], &i.addr)
+            })
+            .collect();
+        for e in errs {
+            trace!("retransmit result {:?}", e);
+            match e {
+                Err(e) => return Err(Error::IO(e)),
+                _ => (),
+            }
+        }
+        Ok(())
+    }
+    pub fn insert(&mut self, ns: &[Node]) {
+        for n in ns.into_iter() {
+            self.data.push(n.clone());
+        }
+        self.data.sort_by_key(Node::key);
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use packet::Blob;
+    use rayon::prelude::*;
+    use std::net::UdpSocket;
+    use std::time::Duration;
+    use subscribers::{Node, Subscribers};
+
+    #[test]
+    pub fn subscriber() {
+        let mut me = Node::default();
+        me.weight = 10;
+        let mut leader = Node::default();
+        leader.weight = 11;
+        let mut s = Subscribers::new(me, leader);
+        assert_eq!(s.data.len(), 2);
+        assert_eq!(s.data[0].weight, 11);
+        assert_eq!(s.data[1].weight, 10);
+        let mut n = Node::default();
+        n.weight = 12;
+        s.insert(&[n]);
+        assert_eq!(s.data.len(), 3);
+        assert_eq!(s.data[0].weight, 12);
+    }
+    #[test]
+    pub fn retransmit() {
+        let s1 = UdpSocket::bind("127.0.0.1:0").expect("bind");
+        let s2 = UdpSocket::bind("127.0.0.1:0").expect("bind");
+        let s3 = UdpSocket::bind("127.0.0.1:0").expect("bind");
+        let n1 = Node::new([0; 8], 0, s1.local_addr().unwrap());
+        let n2 = Node::new([0; 8], 0, s2.local_addr().unwrap());
+        let mut s = Subscribers::new(n1.clone(), n2.clone());
+        let n3 = Node::new([0; 8], 0, s3.local_addr().unwrap());
+        s.insert(&[n3]);
+        let mut b = Blob::default();
+        b.meta.size = 10;
+        let s4 = UdpSocket::bind("127.0.0.1:0").expect("bind");
+        s.retransmit(&mut b, &s4).unwrap();
+        let res: Vec<_> = [s1, s2, s3]
+            .into_par_iter()
+            .map(|s| {
+                let mut b = Blob::default();
+                s.set_read_timeout(Some(Duration::new(1, 0))).unwrap();
+                s.recv_from(&mut b.data).is_err()
+            })
+            .collect();
+        assert_eq!(res, [true, true, false]);
+        let mut n4 = Node::default();
+        n4.addr = "255.255.255.255:1".parse().unwrap();
+        s.insert(&[n4]);
+        assert!(s.retransmit(&mut b, &s4).is_err());
+    }
+}

From 13485074ac51a78013ffe58f17d3695a22163af9 Mon Sep 17 00:00:00 2001
From: Anatoly Yakovenko <aeyakovenko@gmail.com>
Date: Mon, 16 Apr 2018 20:57:15 -0700
Subject: [PATCH 02/10] test cast

---
 src/streamer.rs | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/src/streamer.rs b/src/streamer.rs
index 6ada2c4e00..c2c10ac87a 100644
--- a/src/streamer.rs
+++ b/src/streamer.rs
@@ -429,14 +429,14 @@ mod test {
         let exit = Arc::new(AtomicBool::new(false));
         let subs = Arc::new(RwLock::new(Subscribers::new(
             Node::default(),
-            Node::default(),
+            Node::new([0;8], 0, send.local_addr().unwrap()),
         )));
         let resp_recycler = BlobRecycler::default();
         let (s_reader, r_reader) = channel();
         let t_receiver =
             blob_receiver(exit.clone(), resp_recycler.clone(), read, s_reader).unwrap();
         let (s_window, r_window) = channel();
-        let (s_cast, _r_cast) = channel();
+        let (s_cast, r_cast) = channel();
         let t_window = window(
             exit.clone(),
             subs,
@@ -463,6 +463,11 @@ mod test {
         let mut num = 0;
         get_blobs(r_window, &mut num);
         assert_eq!(num, 10);
+        let mut q = r_cast.recv().unwrap();
+        while let Ok(mut nq) = r_cast.try_recv() {
+            q.append(&mut nq);
+        }
+        assert_eq!(q.len(), 10);
         exit.store(true, Ordering::Relaxed);
         t_receiver.join().expect("join");
         t_responder.join().expect("join");

From fb55ab8c330b3af07c94e2cc7786ce2cd5f3ef4e Mon Sep 17 00:00:00 2001
From: Anatoly Yakovenko <aeyakovenko@gmail.com>
Date: Mon, 16 Apr 2018 21:02:37 -0700
Subject: [PATCH 03/10] format

---
 src/streamer.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/streamer.rs b/src/streamer.rs
index c2c10ac87a..138488899b 100644
--- a/src/streamer.rs
+++ b/src/streamer.rs
@@ -429,7 +429,7 @@ mod test {
         let exit = Arc::new(AtomicBool::new(false));
         let subs = Arc::new(RwLock::new(Subscribers::new(
             Node::default(),
-            Node::new([0;8], 0, send.local_addr().unwrap()),
+            Node::new([0; 8], 0, send.local_addr().unwrap()),
         )));
         let resp_recycler = BlobRecycler::default();
         let (s_reader, r_reader) = channel();

From 374c9921fd1d2f015f17a47d99e276c23bc0e2a1 Mon Sep 17 00:00:00 2001
From: Anatoly Yakovenko <aeyakovenko@gmail.com>
Date: Tue, 17 Apr 2018 11:05:15 -0700
Subject: [PATCH 04/10] comments

---
 src/streamer.rs    | 21 +++++++++++++--------
 src/subscribers.rs | 16 +++++++++-------
 2 files changed, 22 insertions(+), 15 deletions(-)

diff --git a/src/streamer.rs b/src/streamer.rs
index 138488899b..9eb2360c0d 100644
--- a/src/streamer.rs
+++ b/src/streamer.rs
@@ -222,6 +222,11 @@ fn retransmit(
     Ok(())
 }
 
+//service to retransmit messages from the leader to layer 1 nodes
+//see subscriber.rs for network layer definitions
+//window receives blobs from the network
+//for any blobs that originated from the leader, we broadcast
+//to the rest of the network
 pub fn retransmitter(
     sock: UdpSocket,
     exit: Arc<AtomicBool>,
@@ -485,18 +490,18 @@ mod test {
         )));
         let n3 = Node::new([0; 8], 1, read.local_addr().unwrap());
         subs.write().unwrap().insert(&[n3]);
-        let (s_cast, r_cast) = channel();
-        let re = BlobRecycler::default();
+        let (s_retransmit, r_retransmit) = channel();
+        let blob_recycler = BlobRecycler::default();
         let saddr = send.local_addr().unwrap();
-        let t_retransmit = retransmitter(send, exit.clone(), subs, re.clone(), r_cast);
+        let t_retransmit = retransmitter(send, exit.clone(), subs, blob_recycler.clone(), r_retransmit);
         let mut bq = VecDeque::new();
-        let b = re.allocate();
+        let b = blob_recycler.allocate();
         b.write().unwrap().meta.size = 10;
         bq.push_back(b);
-        s_cast.send(bq).unwrap();
-        let (s_recv, r_recv) = channel();
-        let t_receiver = blob_receiver(exit.clone(), re.clone(), read, s_recv).unwrap();
-        let mut oq = r_recv.recv().unwrap();
+        s_retransmit.send(bq).unwrap();
+        let (s_blob_receiver, r_blob_receiver) = channel();
+        let t_receiver = blob_receiver(exit.clone(), blob_recycler.clone(), read, s_blob_receiver).unwrap();
+        let mut oq = r_blob_receiver.recv().unwrap();
         assert_eq!(oq.len(), 1);
         let o = oq.pop_front().unwrap();
         let ro = o.read().unwrap();
diff --git a/src/subscribers.rs b/src/subscribers.rs
index 0a123540b3..f719d2e2da 100644
--- a/src/subscribers.rs
+++ b/src/subscribers.rs
@@ -13,11 +13,7 @@ pub struct Node {
 //sockaddr doesn't implement default
 impl Default for Node {
     fn default() -> Node {
-        Node {
-            id: [0; 8],
-            weight: 0,
-            addr: "0.0.0.0:0".parse().unwrap(),
-        }
+        Node { id: [0; 8], weight: 0, addr: "0.0.0.0:0".parse().unwrap(), }
     }
 }
 
@@ -34,9 +30,13 @@ impl Node {
     }
 }
 
+//Subscriber data structure
+//layer 0, leader
+//layer 1, as many nodes as we can fit to quickly get reliable 2/3+1 finality
+//layer 2, everyone else, if layer 1 is 2**10, layer 2 should be 2**20 number of nodes
 pub struct Subscribers {
-    pub data: Vec<Node>,
-    pub me: Node,
+    data: Vec<Node>,
+    me: Node,
     pub leader: Node,
 }
 
@@ -50,6 +50,8 @@ impl Subscribers {
         h.insert(&[me, leader]);
         h
     }
+
+    //retransmit messages from the leader to layer 1 nodes
     pub fn retransmit(&self, blob: &mut Blob, s: &UdpSocket) -> Result<()> {
         let errs: Vec<_> = self.data
             .par_iter()

From 7c96dea35925b09aed1f35aaf705cb82db95e5a4 Mon Sep 17 00:00:00 2001
From: Anatoly Yakovenko <aeyakovenko@gmail.com>
Date: Tue, 17 Apr 2018 11:05:35 -0700
Subject: [PATCH 05/10] fmt

---
 src/streamer.rs    | 11 +++++++++--
 src/subscribers.rs |  6 +++++-
 2 files changed, 14 insertions(+), 3 deletions(-)

diff --git a/src/streamer.rs b/src/streamer.rs
index 9eb2360c0d..5ad180b845 100644
--- a/src/streamer.rs
+++ b/src/streamer.rs
@@ -493,14 +493,21 @@ mod test {
         let (s_retransmit, r_retransmit) = channel();
         let blob_recycler = BlobRecycler::default();
         let saddr = send.local_addr().unwrap();
-        let t_retransmit = retransmitter(send, exit.clone(), subs, blob_recycler.clone(), r_retransmit);
+        let t_retransmit = retransmitter(
+            send,
+            exit.clone(),
+            subs,
+            blob_recycler.clone(),
+            r_retransmit,
+        );
         let mut bq = VecDeque::new();
         let b = blob_recycler.allocate();
         b.write().unwrap().meta.size = 10;
         bq.push_back(b);
         s_retransmit.send(bq).unwrap();
         let (s_blob_receiver, r_blob_receiver) = channel();
-        let t_receiver = blob_receiver(exit.clone(), blob_recycler.clone(), read, s_blob_receiver).unwrap();
+        let t_receiver =
+            blob_receiver(exit.clone(), blob_recycler.clone(), read, s_blob_receiver).unwrap();
         let mut oq = r_blob_receiver.recv().unwrap();
         assert_eq!(oq.len(), 1);
         let o = oq.pop_front().unwrap();
diff --git a/src/subscribers.rs b/src/subscribers.rs
index f719d2e2da..d840638d91 100644
--- a/src/subscribers.rs
+++ b/src/subscribers.rs
@@ -13,7 +13,11 @@ pub struct Node {
 //sockaddr doesn't implement default
 impl Default for Node {
     fn default() -> Node {
-        Node { id: [0; 8], weight: 0, addr: "0.0.0.0:0".parse().unwrap(), }
+        Node {
+            id: [0; 8],
+            weight: 0,
+            addr: "0.0.0.0:0".parse().unwrap(),
+        }
     }
 }
 

From 249cead13e37926c4b4a1127f72e1e21fa74fe41 Mon Sep 17 00:00:00 2001
From: Anatoly Yakovenko <aeyakovenko@gmail.com>
Date: Tue, 17 Apr 2018 11:07:43 -0700
Subject: [PATCH 06/10] docs

---
 src/streamer.rs | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/src/streamer.rs b/src/streamer.rs
index 5ad180b845..53055c1add 100644
--- a/src/streamer.rs
+++ b/src/streamer.rs
@@ -123,6 +123,8 @@ fn recv_window(
         let rsubs = subs.read().unwrap();
         for b in &dq {
             let p = b.read().unwrap();
+            //TODO this check isn't safe against adverserial packets
+            //we need to maintain a sequence window
             if p.meta.addr() == rsubs.leader.addr {
                 //TODO
                 //need to copy the retransmited blob

From 0b3361597971a88d3787cd224e91b96de4837329 Mon Sep 17 00:00:00 2001
From: Anatoly Yakovenko <aeyakovenko@gmail.com>
Date: Tue, 17 Apr 2018 12:48:06 -0700
Subject: [PATCH 07/10] udpate

---
 src/subscribers.rs | 10 ++--------
 1 file changed, 2 insertions(+), 8 deletions(-)

diff --git a/src/subscribers.rs b/src/subscribers.rs
index d840638d91..67273e0d02 100644
--- a/src/subscribers.rs
+++ b/src/subscribers.rs
@@ -23,11 +23,7 @@ impl Default for Node {
 
 impl Node {
     pub fn new(id: [u64; 8], weight: u64, addr: SocketAddr) -> Node {
-        Node {
-            id: id,
-            weight: weight,
-            addr: addr,
-        }
+        Node { id, weight, addr }
     }
     fn key(&self) -> i64 {
         (self.weight as i64).checked_neg().unwrap()
@@ -80,9 +76,7 @@ impl Subscribers {
         Ok(())
     }
     pub fn insert(&mut self, ns: &[Node]) {
-        for n in ns.into_iter() {
-            self.data.push(n.clone());
-        }
+        self.data.extend_from_slice(ns);
         self.data.sort_by_key(Node::key);
     }
 }

From 97e772e87a88bea5cb8757f86b03c7bcb45dc3e7 Mon Sep 17 00:00:00 2001
From: Anatoly Yakovenko <aeyakovenko@gmail.com>
Date: Tue, 17 Apr 2018 19:46:50 -0700
Subject: [PATCH 08/10] docs

---
 src/packet.rs      |  1 +
 src/streamer.rs    | 15 ++++++++++-----
 src/subscribers.rs | 12 +++++++-----
 3 files changed, 18 insertions(+), 10 deletions(-)

diff --git a/src/packet.rs b/src/packet.rs
index 72e5cf8a46..2106bf5dab 100644
--- a/src/packet.rs
+++ b/src/packet.rs
@@ -1,3 +1,4 @@
+//! The `packet` module defines data structures and methods to pull data from the network.
 use byteorder::{LittleEndian, ReadBytesExt, WriteBytesExt};
 use result::{Error, Result};
 use std::collections::VecDeque;
diff --git a/src/streamer.rs b/src/streamer.rs
index 53055c1add..368365ec7f 100644
--- a/src/streamer.rs
+++ b/src/streamer.rs
@@ -1,3 +1,4 @@
+//! The `streamer` module defines a set of services for effecently pulling data from udp sockets.
 use packet::{Blob, BlobRecycler, PacketRecycler, SharedBlob, SharedPackets, NUM_BLOBS};
 use result::Result;
 use std::collections::VecDeque;
@@ -224,11 +225,15 @@ fn retransmit(
     Ok(())
 }
 
-//service to retransmit messages from the leader to layer 1 nodes
-//see subscriber.rs for network layer definitions
-//window receives blobs from the network
-//for any blobs that originated from the leader, we broadcast
-//to the rest of the network
+/// Service to retransmit messages from the leader to layer 1 nodes.
+/// See `subscribers` for network layer definitions.
+/// # Arguments
+/// * `sock` - Socket to read from.  Read timeout is set to 1.
+/// * `exit` - Boolean to signal system exit.
+/// * `subs` - Shared Subscriber structure.  This structure needs to be updated and popualted by
+/// the accountant.
+/// * `recycler` - Blob recycler.
+/// * `r` - Receive channel for blobs to be retransmitted to all the layer 1 nodes.
 pub fn retransmitter(
     sock: UdpSocket,
     exit: Arc<AtomicBool>,
diff --git a/src/subscribers.rs b/src/subscribers.rs
index 67273e0d02..a0c2cfdd7c 100644
--- a/src/subscribers.rs
+++ b/src/subscribers.rs
@@ -1,3 +1,9 @@
+//! The `subscribers` module defines data structures to keep track of nodes on the network.
+//! The network is arranged in layers:
+//! * layer 0, leader
+//! * layer 1, as many nodes as we can fit to quickly get reliable 2/3+1 finality
+//! * layer 2, everyone else, if layer 1 is 2**10, layer 2 should be 2**20 number of nodes
+//! It's up to the external state machine to keep this updated.
 use packet::Blob;
 use rayon::prelude::*;
 use result::{Error, Result};
@@ -30,10 +36,6 @@ impl Node {
     }
 }
 
-//Subscriber data structure
-//layer 0, leader
-//layer 1, as many nodes as we can fit to quickly get reliable 2/3+1 finality
-//layer 2, everyone else, if layer 1 is 2**10, layer 2 should be 2**20 number of nodes
 pub struct Subscribers {
     data: Vec<Node>,
     me: Node,
@@ -51,7 +53,7 @@ impl Subscribers {
         h
     }
 
-    //retransmit messages from the leader to layer 1 nodes
+    /// retransmit messages from the leader to layer 1 nodes
     pub fn retransmit(&self, blob: &mut Blob, s: &UdpSocket) -> Result<()> {
         let errs: Vec<_> = self.data
             .par_iter()

From d003efb522a5366c5a7c3aa2a58872d37d183ef3 Mon Sep 17 00:00:00 2001
From: Anatoly Yakovenko <aeyakovenko@gmail.com>
Date: Tue, 17 Apr 2018 19:52:46 -0700
Subject: [PATCH 09/10] fix docs

---
 src/subscribers.rs | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/src/subscribers.rs b/src/subscribers.rs
index a0c2cfdd7c..8a44341670 100644
--- a/src/subscribers.rs
+++ b/src/subscribers.rs
@@ -1,8 +1,10 @@
 //! The `subscribers` module defines data structures to keep track of nodes on the network.
 //! The network is arranged in layers:
-//! * layer 0, leader
-//! * layer 1, as many nodes as we can fit to quickly get reliable 2/3+1 finality
-//! * layer 2, everyone else, if layer 1 is 2**10, layer 2 should be 2**20 number of nodes
+//!
+//! * layer 0 - leader
+//! * layer 1 - as many nodes as we can fit to quickly get reliable `2/3+1` finality
+//! * layer 2 - everyone else, if layer 1 is `2^10`, layer 2 should be `2^20` number of nodes
+//!
 //! It's up to the external state machine to keep this updated.
 use packet::Blob;
 use rayon::prelude::*;

From 29f3230089ef497040973fc618b377687a761832 Mon Sep 17 00:00:00 2001
From: Anatoly Yakovenko <aeyakovenko@gmail.com>
Date: Tue, 17 Apr 2018 19:53:18 -0700
Subject: [PATCH 10/10] docs

---
 src/subscribers.rs | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/src/subscribers.rs b/src/subscribers.rs
index 8a44341670..153246d12a 100644
--- a/src/subscribers.rs
+++ b/src/subscribers.rs
@@ -1,9 +1,9 @@
 //! The `subscribers` module defines data structures to keep track of nodes on the network.
 //! The network is arranged in layers:
 //!
-//! * layer 0 - leader
-//! * layer 1 - as many nodes as we can fit to quickly get reliable `2/3+1` finality
-//! * layer 2 - everyone else, if layer 1 is `2^10`, layer 2 should be `2^20` number of nodes
+//! * layer 0 - Leader.
+//! * layer 1 - As many nodes as we can fit to quickly get reliable `2/3+1` finality
+//! * layer 2 - Everyone else, if layer 1 is `2^10`, layer 2 should be able to fit `2^20` number of nodes.
 //!
 //! It's up to the external state machine to keep this updated.
 use packet::Blob;