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

@@ -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;

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()
+            if exit.load(Ordering::Relaxed) {
-                || 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");
+    }
+
 }

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());
+    }
+}