grassrootseconomics/openethereum
8
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Network shutdown

Browse Source
This commit is contained in:
arkpar
2016-06-17 12:58:28 +02:00
parent a03da30510
commit b38488dd07
12 changed files with 205 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
util/src/network/host.rs
View File

@@ -49,7 +49,7 @@ const MAX_HANDSHAKES: usize = 80;
const MAX_HANDSHAKES_PER_ROUND: usize = 32;
const MAINTENANCE_TIMEOUT: u64 = 1000;
#[derive(Debug)]
#[derive(Debug, Clone)]
/// Network service configuration
pub struct NetworkConfiguration {
/// Directory path to store network configuration. None means nothing will be saved
@@ -233,6 +233,11 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
self.io.message(NetworkIoMessage::User(msg));
}
/// Send an IO message
pub fn io_channel(&self) -> IoChannel<NetworkIoMessage<Message>> {
self.io.channel()
}
/// Disable current protocol capability for given peer. If no capabilities left peer gets disconnected.
pub fn disable_peer(&self, peer: PeerId) {
//TODO: remove capability, disconnect if no capabilities left
@@ -327,7 +332,7 @@ pub struct Host<Message> where Message: Send + Sync + Clone {
impl<Message> Host<Message> where Message: Send + Sync + Clone {
/// Create a new instance
pub fn new(config: NetworkConfiguration) -> Result<Host<Message>, UtilError> {
pub fn new(config: NetworkConfiguration, stats: Arc<NetworkStats>) -> Result<Host<Message>, UtilError> {
let mut listen_address = match config.listen_address {
None => SocketAddr::from_str("0.0.0.0:30304").unwrap(),
Some(addr) => addr,
@@ -371,7 +376,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
handlers: RwLock::new(HashMap::new()),
timers: RwLock::new(HashMap::new()),
timer_counter: RwLock::new(USER_TIMER),
stats: Arc::new(NetworkStats::default()),
stats: stats,
pinned_nodes: Vec::new(),
num_sessions: AtomicUsize::new(0),
};
@@ -383,10 +388,6 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
Ok(host)
}
pub fn stats(&self) -> Arc<NetworkStats> {
self.stats.clone()
}
pub fn add_node(&mut self, id: &str) {
match Node::from_str(id) {
Err(e) => { debug!(target: "network", "Could not add node {}: {:?}", id, e); },
@@ -401,8 +402,8 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
}
}
pub fn client_version(&self) -> String {
self.info.read().unwrap().client_version.clone()
pub fn client_version() -> String {
version()
}
pub fn external_url(&self) -> Option<String> {
@@ -415,6 +416,23 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
r
}
pub fn stop(&self, io: &IoContext<NetworkIoMessage<Message>>) -> Result<(), UtilError> {
let mut to_kill = Vec::new();
for e in self.sessions.write().unwrap().iter_mut() {
let mut s = e.lock().unwrap();
if !s.keep_alive(io) {
s.disconnect(io, DisconnectReason::PingTimeout);
to_kill.push(s.token());
}
}
for p in to_kill {
trace!(target: "network", "Ping timeout: {}", p);
self.kill_connection(p, io, true);
}
io.unregister_handler();
Ok(())
}
fn init_public_interface(&self, io: &IoContext<NetworkIoMessage<Message>>) -> Result<(), UtilError> {
io.clear_timer(INIT_PUBLIC).unwrap();
if self.info.read().unwrap().public_endpoint.is_some() {
@@ -767,6 +785,13 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
}
}
impl<Message> Drop for Host<Message> where Message: Send + Sync + Clone {
fn drop(&mut self) {
info!("Dropping host");
}
}
impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Message: Send + Sync + Clone + 'static {
/// Initialize networking
fn initialize(&self, io: &IoContext<NetworkIoMessage<Message>>) {
@@ -831,8 +856,8 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
},
_ => match self.timers.read().unwrap().get(&token).cloned() {
Some(timer) => match self.handlers.read().unwrap().get(timer.protocol).cloned() {
None => { warn!(target: "network", "No handler found for protocol: {:?}", timer.protocol) },
Some(h) => { h.timeout(&NetworkContext::new(io, timer.protocol, None, self.sessions.clone()), timer.token); }
None => { warn!(target: "network", "No handler found for protocol: {:?}", timer.protocol) },
Some(h) => { h.timeout(&NetworkContext::new(io, timer.protocol, None, self.sessions.clone()), timer.token); }
},
None => { warn!("Unknown timer token: {}", token); } // timer is not registerd through us
}

55
util/src/network/service.rs
View File

@@ -28,33 +28,33 @@ use io::*;
pub struct NetworkService<Message> where Message: Send + Sync + Clone + 'static {
io_service: IoService<NetworkIoMessage<Message>>,
host_info: String,
host: Arc<Host<Message>>,
host: RwLock<Option<Arc<Host<Message>>>>,
stats: Arc<NetworkStats>,
panic_handler: Arc<PanicHandler>
panic_handler: Arc<PanicHandler>,
config: NetworkConfiguration,
}
impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'static {
/// Starts IO event loop
pub fn start(config: NetworkConfiguration) -> Result<NetworkService<Message>, UtilError> {
pub fn new(config: NetworkConfiguration) -> Result<NetworkService<Message>, UtilError> {
let panic_handler = PanicHandler::new_in_arc();
let mut io_service = try!(IoService::<NetworkIoMessage<Message>>::start());
let io_service = try!(IoService::<NetworkIoMessage<Message>>::start());
panic_handler.forward_from(&io_service);
let host = Arc::new(try!(Host::new(config)));
let stats = host.stats().clone();
let host_info = host.client_version();
try!(io_service.register_handler(host.clone()));
let stats = Arc::new(NetworkStats::new());
let host_info = Host::<Message>::client_version();
Ok(NetworkService {
io_service: io_service,
host_info: host_info,
stats: stats,
panic_handler: panic_handler,
host: host,
host: RwLock::new(None),
config: config,
})
}
/// Regiter a new protocol handler with the event loop.
pub fn register_protocol(&mut self, handler: Arc<NetworkProtocolHandler<Message>+Send + Sync>, protocol: ProtocolId, versions: &[u8]) -> Result<(), NetworkError> {
pub fn register_protocol(&self, handler: Arc<NetworkProtocolHandler<Message>+Send + Sync>, protocol: ProtocolId, versions: &[u8]) -> Result<(), NetworkError> {
try!(self.io_service.send_message(NetworkIoMessage::AddHandler {
handler: handler,
protocol: protocol,
@@ -69,8 +69,8 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
}
/// Returns underlying io service.
pub fn io(&mut self) -> &mut IoService<NetworkIoMessage<Message>> {
&mut self.io_service
pub fn io(&self) -> &IoService<NetworkIoMessage<Message>> {
&self.io_service
}
/// Returns network statistics.
@@ -80,12 +80,37 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
/// Returns external url if available.
pub fn external_url(&self) -> Option<String> {
self.host.external_url()
let host = self.host.read().unwrap();
host.as_ref().and_then(|h| h.external_url())
}
/// Returns external url if available.
pub fn local_url(&self) -> String {
self.host.local_url()
pub fn local_url(&self) -> Option<String> {
let host = self.host.read().unwrap();
host.as_ref().map(|h| h.local_url())
}
/// Start network IO
pub fn start(&self) -> Result<(), UtilError> {
let mut host = self.host.write().unwrap();
if host.is_none() {
let h = Arc::new(try!(Host::new(self.config.clone(), self.stats.clone())));
try!(self.io_service.register_handler(h.clone()));
*host = Some(h);
}
Ok(())
}
/// Stop network IO
pub fn stop(&self) -> Result<(), UtilError> {
let mut host = self.host.write().unwrap();
if let Some(ref host) = *host {
info!("Unregistering handler");
let io = IoContext::new(self.io_service.channel(), 0); //TODO: take token id from host
host.stop(&io);
}
*host = None;
Ok(())
}
}

2
util/src/network/stats.rs
View File

@@ -65,7 +65,7 @@ impl NetworkStats {
self.sessions.load(Ordering::Relaxed)
}
#[cfg(test)]
/// Create a new empty instance.
pub fn new() -> NetworkStats {
NetworkStats {
recv: AtomicUsize::new(0),