Split IO and network crates (#1828)

* Abort on panic

* Split IO and network crates

* Restore panic handler

* Fixed doc tests

util/io/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+description = "Ethcore IO library"
+homepage = "http://ethcore.io"
+license = "GPL-3.0"
+name = "ethcore-io"
+version = "1.3.0"
+authors = ["Ethcore <admin@ethcore.io>"]
+
+[dependencies]
+mio = { git = "https://github.com/ethcore/mio", branch = "v0.5.x" }
+crossbeam = "0.2"
+parking_lot = "0.2.6"
+log = "0.3"
+slab = "0.2"
+

util/io/src/lib.rs

@@ -0,0 +1,169 @@
+// Copyright 2015, 2016 Ethcore (UK) Ltd.
+// This file is part of Parity.
+
+// Parity is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity.  If not, see <http://www.gnu.org/licenses/>.
+
+//! General IO module.
+//!
+//! Example usage for creating a network service and adding an IO handler:
+//!
+//! ```rust
+//! extern crate ethcore_io;
+//! use ethcore_io::*;
+//! use std::sync::Arc;
+//!
+//! struct MyHandler;
+//!
+//! #[derive(Clone)]
+//! struct MyMessage {
+//! 	data: u32
+//! }
+//!
+//! impl IoHandler<MyMessage> for MyHandler {
+//! 	fn initialize(&self, io: &IoContext<MyMessage>) {
+//!			io.register_timer(0, 1000).unwrap();
+//!		}
+//!
+//!		fn timeout(&self, _io: &IoContext<MyMessage>, timer: TimerToken) {
+//!			println!("Timeout {}", timer);
+//!		}
+//!
+//!		fn message(&self, _io: &IoContext<MyMessage>, message: &MyMessage) {
+//!			println!("Message {}", message.data);
+//!		}
+//! }
+//!
+//! fn main () {
+//! 	let mut service = IoService::<MyMessage>::start().expect("Error creating network service");
+//! 	service.register_handler(Arc::new(MyHandler)).unwrap();
+//!
+//! 	// Wait for quit condition
+//! 	// ...
+//! 	// Drop the service
+//! }
+//! ```
+
+extern crate mio;
+#[macro_use]
+extern crate log as rlog;
+extern crate slab;
+extern crate crossbeam;
+extern crate parking_lot;
+
+mod service;
+mod worker;
+mod panics;
+
+use mio::{EventLoop, Token};
+use std::fmt;
+
+#[derive(Debug)]
+/// IO Error
+pub enum IoError {
+	/// Low level error from mio crate
+	Mio(::std::io::Error),
+	/// Error concerning the Rust standard library's IO subsystem.
+	StdIo(::std::io::Error),
+}
+
+impl fmt::Display for IoError {
+	fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+		// just defer to the std implementation for now.
+		// we can refine the formatting when more variants are added.
+		match *self {
+			IoError::Mio(ref std_err) => std_err.fmt(f),
+			IoError::StdIo(ref std_err) => std_err.fmt(f),
+		}
+	}
+}
+
+impl From<::std::io::Error> for IoError {
+	fn from(err: ::std::io::Error) -> IoError {
+		IoError::StdIo(err)
+	}
+}
+
+impl<Message> From<::mio::NotifyError<service::IoMessage<Message>>> for IoError where Message: Send + Clone {
+	fn from(_err: ::mio::NotifyError<service::IoMessage<Message>>) -> IoError {
+		IoError::Mio(::std::io::Error::new(::std::io::ErrorKind::ConnectionAborted, "Network IO notification error"))
+	}
+}
+
+/// Generic IO handler.
+/// All the handler function are called from within IO event loop.
+/// `Message` type is used as notification data
+pub trait IoHandler<Message>: Send + Sync where Message: Send + Sync + Clone + 'static {
+	/// Initialize the handler
+	fn initialize(&self, _io: &IoContext<Message>) {}
+	/// Timer function called after a timeout created with `HandlerIo::timeout`.
+	fn timeout(&self, _io: &IoContext<Message>, _timer: TimerToken) {}
+	/// Called when a broadcasted message is received. The message can only be sent from a different IO handler.
+	fn message(&self, _io: &IoContext<Message>, _message: &Message) {}
+	/// Called when an IO stream gets closed
+	fn stream_hup(&self, _io: &IoContext<Message>, _stream: StreamToken) {}
+	/// Called when an IO stream can be read from
+	fn stream_readable(&self, _io: &IoContext<Message>, _stream: StreamToken) {}
+	/// Called when an IO stream can be written to
+	fn stream_writable(&self, _io: &IoContext<Message>, _stream: StreamToken) {}
+	/// Register a new stream with the event loop
+	fn register_stream(&self, _stream: StreamToken, _reg: Token, _event_loop: &mut EventLoop<IoManager<Message>>) {}
+	/// Re-register a stream with the event loop
+	fn update_stream(&self, _stream: StreamToken, _reg: Token, _event_loop: &mut EventLoop<IoManager<Message>>) {}
+	/// Deregister a stream. Called whenstream is removed from event loop
+	fn deregister_stream(&self, _stream: StreamToken, _event_loop: &mut EventLoop<IoManager<Message>>) {}
+}
+
+pub use service::TimerToken;
+pub use service::StreamToken;
+pub use service::IoContext;
+pub use service::IoService;
+pub use service::IoChannel;
+pub use service::IoManager;
+pub use service::TOKENS_PER_HANDLER;
+pub use panics::{PanicHandler, MayPanic, OnPanicListener, ForwardPanic};
+
+#[cfg(test)]
+mod tests {
+
+	use std::sync::Arc;
+	use super::*;
+
+	struct MyHandler;
+
+	#[derive(Clone)]
+	struct MyMessage {
+		data: u32
+	}
+
+	impl IoHandler<MyMessage> for MyHandler {
+		fn initialize(&self, io: &IoContext<MyMessage>) {
+			io.register_timer(0, 1000).unwrap();
+		}
+
+		fn timeout(&self, _io: &IoContext<MyMessage>, timer: TimerToken) {
+			println!("Timeout {}", timer);
+		}
+
+		fn message(&self, _io: &IoContext<MyMessage>, message: &MyMessage) {
+			println!("Message {}", message.data);
+		}
+	}
+
+	#[test]
+	fn test_service_register_handler () {
+		let service = IoService::<MyMessage>::start().expect("Error creating network service");
+		service.register_handler(Arc::new(MyHandler)).unwrap();
+	}
+
+}

