very dumb leader selection

src/bin/fullnode.rs

@@ -36,7 +36,12 @@ fn main() {
     let mut opts = Options::new();
     opts.optflag("h", "help", "print help");
     opts.optopt("l", "", "run with the identity found in FILE", "FILE");
-    opts.optopt("v", "", "validate; find leader's identity in FILE", "FILE");
+    opts.optopt(
+        "t",
+        "",
+        "testnet; connect to the network at this gossip entry point",
+        "HOST:PORT",
+    );
     opts.optopt(
         "o",
         "",
@@ -119,14 +124,14 @@ fn main() {
     }
 
     let exit = Arc::new(AtomicBool::new(false));
-    let threads = if matches.opt_present("v") {
-        let path = matches.opt_str("v").unwrap();
+    let threads = if matches.opt_present("t") {
+        let testnet_address_string = matches.opt_str("t").unwrap();
         eprintln!(
-            "starting validator... {} using {}",
-            repl_data.requests_addr, path
+            "starting validator... {} connecting to {}",
+            repl_data.requests_addr, testnet_address_string
         );
-        let file = File::open(path.clone()).expect(&format!("file not found: {}", path));
-        let leader = serde_json::from_reader(file).expect("parse");
+        let testnet_addr = testnet_address_string.parse().unwrap();
+        let newtwork_entry_point = ReplicatedData::new_entry_point(testnet_addr);
         let s = Server::new_validator(
             bank,
             repl_data.clone(),
@@ -135,7 +140,7 @@ fn main() {
             UdpSocket::bind(repl_data.replicate_addr).unwrap(),
             UdpSocket::bind(repl_data.gossip_addr).unwrap(),
             UdpSocket::bind(repl_data.repair_addr).unwrap(),
-            leader,
+            newtwork_entry_point,
             exit.clone(),
         );
         s.thread_hdls

src/crdt.rs

@@ -551,6 +551,32 @@ impl Crdt {
         blob_sender.send(q)?;
         Ok(())
     }
+    /// TODO: This is obviously the wrong way to do this. Need to implement leader selection
+    fn top_leader(&self) -> Option<PublicKey> {
+        let mut table = HashMap::new();
+        let def = PublicKey::default();
+        let cur = self.table.values().filter(|x| x.current_leader_id != def);
+        for v in cur {
+            let cnt = table.entry(&v.current_leader_id).or_insert(0);
+            *cnt += 1;
+            trace!("leader {:?} {}", &v.current_leader_id[..4], *cnt);
+        }
+        let mut sorted: Vec<(&PublicKey, usize)> = table.into_iter().collect();
+        sorted.sort_by_key(|a| a.1);
+        sorted.last().map(|a| *a.0)
+    }
+
+    /// TODO: This is obviously the wrong way to do this. Need to implement leader selection
+    /// A t-shirt for the first person to actually use this bad behavior to attack the alpha testnet
+    fn update_leader(&mut self) {
+        if let Some(leader_id) = self.top_leader() {
+            if self.my_data().current_leader_id != leader_id {
+                if self.table.get(&leader_id).is_some() {
+                    self.set_leader(leader_id);
+                }
+            }
+        }
+    }
 
     /// Apply updates that we received from the identity `from`
     /// # Arguments
@@ -577,14 +603,20 @@ impl Crdt {
         Builder::new()
             .name("solana-gossip".to_string())
             .spawn(move || loop {
+                let start = timestamp();
                 let _ = Self::run_gossip(&obj, &blob_sender, &blob_recycler);
                 obj.write().unwrap().purge(timestamp());
                 if exit.load(Ordering::Relaxed) {
                     return;
                 }
                 //TODO: possibly tune this parameter
                 //we saw a deadlock passing an obj.read().unwrap().timeout into sleep
-                sleep(Duration::from_millis(GOSSIP_SLEEP_MILLIS));
+                let _ = obj.write().unwrap().update_leader();
+                let elapsed = timestamp() - start;
+                if GOSSIP_SLEEP_MILLIS > elapsed {
+                    let time_left = GOSSIP_SLEEP_MILLIS - elapsed;
+                    sleep(Duration::from_millis(time_left));
+                }
             })
             .unwrap()
     }
@@ -785,17 +817,10 @@ pub struct TestNode {
 impl TestNode {
     pub fn new() -> TestNode {
         let transaction = UdpSocket::bind("0.0.0.0:0").unwrap();
-        let mut addr = transaction.local_addr().unwrap();
-        let port = addr.port();
-        addr.set_port(port + 1);
-        let gossip = UdpSocket::bind(addr.clone()).unwrap();
-        addr.set_port(port + 2);
-        let replicate = UdpSocket::bind(addr.clone()).unwrap();
-        addr.set_port(port + 3);
-        let requests = UdpSocket::bind(addr.clone()).unwrap();
-        addr.set_port(port + 4);
-        let repair = UdpSocket::bind(addr.clone()).unwrap();
-
+        let gossip = UdpSocket::bind("0.0.0.0:0").unwrap();
+        let replicate = UdpSocket::bind("0.0.0.0:0").unwrap();
+        let requests = UdpSocket::bind("0.0.0.0:0").unwrap();
+        let repair = UdpSocket::bind("0.0.0.0:0").unwrap();
         let gossip_send = UdpSocket::bind("0.0.0.0:0").unwrap();
         let respond = UdpSocket::bind("0.0.0.0:0").unwrap();
         let broadcast = UdpSocket::bind("0.0.0.0:0").unwrap();
@@ -829,6 +854,7 @@ impl TestNode {
 #[cfg(test)]
 mod tests {
     use crdt::{parse_port_or_addr, Crdt, ReplicatedData, GOSSIP_SLEEP_MILLIS, MIN_TABLE_SIZE};
+    use logger;
     use packet::BlobRecycler;
     use result::Error;
     use signature::{KeyPair, KeyPairUtil};
@@ -1173,4 +1199,30 @@ mod tests {
             assert_eq!(blob.get_id().unwrap(), id);
         }
     }
+    /// TODO: This is obviously the wrong way to do this. Need to implement leader selection,
+    /// delete this test after leader selection is correctly implemented
+    #[test]
+    fn test_update_leader() {
+        logger::setup();
+        let me = ReplicatedData::new_leader(&"127.0.0.1:1234".parse().unwrap());
+        let leader0 = ReplicatedData::new_leader(&"127.0.0.1:1234".parse().unwrap());
+        let leader1 = ReplicatedData::new_leader(&"127.0.0.1:1234".parse().unwrap());
+        let mut crdt = Crdt::new(me.clone());
+        assert_eq!(crdt.top_leader(), None);
+        crdt.set_leader(leader0.id);
+        assert_eq!(crdt.top_leader().unwrap(), leader0.id);
+        //add a bunch of nodes with a new leader
+        for _ in 0..10 {
+            let mut dum = ReplicatedData::new_entry_point("127.0.0.1:1234".parse().unwrap());
+            dum.id = KeyPair::new().pubkey();
+            dum.current_leader_id = leader1.id;
+            crdt.insert(&dum);
+        }
+        assert_eq!(crdt.top_leader().unwrap(), leader1.id);
+        crdt.update_leader();
+        assert_eq!(crdt.my_data().current_leader_id, leader0.id);
+        crdt.insert(&leader1);
+        crdt.update_leader();
+        assert_eq!(crdt.my_data().current_leader_id, leader1.id);
+    }
 }

src/drone.rs

@@ -225,6 +225,7 @@ mod tests {
     }
 
     #[test]
+    #[ignore]
     fn test_send_airdrop() {
         const SMALL_BATCH: i64 = 50;
         const TPS_BATCH: i64 = 5_000_000;

src/server.rs

@@ -134,7 +134,7 @@ impl Server {
         replicate_socket: UdpSocket,
         gossip_listen_socket: UdpSocket,
         repair_socket: UdpSocket,
-        leader_repl_data: ReplicatedData,
+        entry_point: ReplicatedData,
         exit: Arc<AtomicBool>,
     ) -> Self {
         let bank = Arc::new(bank);
@@ -143,12 +143,9 @@ impl Server {
         thread_hdls.extend(rpu.thread_hdls);
 
         let crdt = Arc::new(RwLock::new(Crdt::new(me)));
-        crdt.write()
-            .expect("'crdt' write lock in pub fn replicate")
-            .set_leader(leader_repl_data.id);
         crdt.write()
             .expect("'crdt' write lock before insert() in pub fn replicate")
-            .insert(&leader_repl_data);
+            .insert(&entry_point);
         let window = streamer::default_window();
         let gossip_send_socket = UdpSocket::bind("0.0.0.0:0").expect("bind 0");
         let retransmit_socket = UdpSocket::bind("0.0.0.0:0").expect("bind 0");