From 0cc786a1e5277f96cbd7fe64b55a6f633045c32f Mon Sep 17 00:00:00 2001 From: Ignotus Peverell Date: Fri, 23 Dec 2016 12:15:36 -0800 Subject: [PATCH] Refactored server to share the same struct whether futures are exposed or not. Multi-server test now properly waits for a block instead of a timer. --- grin/src/lib.rs | 2 +- grin/src/server.rs | 48 +++++++++--------------------------------- grin/tests/simulnet.rs | 44 ++++++++++++++++++++++++++++++++++---- 3 files changed, 51 insertions(+), 43 deletions(-) diff --git a/grin/src/lib.rs b/grin/src/lib.rs index 6cf508e9..de97889d 100644 --- a/grin/src/lib.rs +++ b/grin/src/lib.rs @@ -40,4 +40,4 @@ mod adapters; mod miner; mod server; -pub use server::{Server, ServerFut, ServerConfig}; +pub use server::{Server, ServerConfig}; diff --git a/grin/src/server.rs b/grin/src/server.rs index b26483a9..5602ce13 100644 --- a/grin/src/server.rs +++ b/grin/src/server.rs @@ -108,43 +108,8 @@ impl Server { }) } - /// Asks the server to connect to a peer at the provided network address. - pub fn connect_peer(&self, addr: SocketAddr) -> Result<(), Error> { - let handle = self.evt_handle.clone(); - handle.spawn(self.p2p.connect_peer(addr, handle.clone()).map_err(|_| ())); - Ok(()) - } - - /// Start mining for blocks on a separate thread. Relies on a toy miner, - /// mostly for testing. - pub fn start_miner(&self) { - let miner = miner::Miner::new(self.chain_head.clone(), - self.chain_store.clone(), - self.chain_adapter.clone()); - thread::spawn(move || { - miner.run_loop(); - }); - } -} - -/// Implementation of the server that doesn't take control of the event loop -/// and returns futures instead. -pub struct ServerFut { - config: ServerConfig, - /// handle to our network server - p2p: Arc, - /// the reference copy of the current chain state - chain_head: Arc>, - /// data store access - chain_store: Arc, - /// chain adapter to net, required for miner and anything that submits - /// blocks - chain_adapter: Arc, -} - -impl ServerFut { - /// Instantiates and starts a new server. - pub fn start(config: ServerConfig, evt_handle: &reactor::Handle) -> Result { + /// Instantiates a new server associated with the provided future reactor. + pub fn future(config: ServerConfig, evt_handle: &reactor::Handle) -> Result { let (chain_store, head) = try!(store_head(&config)); let shared_head = Arc::new(Mutex::new(head)); @@ -169,7 +134,8 @@ impl ServerFut { } /// Asks the server to connect to a peer at the provided network address. - pub fn connect_peer(&self, addr: SocketAddr, handle: &reactor::Handle) -> Result<(), Error> { + pub fn connect_peer(&self, addr: SocketAddr) -> Result<(), Error> { + let handle = self.evt_handle.clone(); handle.spawn(self.p2p.connect_peer(addr, handle.clone()).map_err(|_| ())); Ok(()) } @@ -184,6 +150,12 @@ impl ServerFut { miner.run_loop(); }); } + + pub fn head(&self) -> chain::Tip { + let head = self.chain_head.clone(); + let h = head.lock().unwrap(); + h.clone() + } } // Helper function to create the chain storage and check if it already has a diff --git a/grin/tests/simulnet.rs b/grin/tests/simulnet.rs index bfcb765b..5d20837b 100644 --- a/grin/tests/simulnet.rs +++ b/grin/tests/simulnet.rs @@ -25,7 +25,8 @@ use std::io; use std::thread; use std::time; -use futures::Future; +use futures::{Future, Poll, Async}; +use futures::task::park; use tokio_core::reactor; #[test] @@ -38,7 +39,7 @@ fn simulate_servers() { // instantiates 5 servers on different ports let mut servers = vec![]; for n in 0..5 { - let s = grin::ServerFut::start( + let s = grin::Server::future( grin::ServerConfig{ db_root: format!("target/grin-{}", n), cuckoo_size: 12, @@ -58,7 +59,42 @@ fn simulate_servers() { // start mining servers[0].start_miner(); + let original_height = servers[0].head().height; - let timeout = reactor::Timeout::new(time::Duration::new(10, 0), &handle.clone()).unwrap(); - evtlp.run(timeout); + // monitor for a change of head on a different server and check we + evtlp.run(change(&servers[4]).and_then(|tip| { + assert!(tip.height == original_height+1); + Ok(()) + })); +} + +// Builds the change future, monitoring for a change of head on the provided server +fn change<'a>(s: &'a grin::Server) -> HeadChange<'a> { + let start_head = s.head(); + HeadChange { + server: s, + original: start_head, + } +} + +/// Future that monitors when a server has had its head updated. +struct HeadChange<'a> { + server: &'a grin::Server, + original: chain::Tip, +} + +impl<'a> Future for HeadChange<'a> { + type Item = chain::Tip; + type Error = (); + + fn poll(&mut self) -> Poll { + let new_head = self.server.head(); + if new_head.last_block_h != self.original.last_block_h { + Ok(Async::Ready(new_head)) + } else { + // egregious polling, asking the task to schedule us every iteration + park().unpark(); + Ok(Async::NotReady) + } + } }