util/io/src/panics.rs

@@ -0,0 +1,192 @@
+// Copyright 2015, 2016 Ethcore (UK) Ltd.
+// This file is part of Parity.
+
+// Parity is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Panic utilities
+
+use std::thread;
+use std::ops::DerefMut;
+use std::sync::Arc;
+use std::default::Default;
+
+use parking_lot::Mutex;
+
+/// Thread-safe closure for handling possible panics
+pub trait OnPanicListener: Send + Sync + 'static {
+	/// Invoke listener
+	fn call(&mut self, arg: &str);
+}
+
+/// Forwards panics from child
+pub trait ForwardPanic {
+	/// Attach `on_panic` listener to `child` and rethrow all panics
+	fn forward_from<S>(&self, child: &S) where S : MayPanic;
+}
+
+/// Trait indicating that the structure catches some of the panics (most probably from spawned threads)
+/// and it's possbile to be notified when one of the threads panics.
+pub trait MayPanic {
+	/// `closure` will be invoked whenever panic in thread is caught
+	fn on_panic<F>(&self, closure: F) where F: OnPanicListener;
+}
+
+struct PanicGuard<'a> {
+	handler: &'a PanicHandler,
+}
+
+impl<'a> Drop for PanicGuard<'a> {
+	fn drop(&mut self) {
+		if thread::panicking() {
+			self.handler.notify_all("Panic!".to_owned());
+		}
+	}
+}
+
+/// Structure that allows to catch panics and notify listeners
+pub struct PanicHandler {
+	listeners: Mutex<Vec<Box<OnPanicListener>>>
+}
+
+impl Default for PanicHandler {
+	fn default() -> Self {
+		PanicHandler::new()
+	}
+}
+
+impl PanicHandler {
+	/// Creates new `PanicHandler` wrapped in `Arc`
+	pub fn new_in_arc() -> Arc<Self> {
+		Arc::new(Self::new())
+	}
+
+	/// Creates new `PanicHandler`
+	pub fn new() -> Self {
+		PanicHandler {
+			listeners: Mutex::new(vec![])
+		}
+	}
+
+	/// Invoke closure and catch any possible panics.
+	/// In case of panic notifies all listeners about it.
+	#[cfg_attr(feature="dev", allow(deprecated))]
+	pub fn catch_panic<G, R>(&self, g: G) -> thread::Result<R> where G: FnOnce() -> R + Send + 'static {
+		let _guard = PanicGuard { handler: self };
+		let result = g();
+		Ok(result)
+	}
+
+	/// Notifies all listeners in case there is a panic.
+	/// You should use `catch_panic` instead of calling this method explicitly.
+	pub fn notify_all(&self, r: String) {
+		let mut listeners = self.listeners.lock();
+		for listener in listeners.deref_mut() {
+			listener.call(&r);
+		}
+	}
+}
+
+impl MayPanic for PanicHandler {
+	fn on_panic<F>(&self, closure: F) where F: OnPanicListener {
+		self.listeners.lock().push(Box::new(closure));
+	}
+}
+
+impl ForwardPanic for Arc<PanicHandler> {
+	fn forward_from<S>(&self, child: &S) where S : MayPanic {
+		let p = self.clone();
+		child.on_panic(move |t| p.notify_all(t));
+	}
+}
+
+impl<F> OnPanicListener for F
+	where F: FnMut(String) + Send + Sync + 'static {
+	fn call(&mut self, arg: &str) {
+		self(arg.to_owned())
+	}
+}
+
+#[test]
+#[ignore] // panic forwarding doesnt work on the same thread in beta
+fn should_notify_listeners_about_panic () {
+	use parking_lot::RwLock;
+	// given
+	let invocations = Arc::new(RwLock::new(vec![]));
+	let i = invocations.clone();
+	let p = PanicHandler::new();
+	p.on_panic(move |t| i.write().push(t));
+
+	// when
+	p.catch_panic(|| panic!("Panic!")).unwrap_err();
+
+	// then
+	assert!(invocations.read()[0] == "Panic!");
+}
+
+#[test]
+#[ignore] // panic forwarding doesnt work on the same thread in beta
+fn should_notify_listeners_about_panic_when_string_is_dynamic () {
+	use parking_lot::RwLock;
+	// given
+	let invocations = Arc::new(RwLock::new(vec![]));
+	let i = invocations.clone();
+	let p = PanicHandler::new();
+	p.on_panic(move |t| i.write().push(t));
+
+	// when
+	p.catch_panic(|| panic!("Panic: {}", 1)).unwrap_err();
+
+	// then
+	assert!(invocations.read()[0] == "Panic: 1");
+}
+
+#[test]
+fn should_notify_listeners_about_panic_in_other_thread () {
+	use std::thread;
+	use parking_lot::RwLock;
+	// given
+	let invocations = Arc::new(RwLock::new(vec![]));
+	let i = invocations.clone();
+	let p = PanicHandler::new();
+	p.on_panic(move |t| i.write().push(t));
+
+	// when
+	let t = thread::spawn(move ||
+		p.catch_panic(|| panic!("Panic!")).unwrap()
+	);
+	t.join().unwrap_err();
+
+	// then
+	assert!(invocations.read()[0] == "Panic!");
+}
+
+#[test]
+#[ignore] // panic forwarding doesnt work on the same thread in beta
+fn should_forward_panics () {
+	use parking_lot::RwLock;
+	// given
+	let invocations = Arc::new(RwLock::new(vec![]));
+	let i = invocations.clone();
+	let p = PanicHandler::new_in_arc();
+	p.on_panic(move |t| i.write().push(t));
+
+	let p2 = PanicHandler::new();
+	p.forward_from(&p2);
+
+	// when
+	p2.catch_panic(|| panic!("Panic!")).unwrap_err();
+
+	// then
+	assert!(invocations.read()[0] == "Panic!");
+}

