Tvu cleanup2

src/bin/testnode.rs

@@ -131,6 +131,7 @@ fn main() {
         gossip_sock.local_addr().unwrap(),
         replicate_sock.local_addr().unwrap(),
         serve_sock.local_addr().unwrap(),
+        events_sock.local_addr().unwrap(),
     );
 
     let mut local = serve_sock.local_addr().unwrap();
@@ -139,7 +140,7 @@ fn main() {
     let respond_socket = UdpSocket::bind(local.clone()).unwrap();
 
     eprintln!("starting server...");
-    let server = Server::new(
+    let server = Server::leader(
         bank,
         last_id,
         Some(Duration::from_millis(1000)),

src/crdt.rs

@@ -31,7 +31,7 @@ use std::thread::{sleep, spawn, JoinHandle};
 use std::time::Duration;
 
 /// Structure to be replicated by the network
-#[derive(Serialize, Deserialize, Clone)]
+#[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct ReplicatedData {
     pub id: PublicKey,
     sig: Signature,
@@ -42,7 +42,9 @@ pub struct ReplicatedData {
     /// address to connect to for replication
     pub replicate_addr: SocketAddr,
     /// address to connect to when this node is leader
-    pub serve_addr: SocketAddr,
+    pub requests_addr: SocketAddr,
+    /// events address
+    pub events_addr: SocketAddr,
     /// current leader identity
     current_leader_id: PublicKey,
     /// last verified hash that was submitted to the leader
@@ -56,15 +58,17 @@ impl ReplicatedData {
         id: PublicKey,
         gossip_addr: SocketAddr,
         replicate_addr: SocketAddr,
-        serve_addr: SocketAddr,
+        requests_addr: SocketAddr,
+        events_addr: SocketAddr,
     ) -> ReplicatedData {
         ReplicatedData {
             id,
             sig: Signature::default(),
             version: 0,
             gossip_addr,
             replicate_addr,
-            serve_addr,
+            requests_addr,
+            events_addr,
             current_leader_id: PublicKey::default(),
             last_verified_hash: Hash::default(),
             last_verified_count: 0,
@@ -515,12 +519,14 @@ mod test {
         let gossip = UdpSocket::bind("0.0.0.0:0").unwrap();
         let replicate = UdpSocket::bind("0.0.0.0:0").unwrap();
         let serve = UdpSocket::bind("0.0.0.0:0").unwrap();
+        let events = UdpSocket::bind("0.0.0.0:0").unwrap();
         let pubkey = KeyPair::new().pubkey();
         let d = ReplicatedData::new(
             pubkey,
             gossip.local_addr().unwrap(),
             replicate.local_addr().unwrap(),
             serve.local_addr().unwrap(),
+            events.local_addr().unwrap(),
         );
         let crdt = Crdt::new(d);
         trace!(
@@ -632,6 +638,7 @@ mod test {
             "127.0.0.1:1234".parse().unwrap(),
             "127.0.0.1:1235".parse().unwrap(),
             "127.0.0.1:1236".parse().unwrap(),
+            "127.0.0.1:1237".parse().unwrap(),
         );
         assert_eq!(d.version, 0);
         let mut crdt = Crdt::new(d.clone());

src/server.rs

@@ -11,13 +11,14 @@ use std::sync::atomic::AtomicBool;
 use std::thread::JoinHandle;
 use std::time::Duration;
 use tpu::Tpu;
+use tvu::Tvu;
 
 pub struct Server {
     pub thread_hdls: Vec<JoinHandle<()>>,
 }
 
 impl Server {
-    pub fn new<W: Write + Send + 'static>(
+    pub fn leader<W: Write + Send + 'static>(
         bank: Bank,
         start_hash: Hash,
         tick_duration: Option<Duration>,
@@ -26,28 +27,51 @@ impl Server {
         events_socket: UdpSocket,
         broadcast_socket: UdpSocket,
         respond_socket: UdpSocket,
-        gossip: UdpSocket,
+        gossip_socket: UdpSocket,
         exit: Arc<AtomicBool>,
         writer: W,
     ) -> Self {
         let bank = Arc::new(bank);
         let mut thread_hdls = vec![];
         let rpu = Rpu::new(bank.clone(), requests_socket, respond_socket, exit.clone());
         thread_hdls.extend(rpu.thread_hdls);
-
         let tpu = Tpu::new(
             bank.clone(),
             start_hash,
             tick_duration,
             me,
             events_socket,
             broadcast_socket,
-            gossip,
+            gossip_socket,
             exit.clone(),
             writer,
         );
         thread_hdls.extend(tpu.thread_hdls);
-
+        Server { thread_hdls }
+    }
+    pub fn validator(
+        bank: Bank,
+        me: ReplicatedData,
+        requests_socket: UdpSocket,
+        respond_socket: UdpSocket,
+        replicate_socket: UdpSocket,
+        gossip_socket: UdpSocket,
+        leader_repl_data: ReplicatedData,
+        exit: Arc<AtomicBool>,
+    ) -> Self {
+        let bank = Arc::new(bank);
+        let mut thread_hdls = vec![];
+        let rpu = Rpu::new(bank.clone(), requests_socket, respond_socket, exit.clone());
+        thread_hdls.extend(rpu.thread_hdls);
+        let tvu = Tvu::new(
+            bank.clone(),
+            me,
+            gossip_socket,
+            replicate_socket,
+            leader_repl_data,
+            exit.clone(),
+        );
+        thread_hdls.extend(tvu.thread_hdls);
         Server { thread_hdls }
     }
 }

src/streamer.rs

@@ -650,12 +650,14 @@ mod test {
         let addr = read.local_addr().unwrap();
         let send = UdpSocket::bind("127.0.0.1:0").expect("bind");
         let serve = UdpSocket::bind("127.0.0.1:0").expect("bind");
+        let event = UdpSocket::bind("127.0.0.1:0").expect("bind");
         let exit = Arc::new(AtomicBool::new(false));
         let rep_data = ReplicatedData::new(
             pubkey_me,
             read.local_addr().unwrap(),
             send.local_addr().unwrap(),
             serve.local_addr().unwrap(),
+            event.local_addr().unwrap(),
         );
         let mut crdt_me = Crdt::new(rep_data);
         let me_id = crdt_me.my_data().id;
@@ -712,21 +714,17 @@ mod test {
         let gossip = UdpSocket::bind("127.0.0.1:0").unwrap();
         let replicate = UdpSocket::bind("127.0.0.1:0").unwrap();
         let serve = UdpSocket::bind("127.0.0.1:0").unwrap();
+        let event = UdpSocket::bind("127.0.0.1:0").unwrap();
         let pubkey = KeyPair::new().pubkey();
         let d = ReplicatedData::new(
             pubkey,
             gossip.local_addr().unwrap(),
             replicate.local_addr().unwrap(),
             serve.local_addr().unwrap(),
+            event.local_addr().unwrap(),
         );
+        trace!("data: {:?}", d);
         let crdt = Crdt::new(d);
-        trace!(
-            "id: {} gossip: {} replicate: {} serve: {}",
-            crdt.my_data().id[0],
-            gossip.local_addr().unwrap(),
-            replicate.local_addr().unwrap(),
-            serve.local_addr().unwrap(),
-        );
         (Arc::new(RwLock::new(crdt)), gossip, replicate, serve)
     }