openethereum/src/io/service.rs

211 lines
6.4 KiB
Rust
Raw Normal View History

2016-01-12 17:33:40 +01:00
use std::thread::{self, JoinHandle};
use mio::*;
use mio::util::{Slab};
use hash::*;
use rlp::*;
use error::*;
use io::{IoError, IoHandler};
2016-01-12 17:33:40 +01:00
pub type TimerToken = usize;
pub type StreamToken = usize;
// Tokens
const MAX_USER_TIMERS: usize = 32;
const USER_TIMER: usize = 0;
const LAST_USER_TIMER: usize = USER_TIMER + MAX_USER_TIMERS - 1;
2016-01-13 15:08:36 +01:00
//const USER_TOKEN: usize = LAST_USER_TIMER + 1;
2016-01-12 17:33:40 +01:00
/// Messages used to communicate with the event loop from other threads.
2016-01-13 13:56:48 +01:00
pub enum IoMessage<Message> where Message: Send + Sized {
2016-01-12 17:33:40 +01:00
/// Shutdown the event loop
Shutdown,
/// Register a new protocol handler.
AddHandler {
2016-01-13 13:56:48 +01:00
handler: Box<IoHandler<Message>+Send>,
2016-01-12 17:33:40 +01:00
},
/// Broadcast a message across all protocol handlers.
2016-01-13 13:56:48 +01:00
UserMessage(Message)
2016-01-12 17:33:40 +01:00
}
/// IO access point. This is passed to all IO handlers and provides an interface to the IO subsystem.
2016-01-13 13:56:48 +01:00
pub struct IoContext<'s, Message> where Message: Send + 'static {
2016-01-12 17:33:40 +01:00
timers: &'s mut Slab<UserTimer>,
2016-01-13 13:56:48 +01:00
pub event_loop: &'s mut EventLoop<IoManager<Message>>,
2016-01-12 17:33:40 +01:00
}
2016-01-13 13:56:48 +01:00
impl<'s, Message> IoContext<'s, Message> where Message: Send + 'static {
2016-01-12 17:33:40 +01:00
/// Create a new IO access point. Takes references to all the data that can be updated within the IO handler.
2016-01-13 13:56:48 +01:00
fn new(event_loop: &'s mut EventLoop<IoManager<Message>>, timers: &'s mut Slab<UserTimer>) -> IoContext<'s, Message> {
2016-01-12 17:33:40 +01:00
IoContext {
event_loop: event_loop,
timers: timers,
}
}
/// Register a new IO timer. Returns a new timer token. 'IoHandler::timeout' will be called with the token.
pub fn register_timer(&mut self, ms: u64) -> Result<TimerToken, UtilError>{
match self.timers.insert(UserTimer {
delay: ms,
}) {
Ok(token) => {
self.event_loop.timeout_ms(token, ms).expect("Error registering user timer");
Ok(token.as_usize())
},
_ => { panic!("Max timers reached") }
}
}
/// Broadcast a message to other IO clients
2016-01-13 13:56:48 +01:00
pub fn message(&mut self, message: Message) {
match self.event_loop.channel().send(IoMessage::UserMessage(message)) {
2016-01-12 17:33:40 +01:00
Ok(_) => {}
Err(e) => { panic!("Error sending io message {:?}", e); }
}
}
}
struct UserTimer {
delay: u64,
}
/// Root IO handler. Manages user handlers, messages and IO timers.
2016-01-13 13:56:48 +01:00
pub struct IoManager<Message> where Message: Send {
2016-01-12 17:33:40 +01:00
timers: Slab<UserTimer>,
2016-01-13 13:56:48 +01:00
handlers: Vec<Box<IoHandler<Message>>>,
2016-01-12 17:33:40 +01:00
}
2016-01-13 13:56:48 +01:00
impl<Message> IoManager<Message> where Message: Send + 'static {
2016-01-12 17:33:40 +01:00
/// Creates a new instance and registers it with the event loop.
2016-01-13 13:56:48 +01:00
pub fn start(event_loop: &mut EventLoop<IoManager<Message>>) -> Result<(), UtilError> {
2016-01-12 17:33:40 +01:00
let mut io = IoManager {
timers: Slab::new_starting_at(Token(USER_TIMER), MAX_USER_TIMERS),
handlers: Vec::new(),
};
try!(event_loop.run(&mut io));
Ok(())
}
}
2016-01-13 13:56:48 +01:00
impl<Message> Handler for IoManager<Message> where Message: Send + 'static {
2016-01-12 17:33:40 +01:00
type Timeout = Token;
2016-01-13 13:56:48 +01:00
type Message = IoMessage<Message>;
2016-01-12 17:33:40 +01:00
fn ready(&mut self, event_loop: &mut EventLoop<Self>, token: Token, events: EventSet) {
if events.is_hup() {
for h in self.handlers.iter_mut() {
2016-01-13 13:56:48 +01:00
h.stream_hup(&mut IoContext::new(event_loop, &mut self.timers), token.as_usize());
2016-01-12 17:33:40 +01:00
}
}
else if events.is_readable() {
for h in self.handlers.iter_mut() {
2016-01-13 13:56:48 +01:00
h.stream_readable(&mut IoContext::new(event_loop, &mut self.timers), token.as_usize());
2016-01-12 17:33:40 +01:00
}
}
else if events.is_writable() {
for h in self.handlers.iter_mut() {
2016-01-13 13:56:48 +01:00
h.stream_writable(&mut IoContext::new(event_loop, &mut self.timers), token.as_usize());
2016-01-12 17:33:40 +01:00
}
}
}
fn timeout(&mut self, event_loop: &mut EventLoop<Self>, token: Token) {
match token.as_usize() {
USER_TIMER ... LAST_USER_TIMER => {
let delay = {
let timer = self.timers.get_mut(token).expect("Unknown user timer token");
timer.delay
};
for h in self.handlers.iter_mut() {
2016-01-13 13:56:48 +01:00
h.timeout(&mut IoContext::new(event_loop, &mut self.timers), token.as_usize());
2016-01-12 17:33:40 +01:00
}
event_loop.timeout_ms(token, delay).expect("Error re-registering user timer");
}
_ => { // Just pass the event down. IoHandler is supposed to re-register it if required.
for h in self.handlers.iter_mut() {
2016-01-13 13:56:48 +01:00
h.timeout(&mut IoContext::new(event_loop, &mut self.timers), token.as_usize());
2016-01-12 17:33:40 +01:00
}
}
}
}
fn notify(&mut self, event_loop: &mut EventLoop<Self>, msg: Self::Message) {
let mut m = msg;
match m {
2016-01-12 17:33:40 +01:00
IoMessage::Shutdown => event_loop.shutdown(),
IoMessage::AddHandler {
handler,
} => {
self.handlers.push(handler);
2016-01-14 16:52:10 +01:00
self.handlers.last_mut().unwrap().initialize(&mut IoContext::new(event_loop, &mut self.timers));
2016-01-12 17:33:40 +01:00
},
IoMessage::UserMessage(ref mut data) => {
2016-01-12 17:33:40 +01:00
for h in self.handlers.iter_mut() {
2016-01-13 13:56:48 +01:00
h.message(&mut IoContext::new(event_loop, &mut self.timers), data);
2016-01-12 17:33:40 +01:00
}
}
}
}
}
2016-01-13 23:13:57 +01:00
/// 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 {
channel: Sender<IoMessage<Message>>
}
impl<Message> IoChannel<Message> where Message: Send {
pub fn send(&mut self, message: Message) -> Result<(), IoError> {
try!(self.channel.send(IoMessage::UserMessage(message)));
Ok(())
}
}
2016-01-12 17:33:40 +01:00
/// General IO Service. Starts an event loop and dispatches IO requests.
2016-01-13 13:56:48 +01:00
/// 'Message' is a notification message type
pub struct IoService<Message> where Message: Send + 'static {
2016-01-12 17:33:40 +01:00
thread: Option<JoinHandle<()>>,
2016-01-13 13:56:48 +01:00
host_channel: Sender<IoMessage<Message>>
2016-01-12 17:33:40 +01:00
}
2016-01-13 13:56:48 +01:00
impl<Message> IoService<Message> where Message: Send + 'static {
2016-01-12 17:33:40 +01:00
/// Starts IO event loop
2016-01-13 13:56:48 +01:00
pub fn start() -> Result<IoService<Message>, UtilError> {
2016-01-12 17:33:40 +01:00
let mut event_loop = EventLoop::new().unwrap();
let channel = event_loop.channel();
let thread = thread::spawn(move || {
2016-01-13 13:56:48 +01:00
IoManager::<Message>::start(&mut event_loop).unwrap(); //TODO:
2016-01-12 17:33:40 +01:00
});
Ok(IoService {
thread: Some(thread),
host_channel: channel
})
}
/// Regiter a IO hadnler with the event loop.
2016-01-13 13:56:48 +01:00
pub fn register_handler(&mut self, handler: Box<IoHandler<Message>+Send>) -> Result<(), IoError> {
2016-01-12 17:33:40 +01:00
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.
2016-01-13 13:56:48 +01:00
pub fn send_message(&mut self, message: Message) -> Result<(), IoError> {
try!(self.host_channel.send(IoMessage::UserMessage(message)));
Ok(())
}
2016-01-13 23:13:57 +01:00
/// Create a new message channel
pub fn channel(&mut self) -> IoChannel<Message> {
IoChannel { channel: self.host_channel.clone() }
}
2016-01-12 17:33:40 +01:00
}
2016-01-13 13:56:48 +01:00
impl<Message> Drop for IoService<Message> where Message: Send {
2016-01-12 17:33:40 +01:00
fn drop(&mut self) {
self.host_channel.send(IoMessage::Shutdown).unwrap();
self.thread.take().unwrap().join().unwrap();
}
}