util/io/src/service.rs

@@ -0,0 +1,409 @@
+// Copyright 2015, 2016 Ethcore (UK) Ltd.
+// This file is part of Parity.
+
+// Parity is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity.  If not, see <http://www.gnu.org/licenses/>.
+
+use std::sync::Arc;
+use std::thread::{self, JoinHandle};
+use std::collections::HashMap;
+use mio::*;
+use crossbeam::sync::chase_lev;
+use slab::Slab;
+use {IoError, IoHandler};
+use worker::{Worker, Work, WorkType};
+use panics::*;
+use parking_lot::{RwLock};
+use std::sync::{Condvar as SCondvar, Mutex as SMutex};
+
+/// Timer ID
+pub type TimerToken = usize;
+/// Timer ID
+pub type StreamToken = usize;
+/// IO Hadndler ID
+pub type HandlerId = usize;
+
+/// Maximum number of tokens a handler can use
+pub const TOKENS_PER_HANDLER: usize = 16384;
+const MAX_HANDLERS: usize = 8;
+
+/// Messages used to communicate with the event loop from other threads.
+#[derive(Clone)]
+pub enum IoMessage<Message> where Message: Send + Clone + Sized {
+	/// Shutdown the event loop
+	Shutdown,
+	/// Register a new protocol handler.
+	AddHandler {
+		handler: Arc<IoHandler<Message>+Send>,
+	},
+	RemoveHandler {
+		handler_id: HandlerId,
+	},
+	AddTimer {
+		handler_id: HandlerId,
+		token: TimerToken,
+		delay: u64,
+	},
+	RemoveTimer {
+		handler_id: HandlerId,
+		token: TimerToken,
+	},
+	RegisterStream {
+		handler_id: HandlerId,
+		token: StreamToken,
+	},
+	DeregisterStream {
+		handler_id: HandlerId,
+		token: StreamToken,
+	},
+	UpdateStreamRegistration {
+		handler_id: HandlerId,
+		token: StreamToken,
+	},
+	/// Broadcast a message across all protocol handlers.
+	UserMessage(Message)
+}
+
+/// IO access point. This is passed to all IO handlers and provides an interface to the IO subsystem.
+pub struct IoContext<Message> where Message: Send + Clone + 'static {
+	channel: IoChannel<Message>,
+	handler: HandlerId,
+}
+
+impl<Message> IoContext<Message> where Message: Send + Clone + 'static {
+	/// Create a new IO access point. Takes references to all the data that can be updated within the IO handler.
+	pub fn new(channel: IoChannel<Message>, handler: HandlerId) -> IoContext<Message> {
+		IoContext {
+			handler: handler,
+			channel: channel,
+		}
+	}
+
+	/// Register a new IO timer. 'IoHandler::timeout' will be called with the token.
+	pub fn register_timer(&self, token: TimerToken, ms: u64) -> Result<(), IoError> {
+		try!(self.channel.send_io(IoMessage::AddTimer {
+			token: token,
+			delay: ms,
+			handler_id: self.handler,
+		}));
+		Ok(())
+	}
+
+	/// Delete a timer.
+	pub fn clear_timer(&self, token: TimerToken) -> Result<(), IoError> {
+		try!(self.channel.send_io(IoMessage::RemoveTimer {
+			token: token,
+			handler_id: self.handler,
+		}));
+		Ok(())
+	}
+
+	/// Register a new IO stream.
+	pub fn register_stream(&self, token: StreamToken) -> Result<(), IoError> {
+		try!(self.channel.send_io(IoMessage::RegisterStream {
+			token: token,
+			handler_id: self.handler,
+		}));
+		Ok(())
+	}
+
+	/// Deregister an IO stream.
+	pub fn deregister_stream(&self, token: StreamToken) -> Result<(), IoError> {
+		try!(self.channel.send_io(IoMessage::DeregisterStream {
+			token: token,
+			handler_id: self.handler,
+		}));
+		Ok(())
+	}
+
+	/// Reregister an IO stream.
+	pub fn update_registration(&self, token: StreamToken) -> Result<(), IoError> {
+		try!(self.channel.send_io(IoMessage::UpdateStreamRegistration {
+			token: token,
+			handler_id: self.handler,
+		}));
+		Ok(())
+	}
+
+	/// Broadcast a message to other IO clients
+	pub fn message(&self, message: Message) -> Result<(), IoError> {
+		try!(self.channel.send(message));
+		Ok(())
+	}
+
+	/// Get message channel
+	pub fn channel(&self) -> IoChannel<Message> {
+		self.channel.clone()
+	}
+
+	/// Unregister current IO handler.
+	pub fn unregister_handler(&self) -> Result<(), IoError> {
+		try!(self.channel.send_io(IoMessage::RemoveHandler {
+			handler_id: self.handler,
+		}));
+		Ok(())
+	}
+
+}
+
+#[derive(Clone)]
+struct UserTimer {
+	delay: u64,
+	timeout: Timeout,
+}
+
+/// Root IO handler. Manages user handlers, messages and IO timers.
+pub struct IoManager<Message> where Message: Send + Sync {
+	timers: Arc<RwLock<HashMap<HandlerId, UserTimer>>>,
+	handlers: Slab<Arc<IoHandler<Message>>, HandlerId>,
+	workers: Vec<Worker>,
+	worker_channel: chase_lev::Worker<Work<Message>>,
+	work_ready: Arc<SCondvar>,
+}
+
+impl<Message> IoManager<Message> where Message: Send + Sync + Clone + 'static {
+	/// Creates a new instance and registers it with the event loop.
+	pub fn start(panic_handler: Arc<PanicHandler>, event_loop: &mut EventLoop<IoManager<Message>>) -> Result<(), IoError> {
+		let (worker, stealer) = chase_lev::deque();
+		let num_workers = 4;
+		let work_ready_mutex =  Arc::new(SMutex::new(()));
+		let work_ready = Arc::new(SCondvar::new());
+		let workers = (0..num_workers).map(|i|
+			Worker::new(
+				i,
+				stealer.clone(),
+				IoChannel::new(event_loop.channel()),
+				work_ready.clone(),
+				work_ready_mutex.clone(),
+				panic_handler.clone(),
+			)
+		).collect();
+
+		let mut io = IoManager {
+			timers: Arc::new(RwLock::new(HashMap::new())),
+			handlers: Slab::new(MAX_HANDLERS),
+			worker_channel: worker,
+			workers: workers,
+			work_ready: work_ready,
+		};
+		try!(event_loop.run(&mut io));
+		Ok(())
+	}
+}
+
+impl<Message> Handler for IoManager<Message> where Message: Send + Clone + Sync + 'static {
+	type Timeout = Token;
+	type Message = IoMessage<Message>;
+
+	fn ready(&mut self, _event_loop: &mut EventLoop<Self>, token: Token, events: EventSet) {
+		let handler_index  = token.as_usize() / TOKENS_PER_HANDLER;
+		let token_id  = token.as_usize() % TOKENS_PER_HANDLER;
+		if let Some(handler) = self.handlers.get(handler_index) {
+			if events.is_hup() {
+				self.worker_channel.push(Work { work_type: WorkType::Hup, token: token_id, handler: handler.clone(), handler_id: handler_index });
+			}
+			else {
+				if events.is_readable() {
+					self.worker_channel.push(Work { work_type: WorkType::Readable, token: token_id, handler: handler.clone(), handler_id: handler_index });
+				}
+				if events.is_writable() {
+					self.worker_channel.push(Work { work_type: WorkType::Writable, token: token_id, handler: handler.clone(), handler_id: handler_index });
+				}
+			}
+			self.work_ready.notify_all();
+		}
+	}
+
+	fn timeout(&mut self, event_loop: &mut EventLoop<Self>, token: Token) {
+		let handler_index  = token.as_usize()  / TOKENS_PER_HANDLER;
+		let token_id  = token.as_usize()  % TOKENS_PER_HANDLER;
+		if let Some(handler) = self.handlers.get(handler_index) {
+			if let Some(timer) = self.timers.read().get(&token.as_usize()) {
+				event_loop.timeout_ms(token, timer.delay).expect("Error re-registering user timer");
+				self.worker_channel.push(Work { work_type: WorkType::Timeout, token: token_id, handler: handler.clone(), handler_id: handler_index });
+				self.work_ready.notify_all();
+			}
+		}
+	}
+
+	fn notify(&mut self, event_loop: &mut EventLoop<Self>, msg: Self::Message) {
+		match msg {
+			IoMessage::Shutdown => {
+				self.workers.clear();
+				event_loop.shutdown();
+			},
+			IoMessage::AddHandler { handler } => {
+				let handler_id = self.handlers.insert(handler.clone()).unwrap_or_else(|_| panic!("Too many handlers registered"));
+				handler.initialize(&IoContext::new(IoChannel::new(event_loop.channel()), handler_id));
+			},
+			IoMessage::RemoveHandler { handler_id } => {
+				// TODO: flush event loop
+				self.handlers.remove(handler_id);
+				// unregister timers
+				let mut timers = self.timers.write();
+				let to_remove: Vec<_> = timers.keys().cloned().filter(|timer_id| timer_id / TOKENS_PER_HANDLER == handler_id).collect();
+				for timer_id in to_remove {
+					let timer = timers.remove(&timer_id).expect("to_remove only contains keys from timers; qed");
+					event_loop.clear_timeout(timer.timeout);
+				}
+			},
+			IoMessage::AddTimer { handler_id, token, delay } => {
+				let timer_id = token + handler_id * TOKENS_PER_HANDLER;
+				let timeout = event_loop.timeout_ms(Token(timer_id), delay).expect("Error registering user timer");
+				self.timers.write().insert(timer_id, UserTimer { delay: delay, timeout: timeout });
+			},
+			IoMessage::RemoveTimer { handler_id, token } => {
+				let timer_id = token + handler_id * TOKENS_PER_HANDLER;
+				if let Some(timer) = self.timers.write().remove(&timer_id) {
+					event_loop.clear_timeout(timer.timeout);
+				}
+			},
+			IoMessage::RegisterStream { handler_id, token } => {
+				if let Some(handler) = self.handlers.get(handler_id) {
+					handler.register_stream(token, Token(token + handler_id * TOKENS_PER_HANDLER), event_loop);
+				}
+			},
+			IoMessage::DeregisterStream { handler_id, token } => {
+				if let Some(handler) = self.handlers.get(handler_id) {
+					handler.deregister_stream(token, event_loop);
+					// unregister a timer associated with the token (if any)
+					let timer_id = token + handler_id * TOKENS_PER_HANDLER;
+					if let Some(timer) = self.timers.write().remove(&timer_id) {
+						event_loop.clear_timeout(timer.timeout);
+					}
+				}
+			},
+			IoMessage::UpdateStreamRegistration { handler_id, token } => {
+				if let Some(handler) = self.handlers.get(handler_id) {
+					handler.update_stream(token, Token(token + handler_id * TOKENS_PER_HANDLER), event_loop);
+				}
+			},
+			IoMessage::UserMessage(data) => {
+				//TODO: better way to iterate the slab
+				for id in 0 .. MAX_HANDLERS {
+					if let Some(h) = self.handlers.get(id) {
+						let handler = h.clone();
+						self.worker_channel.push(Work { work_type: WorkType::Message(data.clone()), token: 0, handler: handler, handler_id: id });
+					}
+				}
+				self.work_ready.notify_all();
+			}
+		}
+	}
+}
+
+/// Allows sending messages into the event loop. All the IO handlers will get the message
+/// in the `message` callback.
+pub struct IoChannel<Message> where Message: Send + Clone{
+	channel: Option<Sender<IoMessage<Message>>>
+}
+
+impl<Message> Clone for IoChannel<Message> where Message: Send + Clone {
+	fn clone(&self) -> IoChannel<Message> {
+		IoChannel {
+			channel: self.channel.clone()
+		}
+	}
+}
+
+impl<Message> IoChannel<Message> where Message: Send + Clone {
+	/// Send a msessage through the channel
+	pub fn send(&self, message: Message) -> Result<(), IoError> {
+		if let Some(ref channel) = self.channel {
+			try!(channel.send(IoMessage::UserMessage(message)));
+		}
+		Ok(())
+	}
+
+	/// Send low level io message
+	pub fn send_io(&self, message: IoMessage<Message>) -> Result<(), IoError> {
+		if let Some(ref channel) = self.channel {
+			try!(channel.send(message))
+		}
+		Ok(())
+	}
+	/// Create a new channel to connected to event loop.
+	pub fn disconnected() -> IoChannel<Message> {
+		IoChannel { channel: None }
+	}
+
+	fn new(channel: Sender<IoMessage<Message>>) -> IoChannel<Message> {
+		IoChannel { channel: Some(channel) }
+	}
+}
+
+/// General IO Service. Starts an event loop and dispatches IO requests.
+/// 'Message' is a notification message type
+pub struct IoService<Message> where Message: Send + Sync + Clone + 'static {
+	panic_handler: Arc<PanicHandler>,
+	thread: Option<JoinHandle<()>>,
+	host_channel: Sender<IoMessage<Message>>,
+}
+
+impl<Message> MayPanic for IoService<Message> where Message: Send + Sync + Clone + 'static {
+	fn on_panic<F>(&self, closure: F) where F: OnPanicListener {
+		self.panic_handler.on_panic(closure);
+	}
+}
+
+impl<Message> IoService<Message> where Message: Send + Sync + Clone + 'static {
+	/// Starts IO event loop
+	pub fn start() -> Result<IoService<Message>, IoError> {
+		let panic_handler = PanicHandler::new_in_arc();
+		let mut config = EventLoopConfig::new();
+		config.messages_per_tick(1024);
+		let mut event_loop = EventLoop::configured(config).expect("Error creating event loop");
+		let channel = event_loop.channel();
+		let panic = panic_handler.clone();
+		let thread = thread::spawn(move || {
+			let p = panic.clone();
+			panic.catch_panic(move || {
+				IoManager::<Message>::start(p, &mut event_loop).unwrap();
+			}).unwrap()
+		});
+		Ok(IoService {
+			panic_handler: panic_handler,
+			thread: Some(thread),
+			host_channel: channel
+		})
+	}
+
+	/// Regiter an IO handler with the event loop.
+	pub fn register_handler(&self, handler: Arc<IoHandler<Message>+Send>) -> Result<(), IoError> {
+		try!(self.host_channel.send(IoMessage::AddHandler {
+			handler: handler,
+		}));
+		Ok(())
+	}
+
+	/// Send a message over the network. Normaly `HostIo::send` should be used. This can be used from non-io threads.
+	pub fn send_message(&self, message: Message) -> Result<(), IoError> {
+		try!(self.host_channel.send(IoMessage::UserMessage(message)));
+		Ok(())
+	}
+
+	/// Create a new message channel
+	pub fn channel(&self) -> IoChannel<Message> {
+		IoChannel { channel: Some(self.host_channel.clone()) }
+	}
+}
+
+impl<Message> Drop for IoService<Message> where Message: Send + Sync + Clone {
+	fn drop(&mut self) {
+		trace!(target: "shutdown", "[IoService] Closing...");
+		self.host_channel.send(IoMessage::Shutdown).unwrap_or_else(|e| warn!("Error on IO service shutdown: {:?}", e));
+		self.thread.take().unwrap().join().ok();
+		trace!(target: "shutdown", "[IoService] Closed.");
+	}
+}
+

util/io/src/worker.rs

@@ -0,0 +1,133 @@
+// Copyright 2015, 2016 Ethcore (UK) Ltd.
+// This file is part of Parity.
+
+// Parity is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Parity is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Parity.  If not, see <http://www.gnu.org/licenses/>.
+
+use std::sync::Arc;
+use std::mem;
+use std::thread::{JoinHandle, self};
+use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
+use crossbeam::sync::chase_lev;
+use service::{HandlerId, IoChannel, IoContext};
+use IoHandler;
+use panics::*;
+
+use std::sync::{Condvar as SCondvar, Mutex as SMutex};
+
+pub enum WorkType<Message> {
+	Readable,
+	Writable,
+	Hup,
+	Timeout,
+	Message(Message)
+}
+
+pub struct Work<Message> {
+	pub work_type: WorkType<Message>,
+	pub token: usize,
+	pub handler_id: HandlerId,
+	pub handler: Arc<IoHandler<Message>>,
+}
+
+/// An IO worker thread
+/// Sorts them ready for blockchain insertion.
+pub struct Worker {
+	thread: Option<JoinHandle<()>>,
+	wait: Arc<SCondvar>,
+	deleting: Arc<AtomicBool>,
+	wait_mutex: Arc<SMutex<()>>,
+}
+
+impl Worker {
+	/// Creates a new worker instance.
+	pub fn new<Message>(index: usize,
+						stealer: chase_lev::Stealer<Work<Message>>,
+						channel: IoChannel<Message>,
+						wait: Arc<SCondvar>,
+						wait_mutex: Arc<SMutex<()>>,
+						panic_handler: Arc<PanicHandler>
+					   ) -> Worker
+					where Message: Send + Sync + Clone + 'static {
+		let deleting = Arc::new(AtomicBool::new(false));
+		let mut worker = Worker {
+			thread: None,
+			wait: wait.clone(),
+			deleting: deleting.clone(),
+			wait_mutex: wait_mutex.clone(),
+		};
+		worker.thread = Some(thread::Builder::new().name(format!("IO Worker #{}", index)).spawn(
+			move || {
+				panic_handler.catch_panic(move || {
+					Worker::work_loop(stealer, channel.clone(), wait, wait_mutex.clone(), deleting)
+				}).unwrap()
+			})
+			.expect("Error creating worker thread"));
+		worker
+	}
+
+	fn work_loop<Message>(stealer: chase_lev::Stealer<Work<Message>>,
+						channel: IoChannel<Message>, wait: Arc<SCondvar>,
+						wait_mutex: Arc<SMutex<()>>,
+						deleting: Arc<AtomicBool>)
+						where Message: Send + Sync + Clone + 'static {
+		loop {
+			{
+				let lock = wait_mutex.lock().unwrap();
+				if deleting.load(AtomicOrdering::Acquire) {
+					return;
+				}
+				let _ = wait.wait(lock);
+			}
+
+			if deleting.load(AtomicOrdering::Acquire) {
+				return;
+			}
+			while let chase_lev::Steal::Data(work) = stealer.steal() {
+				Worker::do_work(work, channel.clone());
+			}
+		}
+	}
+
+	fn do_work<Message>(work: Work<Message>, channel: IoChannel<Message>) where Message: Send + Sync + Clone + 'static {
+		match work.work_type {
+			WorkType::Readable => {
+				work.handler.stream_readable(&IoContext::new(channel, work.handler_id), work.token);
+			},
+			WorkType::Writable => {
+				work.handler.stream_writable(&IoContext::new(channel, work.handler_id), work.token);
+			}
+			WorkType::Hup => {
+				work.handler.stream_hup(&IoContext::new(channel, work.handler_id), work.token);
+			}
+			WorkType::Timeout => {
+				work.handler.timeout(&IoContext::new(channel, work.handler_id), work.token);
+			}
+			WorkType::Message(message) => {
+				work.handler.message(&IoContext::new(channel, work.handler_id), &message);
+			}
+		}
+	}
+}
+
+impl Drop for Worker {
+	fn drop(&mut self) {
+		trace!(target: "shutdown", "[IoWorker] Closing...");
+		let _ = self.wait_mutex.lock().unwrap();
+		self.deleting.store(true, AtomicOrdering::Release);
+		self.wait.notify_all();
+		let thread = mem::replace(&mut self.thread, None).unwrap();
+		thread.join().ok();
+		trace!(target: "shutdown", "[IoWorker] Closed");
+	}
+}