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

Merge branch 'master' into gav

Browse Source
This commit is contained in:
Gav Wood
2016-01-24 00:09:18 +01:00
parent b059d32485 b4bfdf2fd3
commit 788213f48f
70 changed files with 1460 additions and 1099 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
util/src/bytes.rs
View File

@@ -106,18 +106,18 @@ impl<'a> Deref for BytesRef<'a> {
type Target = [u8];
fn deref(&self) -> &[u8] {
match self {
&BytesRef::Flexible(ref bytes) => bytes,
&BytesRef::Fixed(ref bytes) => bytes
match *self {
BytesRef::Flexible(ref bytes) => bytes,
BytesRef::Fixed(ref bytes) => bytes
}
}
}
impl <'a> DerefMut for BytesRef<'a> {
fn deref_mut(&mut self) -> &mut [u8] {
match self {
&mut BytesRef::Flexible(ref mut bytes) => bytes,
&mut BytesRef::Fixed(ref mut bytes) => bytes
match *self {
BytesRef::Flexible(ref mut bytes) => bytes,
BytesRef::Fixed(ref mut bytes) => bytes
}
}
}
@@ -299,7 +299,7 @@ pub trait FromBytes: Sized {
impl FromBytes for String {
fn from_bytes(bytes: &[u8]) -> FromBytesResult<String> {
Ok(::std::str::from_utf8(bytes).unwrap().to_string())
Ok(::std::str::from_utf8(bytes).unwrap().to_owned())
}
}

7
util/src/chainfilter.rs
View File

@@ -323,10 +323,9 @@ impl<'a, D> ChainFilter<'a, D> where D: FilterDataSource
let offset = level_size * index;
// go doooown!
match self.blocks(bloom, from_block, to_block, max_level, offset) {
Some(blocks) => result.extend(blocks),
None => ()
};
if let Some(blocks) = self.blocks(bloom, from_block, to_block, max_level, offset) {
result.extend(blocks);
}
}
result

29
util/src/hash.rs
View File

@@ -207,11 +207,11 @@ macro_rules! impl_hash {
impl FromJson for $from {
fn from_json(json: &Json) -> Self {
match json {
&Json::String(ref s) => {
match *json {
Json::String(ref s) => {
match s.len() % 2 {
0 => FromStr::from_str(clean_0x(s)).unwrap(),
_ => FromStr::from_str(&("0".to_string() + &(clean_0x(s).to_string()))[..]).unwrap()
_ => FromStr::from_str(&("0".to_owned() + &(clean_0x(s).to_owned()))[..]).unwrap()
}
},
_ => Default::default(),
@@ -221,7 +221,7 @@ macro_rules! impl_hash {
impl fmt::Debug for $from {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
for i in self.0.iter() {
for i in &self.0[..] {
try!(write!(f, "{:02x}", i));
}
Ok(())
@@ -229,11 +229,11 @@ macro_rules! impl_hash {
}
impl fmt::Display for $from {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
for i in self.0[0..2].iter() {
for i in &self.0[0..2] {
try!(write!(f, "{:02x}", i));
}
try!(write!(f, "…"));
for i in self.0[$size - 4..$size].iter() {
for i in &self.0[$size - 4..$size] {
try!(write!(f, "{:02x}", i));
}
Ok(())
@@ -291,36 +291,36 @@ macro_rules! impl_hash {
impl Index<usize> for $from {
type Output = u8;
fn index<'a>(&'a self, index: usize) -> &'a u8 {
fn index(&self, index: usize) -> &u8 {
&self.0[index]
}
}
impl IndexMut<usize> for $from {
fn index_mut<'a>(&'a mut self, index: usize) -> &'a mut u8 {
fn index_mut(&mut self, index: usize) -> &mut u8 {
&mut self.0[index]
}
}
impl Index<ops::Range<usize>> for $from {
type Output = [u8];
fn index<'a>(&'a self, index: ops::Range<usize>) -> &'a [u8] {
fn index(&self, index: ops::Range<usize>) -> &[u8] {
&self.0[index]
}
}
impl IndexMut<ops::Range<usize>> for $from {
fn index_mut<'a>(&'a mut self, index: ops::Range<usize>) -> &'a mut [u8] {
fn index_mut(&mut self, index: ops::Range<usize>) -> &mut [u8] {
&mut self.0[index]
}
}
impl Index<ops::RangeFull> for $from {
type Output = [u8];
fn index<'a>(&'a self, _index: ops::RangeFull) -> &'a [u8] {
fn index(&self, _index: ops::RangeFull) -> &[u8] {
&self.0
}
}
impl IndexMut<ops::RangeFull> for $from {
fn index_mut<'a>(&'a mut self, _index: ops::RangeFull) -> &'a mut [u8] {
fn index_mut(&mut self, _index: ops::RangeFull) -> &mut [u8] {
&mut self.0
}
}
@@ -440,9 +440,9 @@ macro_rules! impl_hash {
fn from(s: &'_ str) -> $from {
use std::str::FromStr;
if s.len() % 2 == 1 {
$from::from_str(&("0".to_string() + &(clean_0x(s).to_string()))[..]).unwrap_or($from::new())
$from::from_str(&("0".to_owned() + &(clean_0x(s).to_owned()))[..]).unwrap_or_else(|_| $from::new())
} else {
$from::from_str(clean_0x(s)).unwrap_or($from::new())
$from::from_str(clean_0x(s)).unwrap_or_else(|_| $from::new())
}
}
}
@@ -565,6 +565,7 @@ mod tests {
use std::str::FromStr;
#[test]
#[allow(eq_op)]
fn hash() {
let h = H64([0x01, 0x23, 0x45, 0x67, 0x89, 0xab, 0xcd, 0xef]);
assert_eq!(H64::from_str("0123456789abcdef").unwrap(), h);

61
util/src/io/mod.rs
View File

@@ -8,27 +8,28 @@
///
/// struct MyHandler;
///
/// #[derive(Clone)]
/// struct MyMessage {
/// data: u32
/// }
///
/// impl IoHandler<MyMessage> for MyHandler {
/// fn initialize(&mut self, io: &mut IoContext<MyMessage>) {
/// io.register_timer(1000).unwrap();
/// fn initialize(&self, io: &IoContext<MyMessage>) {
/// io.register_timer(0, 1000).unwrap();
/// }
///
/// fn timeout(&mut self, _io: &mut IoContext<MyMessage>, timer: TimerToken) {
/// fn timeout(&self, _io: &IoContext<MyMessage>, timer: TimerToken) {
/// println!("Timeout {}", timer);
/// }
///
/// fn message(&mut self, _io: &mut IoContext<MyMessage>, message: &mut MyMessage) {
/// 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(Box::new(MyHandler)).unwrap();
/// service.register_handler(Arc::new(MyHandler)).unwrap();
///
/// // Wait for quit condition
/// // ...
@@ -36,6 +37,9 @@
/// }
/// ```
mod service;
mod worker;
use mio::{EventLoop, Token};
#[derive(Debug)]
/// TODO [arkpar] Please document me
@@ -44,7 +48,7 @@ pub enum IoError {
Mio(::std::io::Error),
}
impl<Message> From<::mio::NotifyError<service::IoMessage<Message>>> for IoError where Message: Send {
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"))
}
@@ -53,54 +57,63 @@ impl<Message> From<::mio::NotifyError<service::IoMessage<Message>>> for IoError
/// 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 where Message: Send + 'static {
pub trait IoHandler<Message>: Send + Sync where Message: Send + Sync + Clone + 'static {
/// Initialize the handler
fn initialize<'s>(&'s mut self, _io: &mut IoContext<'s, Message>) {}
fn initialize(&self, _io: &IoContext<Message>) {}
/// Timer function called after a timeout created with `HandlerIo::timeout`.
fn timeout<'s>(&'s mut self, _io: &mut IoContext<'s, Message>, _timer: TimerToken) {}
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<'s>(&'s mut self, _io: &mut IoContext<'s, Message>, _message: &'s mut Message) {} // TODO: make message immutable and provide internal channel for adding network handler
fn message(&self, _io: &IoContext<Message>, _message: &Message) {}
/// Called when an IO stream gets closed
fn stream_hup<'s>(&'s mut self, _io: &mut IoContext<'s, Message>, _stream: StreamToken) {}
fn stream_hup(&self, _io: &IoContext<Message>, _stream: StreamToken) {}
/// Called when an IO stream can be read from
fn stream_readable<'s>(&'s mut self, _io: &mut IoContext<'s, Message>, _stream: StreamToken) {}
fn stream_readable(&self, _io: &IoContext<Message>, _stream: StreamToken) {}
/// Called when an IO stream can be written to
fn stream_writable<'s>(&'s mut self, _io: &mut IoContext<'s, Message>, _stream: StreamToken) {}
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>>) {}
}
/// TODO [arkpar] Please document me
pub type TimerToken = service::TimerToken;
pub use io::service::TimerToken;
/// TODO [arkpar] Please document me
pub type StreamToken = service::StreamToken;
pub use io::service::StreamToken;
/// TODO [arkpar] Please document me
pub type IoContext<'s, M> = service::IoContext<'s, M>;
pub use io::service::IoContext;
/// TODO [arkpar] Please document me
pub type IoService<M> = service::IoService<M>;
pub use io::service::IoService;
/// TODO [arkpar] Please document me
pub type IoChannel<M> = service::IoChannel<M>;
//pub const USER_TOKEN_START: usize = service::USER_TOKEN; // TODO: ICE in rustc 1.7.0-nightly (49c382779 2016-01-12)
pub use io::service::IoChannel;
/// TODO [arkpar] Please document me
pub use io::service::IoManager;
/// TODO [arkpar] Please document me
pub use io::service::TOKENS_PER_HANDLER;
#[cfg(test)]
mod tests {
use std::sync::Arc;
use io::*;
struct MyHandler;
#[derive(Clone)]
struct MyMessage {
data: u32
}
impl IoHandler<MyMessage> for MyHandler {
fn initialize(&mut self, io: &mut IoContext<MyMessage>) {
io.register_timer(1000).unwrap();
fn initialize(&self, io: &IoContext<MyMessage>) {
io.register_timer(0, 1000).unwrap();
}
fn timeout(&mut self, _io: &mut IoContext<MyMessage>, timer: TimerToken) {
fn timeout(&self, _io: &IoContext<MyMessage>, timer: TimerToken) {
println!("Timeout {}", timer);
}
fn message(&mut self, _io: &mut IoContext<MyMessage>, message: &mut MyMessage) {
fn message(&self, _io: &IoContext<MyMessage>, message: &MyMessage) {
println!("Message {}", message.data);
}
}
@@ -108,7 +121,7 @@ mod tests {
#[test]
fn test_service_register_handler () {
let mut service = IoService::<MyMessage>::start().expect("Error creating network service");
service.register_handler(Box::new(MyHandler)).unwrap();
service.register_handler(Arc::new(MyHandler)).unwrap();
}
}

268
util/src/io/service.rs
View File

@@ -1,148 +1,229 @@
use std::sync::*;
use std::thread::{self, JoinHandle};
use std::collections::HashMap;
use mio::*;
use mio::util::{Slab};
use hash::*;
use rlp::*;
use error::*;
use io::{IoError, IoHandler};
use arrayvec::*;
use crossbeam::sync::chase_lev;
use io::worker::{Worker, Work, WorkType};
/// Timer ID
pub type TimerToken = usize;
/// Timer ID
pub type StreamToken = usize;
/// IO Hadndler ID
pub type HandlerId = usize;
// Tokens
const MAX_USER_TIMERS: usize = 32;
const USER_TIMER: usize = 0;
const LAST_USER_TIMER: usize = USER_TIMER + MAX_USER_TIMERS - 1;
//const USER_TOKEN: usize = LAST_USER_TIMER + 1;
/// Maximum number of tokens a handler can use
pub const TOKENS_PER_HANDLER: usize = 16384;
/// Messages used to communicate with the event loop from other threads.
pub enum IoMessage<Message> where Message: Send + Sized {
#[derive(Clone)]
pub enum IoMessage<Message> where Message: Send + Clone + Sized {
/// Shutdown the event loop
Shutdown,
/// Register a new protocol handler.
AddHandler {
handler: Box<IoHandler<Message>+Send>,
handler: Arc<IoHandler<Message>+Send>,
},
AddTimer {
handler_id: HandlerId,
token: TimerToken,
delay: u64,
},
RemoveTimer {
handler_id: HandlerId,
token: TimerToken,
},
RegisterStream {
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<'s, Message> where Message: Send + 'static {
timers: &'s mut Slab<UserTimer>,
/// Low leve MIO Event loop for custom handler registration.
pub event_loop: &'s mut EventLoop<IoManager<Message>>,
pub struct IoContext<Message> where Message: Send + Clone + 'static {
channel: IoChannel<Message>,
handler: HandlerId,
}
impl<'s, Message> IoContext<'s, Message> where Message: Send + 'static {
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.
fn new(event_loop: &'s mut EventLoop<IoManager<Message>>, timers: &'s mut Slab<UserTimer>) -> IoContext<'s, Message> {
pub fn new(channel: IoChannel<Message>, handler: HandlerId) -> IoContext<Message> {
IoContext {
event_loop: event_loop,
timers: timers,
handler: handler,
channel: channel,
}
}
/// 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 {
/// Register a new IO timer. 'IoHandler::timeout' will be called with the token.
pub fn register_timer(&self, token: TimerToken, ms: u64) -> Result<(), UtilError> {
try!(self.channel.send_io(IoMessage::AddTimer {
token: token,
delay: ms,
}) {
Ok(token) => {
self.event_loop.timeout_ms(token, ms).expect("Error registering user timer");
Ok(token.as_usize())
},
_ => { panic!("Max timers reached") }
}
handler_id: self.handler,
}));
Ok(())
}
/// Delete a timer.
pub fn clear_timer(&self, token: TimerToken) -> Result<(), UtilError> {
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<(), UtilError> {
try!(self.channel.send_io(IoMessage::RegisterStream {
token: token,
handler_id: self.handler,
}));
Ok(())
}
/// Reregister an IO stream.
pub fn update_registration(&self, token: StreamToken) -> Result<(), UtilError> {
try!(self.channel.send_io(IoMessage::UpdateStreamRegistration {
token: token,
handler_id: self.handler,
}));
Ok(())
}
/// Broadcast a message to other IO clients
pub fn message(&mut self, message: Message) {
match self.event_loop.channel().send(IoMessage::UserMessage(message)) {
Ok(_) => {}
Err(e) => { panic!("Error sending io message {:?}", e); }
}
pub fn message(&self, message: Message) {
self.channel.send(message).expect("Error seding message");
}
/// Get message channel
pub fn channel(&self) -> IoChannel<Message> {
self.channel.clone()
}
}
#[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 {
timers: Slab<UserTimer>,
handlers: Vec<Box<IoHandler<Message>>>,
pub struct IoManager<Message> where Message: Send + Sync {
timers: Arc<RwLock<HashMap<HandlerId, UserTimer>>>,
handlers: Vec<Arc<IoHandler<Message>>>,
_workers: Vec<Worker>,
worker_channel: chase_lev::Worker<Work<Message>>,
work_ready: Arc<Condvar>,
}
impl<Message> IoManager<Message> where Message: Send + 'static {
impl<Message> IoManager<Message> where Message: Send + Sync + Clone + 'static {
/// Creates a new instance and registers it with the event loop.
pub fn start(event_loop: &mut EventLoop<IoManager<Message>>) -> Result<(), UtilError> {
let (worker, stealer) = chase_lev::deque();
let num_workers = 4;
let work_ready_mutex = Arc::new(Mutex::new(()));
let work_ready = Arc::new(Condvar::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())).collect();
let mut io = IoManager {
timers: Slab::new_starting_at(Token(USER_TIMER), MAX_USER_TIMERS),
timers: Arc::new(RwLock::new(HashMap::new())),
handlers: Vec::new(),
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 + 'static {
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) {
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 handler_index >= self.handlers.len() {
panic!("Unexpected stream token: {}", token.as_usize());
}
let handler = self.handlers[handler_index].clone();
if events.is_hup() {
for h in self.handlers.iter_mut() {
h.stream_hup(&mut IoContext::new(event_loop, &mut self.timers), token.as_usize());
}
}
else if events.is_readable() {
for h in self.handlers.iter_mut() {
h.stream_readable(&mut IoContext::new(event_loop, &mut self.timers), token.as_usize());
}
}
else if events.is_writable() {
for h in self.handlers.iter_mut() {
h.stream_writable(&mut IoContext::new(event_loop, &mut self.timers), token.as_usize());
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) {
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() {
h.timeout(&mut IoContext::new(event_loop, &mut self.timers), token.as_usize());
}
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() {
h.timeout(&mut IoContext::new(event_loop, &mut self.timers), token.as_usize());
}
}
let handler_index = token.as_usize() / TOKENS_PER_HANDLER;
let token_id = token.as_usize() % TOKENS_PER_HANDLER;
if handler_index >= self.handlers.len() {
panic!("Unexpected timer token: {}", token.as_usize());
}
if let Some(timer) = self.timers.read().unwrap().get(&token.as_usize()) {
event_loop.timeout_ms(token, timer.delay).expect("Error re-registering user timer");
let handler = self.handlers[handler_index].clone();
self.worker_channel.push(Work { work_type: WorkType::Timeout, token: token_id, handler: handler, handler_id: handler_index });
self.work_ready.notify_all();
}
}
fn notify(&mut self, event_loop: &mut EventLoop<Self>, msg: Self::Message) {
let mut m = msg;
match m {
match msg {
IoMessage::Shutdown => event_loop.shutdown(),
IoMessage::AddHandler {
handler,
} => {
self.handlers.push(handler);
self.handlers.last_mut().unwrap().initialize(&mut IoContext::new(event_loop, &mut self.timers));
IoMessage::AddHandler { handler } => {
let handler_id = {
self.handlers.push(handler.clone());
self.handlers.len() - 1
};
handler.initialize(&IoContext::new(IoChannel::new(event_loop.channel()), handler_id));
},
IoMessage::UserMessage(ref mut data) => {
for h in self.handlers.iter_mut() {
h.message(&mut IoContext::new(event_loop, &mut self.timers), data);
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().unwrap().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().unwrap().remove(&timer_id) {
event_loop.clear_timeout(timer.timeout);
}
},
IoMessage::RegisterStream { handler_id, token } => {
let handler = self.handlers.get(handler_id).expect("Unknown handler id").clone();
handler.register_stream(token, Token(token + handler_id * TOKENS_PER_HANDLER), event_loop);
},
IoMessage::UpdateStreamRegistration { handler_id, token } => {
let handler = self.handlers.get(handler_id).expect("Unknown handler id").clone();
handler.update_stream(token, Token(token + handler_id * TOKENS_PER_HANDLER), event_loop);
},
IoMessage::UserMessage(data) => {
for n in 0 .. self.handlers.len() {
let handler = self.handlers[n].clone();
self.worker_channel.push(Work { work_type: WorkType::Message(data.clone()), token: 0, handler: handler, handler_id: n });
}
self.work_ready.notify_all();
}
}
}
@@ -150,11 +231,19 @@ impl<Message> Handler for IoManager<Message> where Message: Send + 'static {
/// 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 {
pub struct IoChannel<Message> where Message: Send + Clone{
channel: Option<Sender<IoMessage<Message>>>
}
impl<Message> IoChannel<Message> where Message: Send {
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 {
@@ -163,20 +252,31 @@ impl<Message> IoChannel<Message> where Message: Send {
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 + 'static {
pub struct IoService<Message> where Message: Send + Sync + Clone + 'static {
thread: Option<JoinHandle<()>>,
host_channel: Sender<IoMessage<Message>>
host_channel: Sender<IoMessage<Message>>,
}
impl<Message> IoService<Message> where Message: Send + 'static {
impl<Message> IoService<Message> where Message: Send + Sync + Clone + 'static {
/// Starts IO event loop
pub fn start() -> Result<IoService<Message>, UtilError> {
let mut event_loop = EventLoop::new().unwrap();
@@ -191,7 +291,7 @@ impl<Message> IoService<Message> where Message: Send + 'static {
}
/// Regiter a IO hadnler with the event loop.
pub fn register_handler(&mut self, handler: Box<IoHandler<Message>+Send>) -> Result<(), IoError> {
pub fn register_handler(&mut self, handler: Arc<IoHandler<Message>+Send>) -> Result<(), IoError> {
try!(self.host_channel.send(IoMessage::AddHandler {
handler: handler,
}));
@@ -210,10 +310,10 @@ impl<Message> IoService<Message> where Message: Send + 'static {
}
}
impl<Message> Drop for IoService<Message> where Message: Send {
impl<Message> Drop for IoService<Message> where Message: Send + Sync + Clone {
fn drop(&mut self) {
self.host_channel.send(IoMessage::Shutdown).unwrap();
self.thread.take().unwrap().join().unwrap();
self.thread.take().unwrap().join().ok();
}
}

99
util/src/io/worker.rs Normal file
View File

@@ -0,0 +1,99 @@
use std::sync::*;
use std::mem;
use std::thread::{JoinHandle, self};
use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
use crossbeam::sync::chase_lev;
use io::service::{HandlerId, IoChannel, IoContext};
use io::{IoHandler};
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<Condvar>,
deleting: Arc<AtomicBool>,
}
impl Worker {
/// Creates a new worker instance.
pub fn new<Message>(index: usize,
stealer: chase_lev::Stealer<Work<Message>>,
channel: IoChannel<Message>,
wait: Arc<Condvar>,
wait_mutex: Arc<Mutex<()>>) -> 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(),
};
worker.thread = Some(thread::Builder::new().name(format!("IO Worker #{}", index)).spawn(
move || Worker::work_loop(stealer, channel.clone(), wait, wait_mutex.clone(), deleting))
.expect("Error creating worker thread"));
worker
}
fn work_loop<Message>(stealer: chase_lev::Stealer<Work<Message>>,
channel: IoChannel<Message>, wait: Arc<Condvar>,
wait_mutex: Arc<Mutex<()>>,
deleting: Arc<AtomicBool>)
where Message: Send + Sync + Clone + 'static {
while !deleting.load(AtomicOrdering::Relaxed) {
{
let lock = wait_mutex.lock().unwrap();
let _ = wait.wait(lock).unwrap();
if deleting.load(AtomicOrdering::Relaxed) {
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) {
self.deleting.store(true, AtomicOrdering::Relaxed);
self.wait.notify_all();
let thread = mem::replace(&mut self.thread, None).unwrap();
thread.join().ok();
}
}

12
util/src/journaldb.rs
View File

@@ -34,6 +34,16 @@ impl JournalDB {
}
}
/// Create a new instance given a shared `backing` database.
pub fn new_with_arc(backing: Arc<DB>) -> JournalDB {
JournalDB {
forward: OverlayDB::new_with_arc(backing.clone()),
backing: backing,
inserts: vec![],
removes: vec![],
}
}
/// Create a new instance with an anonymous temporary database.
pub fn new_temp() -> JournalDB {
let mut dir = env::temp_dir();
@@ -96,7 +106,7 @@ impl JournalDB {
})) {
let rlp = Rlp::new(&rlp_data);
let to_remove: Vec<H256> = rlp.val_at(if canon_id == rlp.val_at(0) {2} else {1});
for i in to_remove.iter() {
for i in &to_remove {
self.forward.remove(i);
}
try!(self.backing.delete(&last));

28
util/src/json_aid.rs
View File

@@ -11,18 +11,18 @@ pub fn clean(s: &str) -> &str {
fn u256_from_str(s: &str) -> U256 {
if s.len() >= 2 && &s[0..2] == "0x" {
U256::from_str(&s[2..]).unwrap_or(U256::from(0))
U256::from_str(&s[2..]).unwrap_or_else(|_| U256::zero())
} else {
U256::from_dec_str(s).unwrap_or(U256::from(0))
U256::from_dec_str(s).unwrap_or_else(|_| U256::zero())
}
}
impl FromJson for Bytes {
fn from_json(json: &Json) -> Self {
match json {
&Json::String(ref s) => match s.len() % 2 {
0 => FromHex::from_hex(clean(s)).unwrap_or(vec![]),
_ => FromHex::from_hex(&("0".to_string() + &(clean(s).to_string()))[..]).unwrap_or(vec![]),
match *json {
Json::String(ref s) => match s.len() % 2 {
0 => FromHex::from_hex(clean(s)).unwrap_or_else(|_| vec![]),
_ => FromHex::from_hex(&("0".to_owned() + &(clean(s).to_owned()))[..]).unwrap_or_else(|_| vec![]),
},
_ => vec![],
}
@@ -31,8 +31,8 @@ impl FromJson for Bytes {
impl FromJson for BTreeMap<H256, H256> {
fn from_json(json: &Json) -> Self {
match json {
&Json::Object(ref o) => o.iter().map(|(key, value)| (x!(&u256_from_str(key)), x!(&U256::from_json(value)))).collect(),
match *json {
Json::Object(ref o) => o.iter().map(|(key, value)| (x!(&u256_from_str(key)), x!(&U256::from_json(value)))).collect(),
_ => BTreeMap::new(),
}
}
@@ -40,8 +40,8 @@ impl FromJson for BTreeMap<H256, H256> {
impl<T> FromJson for Vec<T> where T: FromJson {
fn from_json(json: &Json) -> Self {
match json {
&Json::Array(ref o) => o.iter().map(|x|T::from_json(x)).collect(),
match *json {
Json::Array(ref o) => o.iter().map(|x|T::from_json(x)).collect(),
_ => Vec::new(),
}
}
@@ -49,9 +49,9 @@ impl<T> FromJson for Vec<T> where T: FromJson {
impl<T> FromJson for Option<T> where T: FromJson {
fn from_json(json: &Json) -> Self {
match json {
&Json::String(ref o) if o.is_empty() => None,
&Json::Null => None,
match *json {
Json::String(ref o) if o.is_empty() => None,
Json::Null => None,
_ => Some(FromJson::from_json(json)),
}
}
@@ -135,4 +135,4 @@ fn option_types() {
assert_eq!(None, v);
let v: Option<u16> = xjson!(&j["empty"]);
assert_eq!(None, v);
}
}

4
util/src/lib.rs
View File

@@ -2,6 +2,9 @@
#![feature(op_assign_traits)]
#![feature(augmented_assignments)]
#![feature(associated_consts)]
#![feature(plugin)]
#![plugin(clippy)]
#![allow(needless_range_loop, match_bool)]
//! Ethcore-util library
//!
//! ### Rust version:
@@ -51,6 +54,7 @@ extern crate crypto as rcrypto;
extern crate secp256k1;
extern crate arrayvec;
extern crate elastic_array;
extern crate crossbeam;
/// TODO [Gav Wood] Please document me
pub mod standard;

6
util/src/misc.rs
View File

@@ -18,13 +18,13 @@ impl<T> Diff<T> where T: Eq {
pub fn new(pre: T, post: T) -> Self { if pre == post { Diff::Same } else { Diff::Changed(pre, post) } }
/// Get the before value, if there is one.
pub fn pre(&self) -> Option<&T> { match self { &Diff::Died(ref x) | &Diff::Changed(ref x, _) => Some(x), _ => None } }
pub fn pre(&self) -> Option<&T> { match *self { Diff::Died(ref x) | Diff::Changed(ref x, _) => Some(x), _ => None } }
/// Get the after value, if there is one.
pub fn post(&self) -> Option<&T> { match self { &Diff::Born(ref x) | &Diff::Changed(_, ref x) => Some(x), _ => None } }
pub fn post(&self) -> Option<&T> { match *self { Diff::Born(ref x) | Diff::Changed(_, ref x) => Some(x), _ => None } }
/// Determine whether there was a change or not.
pub fn is_same(&self) -> bool { match self { &Diff::Same => true, _ => false }}
pub fn is_same(&self) -> bool { match *self { Diff::Same => true, _ => false }}
}
#[derive(PartialEq,Eq,Clone,Copy)]

66
util/src/network/connection.rs
View File

@@ -1,5 +1,5 @@
use std::collections::VecDeque;
use mio::{Handler, Token, EventSet, EventLoop, Timeout, PollOpt, TryRead, TryWrite};
use mio::{Handler, Token, EventSet, EventLoop, PollOpt, TryRead, TryWrite};
use mio::tcp::*;
use hash::*;
use sha3::*;
@@ -7,6 +7,7 @@ use bytes::*;
use rlp::*;
use std::io::{self, Cursor, Read};
use error::*;
use io::{IoContext, StreamToken};
use network::error::NetworkError;
use network::handshake::Handshake;
use crypto;
@@ -17,11 +18,12 @@ use rcrypto::buffer::*;
use tiny_keccak::Keccak;
const ENCRYPTED_HEADER_LEN: usize = 32;
const RECIEVE_PAYLOAD_TIMEOUT: u64 = 30000;
/// Low level tcp connection
pub struct Connection {
/// Connection id (token)
pub token: Token,
pub token: StreamToken,
/// Network socket
pub socket: TcpStream,
/// Receive buffer
@@ -45,14 +47,14 @@ pub enum WriteStatus {
impl Connection {
/// Create a new connection with given id and socket.
pub fn new(token: Token, socket: TcpStream) -> Connection {
pub fn new(token: StreamToken, socket: TcpStream) -> Connection {
Connection {
token: token,
socket: socket,
send_queue: VecDeque::new(),
rec_buf: Bytes::new(),
rec_size: 0,
interest: EventSet::hup(),
interest: EventSet::hup() | EventSet::readable(),
}
}
@@ -86,7 +88,7 @@ impl Connection {
/// Add a packet to send queue.
pub fn send(&mut self, data: Bytes) {
if data.len() != 0 {
if !data.is_empty() {
self.send_queue.push_back(Cursor::new(data));
}
if !self.interest.is_writable() {
@@ -132,20 +134,19 @@ impl Connection {
}
/// Register this connection with the IO event loop.
pub fn register<Host: Handler>(&mut self, event_loop: &mut EventLoop<Host>) -> io::Result<()> {
trace!(target: "net", "connection register; token={:?}", self.token);
self.interest.insert(EventSet::readable());
event_loop.register(&self.socket, self.token, self.interest, PollOpt::edge() | PollOpt::oneshot()).or_else(|e| {
error!("Failed to register {:?}, {:?}", self.token, e);
pub fn register_socket<Host: Handler>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> io::Result<()> {
trace!(target: "net", "connection register; token={:?}", reg);
event_loop.register(&self.socket, reg, self.interest, PollOpt::edge() | PollOpt::oneshot()).or_else(|e| {
error!("Failed to register {:?}, {:?}", reg, e);
Err(e)
})
}
/// Update connection registration. Should be called at the end of the IO handler.
pub fn reregister<Host: Handler>(&mut self, event_loop: &mut EventLoop<Host>) -> io::Result<()> {
trace!(target: "net", "connection reregister; token={:?}", self.token);
event_loop.reregister( &self.socket, self.token, self.interest, PollOpt::edge() | PollOpt::oneshot()).or_else(|e| {
error!("Failed to reregister {:?}, {:?}", self.token, e);
pub fn update_socket<Host: Handler>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> io::Result<()> {
trace!(target: "net", "connection reregister; token={:?}", reg);
event_loop.reregister( &self.socket, reg, self.interest, PollOpt::edge() | PollOpt::oneshot()).or_else(|e| {
error!("Failed to reregister {:?}, {:?}", reg, e);
Err(e)
})
}
@@ -182,8 +183,6 @@ pub struct EncryptedConnection {
ingress_mac: Keccak,
/// Read state
read_state: EncryptedConnectionState,
/// Disconnect timeout
idle_timeout: Option<Timeout>,
/// Protocol id for the last received packet
protocol_id: u16,
/// Payload expected to be received for the last header.
@@ -192,7 +191,7 @@ pub struct EncryptedConnection {
impl EncryptedConnection {
/// Create an encrypted connection out of the handshake. Consumes a handshake object.
pub fn new(handshake: Handshake) -> Result<EncryptedConnection, UtilError> {
pub fn new(mut handshake: Handshake) -> Result<EncryptedConnection, UtilError> {
let shared = try!(crypto::ecdh::agree(handshake.ecdhe.secret(), &handshake.remote_public));
let mut nonce_material = H512::new();
if handshake.originated {
@@ -227,6 +226,7 @@ impl EncryptedConnection {
ingress_mac.update(&mac_material);
ingress_mac.update(if handshake.originated { &handshake.ack_cipher } else { &handshake.auth_cipher });
handshake.connection.expect(ENCRYPTED_HEADER_LEN);
Ok(EncryptedConnection {
connection: handshake.connection,
encoder: encoder,
@@ -235,7 +235,6 @@ impl EncryptedConnection {
egress_mac: egress_mac,
ingress_mac: ingress_mac,
read_state: EncryptedConnectionState::Header,
idle_timeout: None,
protocol_id: 0,
payload_len: 0
})
@@ -337,16 +336,14 @@ impl EncryptedConnection {
}
/// Readable IO handler. Tracker receive status and returns decoded packet if avaialable.
pub fn readable<Host:Handler>(&mut self, event_loop: &mut EventLoop<Host>) -> Result<Option<Packet>, UtilError> {
self.idle_timeout.map(|t| event_loop.clear_timeout(t));
pub fn readable<Message>(&mut self, io: &IoContext<Message>) -> Result<Option<Packet>, UtilError> where Message: Send + Clone{
io.clear_timer(self.connection.token).unwrap();
match self.read_state {
EncryptedConnectionState::Header => {
match try!(self.connection.readable()) {
Some(data) => {
try!(self.read_header(&data));
},
None => {}
};
if let Some(data) = try!(self.connection.readable()) {
try!(self.read_header(&data));
try!(io.register_timer(self.connection.token, RECIEVE_PAYLOAD_TIMEOUT));
}
Ok(None)
},
EncryptedConnectionState::Payload => {
@@ -363,24 +360,15 @@ impl EncryptedConnection {
}
/// Writable IO handler. Processes send queeue.
pub fn writable<Host:Handler>(&mut self, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
self.idle_timeout.map(|t| event_loop.clear_timeout(t));
pub fn writable<Message>(&mut self, io: &IoContext<Message>) -> Result<(), UtilError> where Message: Send + Clone {
io.clear_timer(self.connection.token).unwrap();
try!(self.connection.writable());
Ok(())
}
/// Register this connection with the event handler.
pub fn register<Host:Handler<Timeout=Token>>(&mut self, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
self.connection.expect(ENCRYPTED_HEADER_LEN);
self.idle_timeout.map(|t| event_loop.clear_timeout(t));
self.idle_timeout = event_loop.timeout_ms(self.connection.token, 1800).ok();
try!(self.connection.reregister(event_loop));
Ok(())
}
/// Update connection registration. This should be called at the end of the event loop.
pub fn reregister<Host:Handler>(&mut self, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
try!(self.connection.reregister(event_loop));
pub fn update_socket<Host:Handler>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
try!(self.connection.update_socket(reg, event_loop));
Ok(())
}
}

21
util/src/network/discovery.rs
View File

@@ -62,7 +62,7 @@ impl Discovery {
discovery_round: 0,
discovery_id: NodeId::new(),
discovery_nodes: HashSet::new(),
node_buckets: (0..NODE_BINS).map(|x| NodeBucket::new(x)).collect(),
node_buckets: (0..NODE_BINS).map(NodeBucket::new).collect(),
}
}
@@ -122,7 +122,8 @@ impl Discovery {
ret
}
fn nearest_node_entries<'b>(source: &NodeId, target: &NodeId, buckets: &'b Vec<NodeBucket>) -> Vec<&'b NodeId>
#[allow(cyclomatic_complexity)]
fn nearest_node_entries<'b>(source: &NodeId, target: &NodeId, buckets: &'b [NodeBucket]) -> Vec<&'b NodeId>
{
// send ALPHA FindNode packets to nodes we know, closest to target
const LAST_BIN: u32 = NODE_BINS - 1;
@@ -136,21 +137,21 @@ impl Discovery {
if head > 1 && tail != LAST_BIN {
while head != tail && head < NODE_BINS && count < BUCKET_SIZE
{
for n in buckets[head as usize].nodes.iter()
for n in &buckets[head as usize].nodes
{
if count < BUCKET_SIZE {
count += 1;
found.entry(Discovery::distance(target, &n)).or_insert(Vec::new()).push(n);
found.entry(Discovery::distance(target, &n)).or_insert_with(Vec::new).push(n);
}
else {
break;
}
}
if count < BUCKET_SIZE && tail != 0 {
for n in buckets[tail as usize].nodes.iter() {
for n in &buckets[tail as usize].nodes {
if count < BUCKET_SIZE {
count += 1;
found.entry(Discovery::distance(target, &n)).or_insert(Vec::new()).push(n);
found.entry(Discovery::distance(target, &n)).or_insert_with(Vec::new).push(n);
}
else {
break;
@@ -166,10 +167,10 @@ impl Discovery {
}
else if head < 2 {
while head < NODE_BINS && count < BUCKET_SIZE {
for n in buckets[head as usize].nodes.iter() {
for n in &buckets[head as usize].nodes {
if count < BUCKET_SIZE {
count += 1;
found.entry(Discovery::distance(target, &n)).or_insert(Vec::new()).push(n);
found.entry(Discovery::distance(target, &n)).or_insert_with(Vec::new).push(n);
}
else {
break;
@@ -180,10 +181,10 @@ impl Discovery {
}
else {
while tail > 0 && count < BUCKET_SIZE {
for n in buckets[tail as usize].nodes.iter() {
for n in &buckets[tail as usize].nodes {
if count < BUCKET_SIZE {
count += 1;
found.entry(Discovery::distance(target, &n)).or_insert(Vec::new()).push(n);
found.entry(Discovery::distance(target, &n)).or_insert_with(Vec::new).push(n);
}
else {
break;

6
util/src/network/error.rs
View File

@@ -19,11 +19,17 @@ pub enum DisconnectReason
}
#[derive(Debug)]
/// Network error.
pub enum NetworkError {
/// Authentication error.
Auth,
/// Unrecognised protocol.
BadProtocol,
/// Peer not found.
PeerNotFound,
/// Peer is diconnected.
Disconnect(DisconnectReason),
/// Socket IO error.
Io(IoError),
}

54
util/src/network/handshake.rs
View File

@@ -10,6 +10,7 @@ use network::host::{HostInfo};
use network::node::NodeId;
use error::*;
use network::error::NetworkError;
use io::{IoContext, StreamToken};
#[derive(PartialEq, Eq, Debug)]
enum HandshakeState {
@@ -33,8 +34,6 @@ pub struct Handshake {
state: HandshakeState,
/// Outgoing or incoming connection
pub originated: bool,
/// Disconnect timeout
idle_timeout: Option<Timeout>,
/// ECDH ephemeral
pub ecdhe: KeyPair,
/// Connection nonce
@@ -51,16 +50,16 @@ pub struct Handshake {
const AUTH_PACKET_SIZE: usize = 307;
const ACK_PACKET_SIZE: usize = 210;
const HANDSHAKE_TIMEOUT: u64 = 30000;
impl Handshake {
/// Create a new handshake object
pub fn new(token: Token, id: &NodeId, socket: TcpStream, nonce: &H256) -> Result<Handshake, UtilError> {
pub fn new(token: StreamToken, id: &NodeId, socket: TcpStream, nonce: &H256) -> Result<Handshake, UtilError> {
Ok(Handshake {
id: id.clone(),
connection: Connection::new(token, socket),
originated: false,
state: HandshakeState::New,
idle_timeout: None,
ecdhe: try!(KeyPair::create()),
nonce: nonce.clone(),
remote_public: Public::new(),
@@ -71,8 +70,9 @@ impl Handshake {
}
/// Start a handhsake
pub fn start(&mut self, host: &HostInfo, originated: bool) -> Result<(), UtilError> {
pub fn start<Message>(&mut self, io: &IoContext<Message>, host: &HostInfo, originated: bool) -> Result<(), UtilError> where Message: Send + Clone{
self.originated = originated;
io.register_timer(self.connection.token, HANDSHAKE_TIMEOUT).ok();
if originated {
try!(self.write_auth(host));
}
@@ -89,50 +89,48 @@ impl Handshake {
}
/// Readable IO handler. Drives the state change.
pub fn readable<Host:Handler>(&mut self, event_loop: &mut EventLoop<Host>, host: &HostInfo) -> Result<(), UtilError> {
self.idle_timeout.map(|t| event_loop.clear_timeout(t));
pub fn readable<Message>(&mut self, io: &IoContext<Message>, host: &HostInfo) -> Result<(), UtilError> where Message: Send + Clone {
io.clear_timer(self.connection.token).unwrap();
match self.state {
HandshakeState::ReadingAuth => {
match try!(self.connection.readable()) {
Some(data) => {
try!(self.read_auth(host, &data));
try!(self.write_ack());
},
None => {}
if let Some(data) = try!(self.connection.readable()) {
try!(self.read_auth(host, &data));
try!(self.write_ack());
};
},
HandshakeState::ReadingAck => {
match try!(self.connection.readable()) {
Some(data) => {
try!(self.read_ack(host, &data));
self.state = HandshakeState::StartSession;
},
None => {}
if let Some(data) = try!(self.connection.readable()) {
try!(self.read_ack(host, &data));
self.state = HandshakeState::StartSession;
};
},
HandshakeState::StartSession => {},
_ => { panic!("Unexpected state"); }
}
if self.state != HandshakeState::StartSession {
try!(self.connection.reregister(event_loop));
try!(io.update_registration(self.connection.token));
}
Ok(())
}
/// Writabe IO handler.
pub fn writable<Host:Handler>(&mut self, event_loop: &mut EventLoop<Host>, _host: &HostInfo) -> Result<(), UtilError> {
self.idle_timeout.map(|t| event_loop.clear_timeout(t));
pub fn writable<Message>(&mut self, io: &IoContext<Message>, _host: &HostInfo) -> Result<(), UtilError> where Message: Send + Clone {
io.clear_timer(self.connection.token).unwrap();
try!(self.connection.writable());
if self.state != HandshakeState::StartSession {
try!(self.connection.reregister(event_loop));
io.update_registration(self.connection.token).unwrap();
}
Ok(())
}
/// Register the IO handler with the event loop
pub fn register<Host:Handler<Timeout=Token>>(&mut self, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
self.idle_timeout.map(|t| event_loop.clear_timeout(t));
self.idle_timeout = event_loop.timeout_ms(self.connection.token, 1800).ok();
try!(self.connection.register(event_loop));
/// Register the socket with the event loop
pub fn register_socket<Host:Handler<Timeout=Token>>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
try!(self.connection.register_socket(reg, event_loop));
Ok(())
}
pub fn update_socket<Host:Handler<Timeout=Token>>(&self, reg: Token, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
try!(self.connection.update_socket(reg, event_loop));
Ok(())
}

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

@@ -1,8 +1,9 @@
use std::mem;
use std::net::{SocketAddr};
use std::collections::{HashMap};
use std::hash::{Hasher};
use std::str::{FromStr};
use std::sync::*;
use std::ops::*;
use mio::*;
use mio::tcp::*;
use mio::udp::*;
@@ -64,19 +65,25 @@ pub type PacketId = u8;
pub type ProtocolId = &'static str;
/// Messages used to communitate with the event loop from other threads.
pub enum NetworkIoMessage<Message> where Message: Send {
#[derive(Clone)]
pub enum NetworkIoMessage<Message> where Message: Send + Sync + Clone {
/// Register a new protocol handler.
AddHandler {
handler: Option<Box<NetworkProtocolHandler<Message>+Send>>,
/// Handler shared instance.
handler: Arc<NetworkProtocolHandler<Message> + Sync>,
/// Protocol Id.
protocol: ProtocolId,
/// Supported protocol versions.
versions: Vec<u8>,
},
/// Send data over the network.
Send {
peer: PeerId,
packet_id: PacketId,
/// Register a new protocol timer
AddTimer {
/// Protocol Id.
protocol: ProtocolId,
data: Vec<u8>,
/// Timer token.
token: TimerToken,
/// Timer delay in milliseconds.
delay: u64,
},
/// User message
User(Message),
@@ -104,46 +111,45 @@ impl Encodable for CapabilityInfo {
}
/// IO access point. This is passed to all IO handlers and provides an interface to the IO subsystem.
pub struct NetworkContext<'s, 'io, Message> where Message: Send + 'static, 'io: 's {
io: &'s mut IoContext<'io, NetworkIoMessage<Message>>,
pub struct NetworkContext<'s, Message> where Message: Send + Sync + Clone + 'static, 's {
io: &'s IoContext<NetworkIoMessage<Message>>,
protocol: ProtocolId,
connections: &'s mut Slab<ConnectionEntry>,
timers: &'s mut HashMap<TimerToken, ProtocolId>,
connections: Arc<RwLock<Slab<SharedConnectionEntry>>>,
session: Option<StreamToken>,
}
impl<'s, 'io, Message> NetworkContext<'s, 'io, Message> where Message: Send + 'static, {
impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone + 'static, {
/// Create a new network IO access point. Takes references to all the data that can be updated within the IO handler.
fn new(io: &'s mut IoContext<'io, NetworkIoMessage<Message>>,
fn new(io: &'s IoContext<NetworkIoMessage<Message>>,
protocol: ProtocolId,
session: Option<StreamToken>, connections: &'s mut Slab<ConnectionEntry>,
timers: &'s mut HashMap<TimerToken, ProtocolId>) -> NetworkContext<'s, 'io, Message> {
session: Option<StreamToken>, connections: Arc<RwLock<Slab<SharedConnectionEntry>>>) -> NetworkContext<'s, Message> {
NetworkContext {
io: io,
protocol: protocol,
session: session,
connections: connections,
timers: timers,
}
}
/// Send a packet over the network to another peer.
pub fn send(&mut self, peer: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), UtilError> {
match self.connections.get_mut(peer) {
Some(&mut ConnectionEntry::Session(ref mut s)) => {
s.send_packet(self.protocol, packet_id as u8, &data).unwrap_or_else(|e| {
warn!(target: "net", "Send error: {:?}", e);
}); //TODO: don't copy vector data
},
_ => {
warn!(target: "net", "Send: Peer does not exist");
pub fn send(&self, peer: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), UtilError> {
if let Some(connection) = self.connections.read().unwrap().get(peer).cloned() {
match *connection.lock().unwrap().deref_mut() {
ConnectionEntry::Session(ref mut s) => {
s.send_packet(self.protocol, packet_id as u8, &data).unwrap_or_else(|e| {
warn!(target: "net", "Send error: {:?}", e);
}); //TODO: don't copy vector data
},
_ => warn!(target: "net", "Send: Peer is not connected yet")
}
} else {
warn!(target: "net", "Send: Peer does not exist")
}
Ok(())
}
/// Respond to a current network message. Panics if no there is no packet in the context.
pub fn respond(&mut self, packet_id: PacketId, data: Vec<u8>) -> Result<(), UtilError> {
pub fn respond(&self, packet_id: PacketId, data: Vec<u8>) -> Result<(), UtilError> {
match self.session {
Some(session) => self.send(session, packet_id, data),
None => {
@@ -153,31 +159,28 @@ impl<'s, 'io, Message> NetworkContext<'s, 'io, Message> where Message: Send + 's
}
/// Disable current protocol capability for given peer. If no capabilities left peer gets disconnected.
pub fn disable_peer(&mut self, _peer: PeerId) {
pub fn disable_peer(&self, _peer: PeerId) {
//TODO: remove capability, disconnect if no capabilities left
}
/// Register a new IO timer. Returns a new timer token. 'NetworkProtocolHandler::timeout' will be called with the token.
pub fn register_timer(&mut self, ms: u64) -> Result<TimerToken, UtilError>{
match self.io.register_timer(ms) {
Ok(token) => {
self.timers.insert(token, self.protocol);
Ok(token)
},
e => e,
}
/// Register a new IO timer. 'IoHandler::timeout' will be called with the token.
pub fn register_timer(&self, token: TimerToken, ms: u64) -> Result<(), UtilError> {
self.io.message(NetworkIoMessage::AddTimer {
token: token,
delay: ms,
protocol: self.protocol,
});
Ok(())
}
/// Returns peer identification string
pub fn peer_info(&self, peer: PeerId) -> String {
match self.connections.get(peer) {
Some(&ConnectionEntry::Session(ref s)) => {
s.info.client_version.clone()
},
_ => {
"unknown".to_string()
if let Some(connection) = self.connections.read().unwrap().get(peer).cloned() {
if let ConnectionEntry::Session(ref s) = *connection.lock().unwrap().deref() {
return s.info.client_version.clone()
}
}
"unknown".to_owned()
}
}
@@ -213,7 +216,7 @@ impl HostInfo {
/// Increments and returns connection nonce.
pub fn next_nonce(&mut self) -> H256 {
self.nonce = self.nonce.sha3();
return self.nonce.clone();
self.nonce.clone()
}
}
@@ -222,66 +225,92 @@ enum ConnectionEntry {
Session(Session)
}
/// Root IO handler. Manages protocol handlers, IO timers and network connections.
pub struct Host<Message> where Message: Send {
pub info: HostInfo,
udp_socket: UdpSocket,
listener: TcpListener,
connections: Slab<ConnectionEntry>,
timers: HashMap<TimerToken, ProtocolId>,
nodes: HashMap<NodeId, Node>,
handlers: HashMap<ProtocolId, Box<NetworkProtocolHandler<Message>>>,
type SharedConnectionEntry = Arc<Mutex<ConnectionEntry>>;
#[derive(Copy, Clone)]
struct ProtocolTimer {
pub protocol: ProtocolId,
pub token: TimerToken, // Handler level token
}
impl<Message> Host<Message> where Message: Send {
/// Root IO handler. Manages protocol handlers, IO timers and network connections.
pub struct Host<Message> where Message: Send + Sync + Clone {
pub info: RwLock<HostInfo>,
udp_socket: Mutex<UdpSocket>,
tcp_listener: Mutex<TcpListener>,
connections: Arc<RwLock<Slab<SharedConnectionEntry>>>,
nodes: RwLock<HashMap<NodeId, Node>>,
handlers: RwLock<HashMap<ProtocolId, Arc<NetworkProtocolHandler<Message>>>>,
timers: RwLock<HashMap<TimerToken, ProtocolTimer>>,
timer_counter: RwLock<usize>,
}
impl<Message> Host<Message> where Message: Send + Sync + Clone {
pub fn new() -> Host<Message> {
let config = NetworkConfiguration::new();
let addr = config.listen_address;
// Setup the server socket
let listener = TcpListener::bind(&addr).unwrap();
let tcp_listener = TcpListener::bind(&addr).unwrap();
let udp_socket = UdpSocket::bound(&addr).unwrap();
Host::<Message> {
info: HostInfo {
let host = Host::<Message> {
info: RwLock::new(HostInfo {
keys: KeyPair::create().unwrap(),
config: config,
nonce: H256::random(),
protocol_version: 4,
client_version: "parity".to_string(),
client_version: "parity".to_owned(),
listen_port: 0,
capabilities: Vec::new(),
},
udp_socket: udp_socket,
listener: listener,
connections: Slab::new_starting_at(FIRST_CONNECTION, MAX_CONNECTIONS),
timers: HashMap::new(),
nodes: HashMap::new(),
handlers: HashMap::new(),
}
}),
udp_socket: Mutex::new(udp_socket),
tcp_listener: Mutex::new(tcp_listener),
connections: Arc::new(RwLock::new(Slab::new_starting_at(FIRST_CONNECTION, MAX_CONNECTIONS))),
nodes: RwLock::new(HashMap::new()),
handlers: RwLock::new(HashMap::new()),
timers: RwLock::new(HashMap::new()),
timer_counter: RwLock::new(LAST_CONNECTION + 1),
};
let port = host.info.read().unwrap().config.listen_address.port();
host.info.write().unwrap().deref_mut().listen_port = port;
/*
match ::ifaces::Interface::get_all().unwrap().into_iter().filter(|x| x.kind == ::ifaces::Kind::Packet && x.addr.is_some()).next() {
Some(iface) => config.public_address = iface.addr.unwrap(),
None => warn!("No public network interface"),
*/
host
}
pub fn add_node(&mut self, id: &str) {
match Node::from_str(id) {
Err(e) => { warn!("Could not add node: {:?}", e); },
Ok(n) => {
self.nodes.insert(n.id.clone(), n);
self.nodes.write().unwrap().insert(n.id.clone(), n);
}
}
}
fn maintain_network(&mut self, io: &mut IoContext<NetworkIoMessage<Message>>) {
pub fn client_version(&self) -> String {
self.info.read().unwrap().client_version.clone()
}
pub fn client_id(&self) -> NodeId {
self.info.read().unwrap().id().clone()
}
fn maintain_network(&self, io: &IoContext<NetworkIoMessage<Message>>) {
self.connect_peers(io);
io.event_loop.timeout_ms(Token(IDLE), MAINTENANCE_TIMEOUT).unwrap();
}
fn have_session(&self, id: &NodeId) -> bool {
self.connections.iter().any(|e| match e { &ConnectionEntry::Session(ref s) => s.info.id.eq(&id), _ => false })
self.connections.read().unwrap().iter().any(|e| match *e.lock().unwrap().deref() { ConnectionEntry::Session(ref s) => s.info.id.eq(&id), _ => false })
}
fn connecting_to(&self, id: &NodeId) -> bool {
self.connections.iter().any(|e| match e { &ConnectionEntry::Handshake(ref h) => h.id.eq(&id), _ => false })
self.connections.read().unwrap().iter().any(|e| match *e.lock().unwrap().deref() { ConnectionEntry::Handshake(ref h) => h.id.eq(&id), _ => false })
}
fn connect_peers(&mut self, io: &mut IoContext<NetworkIoMessage<Message>>) {
fn connect_peers(&self, io: &IoContext<NetworkIoMessage<Message>>) {
struct NodeInfo {
id: NodeId,
peer_type: PeerType
@@ -292,18 +321,19 @@ impl<Message> Host<Message> where Message: Send {
let mut req_conn = 0;
//TODO: use nodes from discovery here
//for n in self.node_buckets.iter().flat_map(|n| &n.nodes).map(|id| NodeInfo { id: id.clone(), peer_type: self.nodes.get(id).unwrap().peer_type}) {
for n in self.nodes.values().map(|n| NodeInfo { id: n.id.clone(), peer_type: n.peer_type }) {
let pin = self.info.read().unwrap().deref().config.pin;
for n in self.nodes.read().unwrap().values().map(|n| NodeInfo { id: n.id.clone(), peer_type: n.peer_type }) {
let connected = self.have_session(&n.id) || self.connecting_to(&n.id);
let required = n.peer_type == PeerType::Required;
if connected && required {
req_conn += 1;
}
else if !connected && (!self.info.config.pin || required) {
else if !connected && (!pin || required) {
to_connect.push(n);
}
}
for n in to_connect.iter() {
for n in &to_connect {
if n.peer_type == PeerType::Required {
if req_conn < IDEAL_PEERS {
self.connect_peer(&n.id, io);
@@ -312,13 +342,12 @@ impl<Message> Host<Message> where Message: Send {
}
}
if !self.info.config.pin
{
if !pin {
let pending_count = 0; //TODO:
let peer_count = 0;
let mut open_slots = IDEAL_PEERS - peer_count - pending_count + req_conn;
if open_slots > 0 {
for n in to_connect.iter() {
for n in &to_connect {
if n.peer_type == PeerType::Optional && open_slots > 0 {
open_slots -= 1;
self.connect_peer(&n.id, io);
@@ -328,23 +357,27 @@ impl<Message> Host<Message> where Message: Send {
}
}
fn connect_peer(&mut self, id: &NodeId, io: &mut IoContext<NetworkIoMessage<Message>>) {
#[allow(single_match)]
#[allow(block_in_if_condition_stmt)]
fn connect_peer(&self, id: &NodeId, io: &IoContext<NetworkIoMessage<Message>>) {
if self.have_session(id)
{
warn!("Aborted connect. Node already connected.");
return;
}
if self.connecting_to(id)
{
if self.connecting_to(id) {
warn!("Aborted connect. Node already connecting.");
return;
}
let socket = {
let node = self.nodes.get_mut(id).unwrap();
node.last_attempted = Some(::time::now());
match TcpStream::connect(&node.endpoint.address) {
let address = {
let mut nodes = self.nodes.write().unwrap();
let node = nodes.get_mut(id).unwrap();
node.last_attempted = Some(::time::now());
node.endpoint.address
};
match TcpStream::connect(&address) {
Ok(socket) => socket,
Err(_) => {
warn!("Cannot connect to node");
@@ -353,216 +386,182 @@ impl<Message> Host<Message> where Message: Send {
}
};
let nonce = self.info.next_nonce();
match self.connections.insert_with(|token| ConnectionEntry::Handshake(Handshake::new(Token(token), id, socket, &nonce).expect("Can't create handshake"))) {
Some(token) => {
match self.connections.get_mut(token) {
Some(&mut ConnectionEntry::Handshake(ref mut h)) => {
h.start(&self.info, true)
.and_then(|_| h.register(io.event_loop))
.unwrap_or_else (|e| {
debug!(target: "net", "Handshake create error: {:?}", e);
});
},
_ => {}
}
},
None => { warn!("Max connections reached") }
let nonce = self.info.write().unwrap().next_nonce();
if self.connections.write().unwrap().insert_with(|token| {
let mut handshake = Handshake::new(token, id, socket, &nonce).expect("Can't create handshake");
handshake.start(io, &self.info.read().unwrap(), true).and_then(|_| io.register_stream(token)).unwrap_or_else (|e| {
debug!(target: "net", "Handshake create error: {:?}", e);
});
Arc::new(Mutex::new(ConnectionEntry::Handshake(handshake)))
}).is_none() {
warn!("Max connections reached");
}
}
fn accept(&mut self, _io: &mut IoContext<NetworkIoMessage<Message>>) {
fn accept(&self, _io: &IoContext<NetworkIoMessage<Message>>) {
trace!(target: "net", "accept");
}
fn connection_writable<'s>(&'s mut self, token: StreamToken, io: &mut IoContext<'s, NetworkIoMessage<Message>>) {
let mut kill = false;
#[allow(single_match)]
fn connection_writable(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) {
let mut create_session = false;
match self.connections.get_mut(token) {
Some(&mut ConnectionEntry::Handshake(ref mut h)) => {
h.writable(io.event_loop, &self.info).unwrap_or_else(|e| {
debug!(target: "net", "Handshake write error: {:?}", e);
kill = true;
});
create_session = h.done();
},
Some(&mut ConnectionEntry::Session(ref mut s)) => {
s.writable(io.event_loop, &self.info).unwrap_or_else(|e| {
debug!(target: "net", "Session write error: {:?}", e);
kill = true;
});
let mut kill = false;
if let Some(connection) = self.connections.read().unwrap().get(token).cloned() {
match *connection.lock().unwrap().deref_mut() {
ConnectionEntry::Handshake(ref mut h) => {
match h.writable(io, &self.info.read().unwrap()) {
Err(e) => {
debug!(target: "net", "Handshake write error: {:?}", e);
kill = true;
},
Ok(_) => ()
}
if h.done() {
create_session = true;
}
},
ConnectionEntry::Session(ref mut s) => {
match s.writable(io, &self.info.read().unwrap()) {
Err(e) => {
debug!(target: "net", "Session write error: {:?}", e);
kill = true;
},
Ok(_) => ()
}
io.update_registration(token).unwrap_or_else(|e| debug!(target: "net", "Session registration error: {:?}", e));
}
}
_ => {
warn!(target: "net", "Received event for unknown connection");
}
}
}
if kill {
self.kill_connection(token, io);
self.kill_connection(token, io); //TODO: mark connection as dead an check in kill_connection
return;
} else if create_session {
self.start_session(token, io);
}
match self.connections.get_mut(token) {
Some(&mut ConnectionEntry::Session(ref mut s)) => {
s.reregister(io.event_loop).unwrap_or_else(|e| debug!(target: "net", "Session registration error: {:?}", e));
},
_ => (),
io.update_registration(token).unwrap_or_else(|e| debug!(target: "net", "Session registration error: {:?}", e));
}
}
fn connection_closed<'s>(&'s mut self, token: TimerToken, io: &mut IoContext<'s, NetworkIoMessage<Message>>) {
fn connection_closed(&self, token: TimerToken, io: &IoContext<NetworkIoMessage<Message>>) {
self.kill_connection(token, io);
}
fn connection_readable<'s>(&'s mut self, token: StreamToken, io: &mut IoContext<'s, NetworkIoMessage<Message>>) {
let mut kill = false;
let mut create_session = false;
fn connection_readable(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) {
let mut ready_data: Vec<ProtocolId> = Vec::new();
let mut packet_data: Option<(ProtocolId, PacketId, Vec<u8>)> = None;
match self.connections.get_mut(token) {
Some(&mut ConnectionEntry::Handshake(ref mut h)) => {
h.readable(io.event_loop, &self.info).unwrap_or_else(|e| {
debug!(target: "net", "Handshake read error: {:?}", e);
kill = true;
});
create_session = h.done();
},
Some(&mut ConnectionEntry::Session(ref mut s)) => {
let sd = { s.readable(io.event_loop, &self.info).unwrap_or_else(|e| {
debug!(target: "net", "Session read error: {:?}", e);
kill = true;
SessionData::None
}) };
match sd {
SessionData::Ready => {
for (p, _) in self.handlers.iter_mut() {
if s.have_capability(p) {
ready_data.push(p);
let mut create_session = false;
let mut kill = false;
if let Some(connection) = self.connections.read().unwrap().get(token).cloned() {
match *connection.lock().unwrap().deref_mut() {
ConnectionEntry::Handshake(ref mut h) => {
if let Err(e) = h.readable(io, &self.info.read().unwrap()) {
debug!(target: "net", "Handshake read error: {:?}", e);
kill = true;
}
if h.done() {
create_session = true;
}
},
ConnectionEntry::Session(ref mut s) => {
match s.readable(io, &self.info.read().unwrap()) {
Err(e) => {
debug!(target: "net", "Handshake read error: {:?}", e);
kill = true;
},
Ok(SessionData::Ready) => {
for (p, _) in self.handlers.read().unwrap().iter() {
if s.have_capability(p) {
ready_data.push(p);
}
}
}
},
SessionData::Packet {
data,
protocol,
packet_id,
} => {
match self.handlers.get_mut(protocol) {
None => { warn!(target: "net", "No handler found for protocol: {:?}", protocol) },
Some(_) => packet_data = Some((protocol, packet_id, data)),
}
},
SessionData::None => {},
}
}
_ => {
warn!(target: "net", "Received event for unknown connection");
}
}
if kill {
self.kill_connection(token, io);
return;
}
if create_session {
self.start_session(token, io);
}
for p in ready_data {
let mut h = self.handlers.get_mut(p).unwrap();
h.connected(&mut NetworkContext::new(io, p, Some(token), &mut self.connections, &mut self.timers), &token);
}
if let Some((p, packet_id, data)) = packet_data {
let mut h = self.handlers.get_mut(p).unwrap();
h.read(&mut NetworkContext::new(io, p, Some(token), &mut self.connections, &mut self.timers), &token, packet_id, &data[1..]);
}
match self.connections.get_mut(token) {
Some(&mut ConnectionEntry::Session(ref mut s)) => {
s.reregister(io.event_loop).unwrap_or_else(|e| debug!(target: "net", "Session registration error: {:?}", e));
},
_ => (),
}
}
fn start_session(&mut self, token: StreamToken, io: &mut IoContext<NetworkIoMessage<Message>>) {
let info = &self.info;
// TODO: use slab::replace_with (currently broken)
/*
match self.connections.remove(token) {
Some(ConnectionEntry::Handshake(h)) => {
match Session::new(h, io.event_loop, info) {
Ok(session) => {
assert!(token == self.connections.insert(ConnectionEntry::Session(session)).ok().unwrap());
},
Err(e) => {
debug!(target: "net", "Session construction error: {:?}", e);
},
Ok(SessionData::Packet {
data,
protocol,
packet_id,
}) => {
match self.handlers.read().unwrap().get(protocol) {
None => { warn!(target: "net", "No handler found for protocol: {:?}", protocol) },
Some(_) => packet_data = Some((protocol, packet_id, data)),
}
},
Ok(SessionData::None) => {},
}
}
},
_ => panic!("Error updating slab with session")
}*/
self.connections.replace_with(token, |c| {
match c {
ConnectionEntry::Handshake(h) => Session::new(h, io.event_loop, info)
.map(|s| Some(ConnectionEntry::Session(s)))
.unwrap_or_else(|e| {
debug!(target: "net", "Session construction error: {:?}", e);
None
}),
_ => { panic!("No handshake to create a session from"); }
}
}).expect("Error updating slab with session");
}
if kill {
self.kill_connection(token, io); //TODO: mark connection as dead an check in kill_connection
return;
} else if create_session {
self.start_session(token, io);
io.update_registration(token).unwrap_or_else(|e| debug!(target: "net", "Session registration error: {:?}", e));
}
for p in ready_data {
let h = self.handlers.read().unwrap().get(p).unwrap().clone();
h.connected(&NetworkContext::new(io, p, Some(token), self.connections.clone()), &token);
}
if let Some((p, packet_id, data)) = packet_data {
let h = self.handlers.read().unwrap().get(p).unwrap().clone();
h.read(&NetworkContext::new(io, p, Some(token), self.connections.clone()), &token, packet_id, &data[1..]);
}
io.update_registration(token).unwrap_or_else(|e| debug!(target: "net", "Token registration error: {:?}", e));
}
fn connection_timeout<'s>(&'s mut self, token: StreamToken, io: &mut IoContext<'s, NetworkIoMessage<Message>>) {
fn start_session(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) {
self.connections.write().unwrap().replace_with(token, |c| {
match Arc::try_unwrap(c).ok().unwrap().into_inner().unwrap() {
ConnectionEntry::Handshake(h) => {
let session = Session::new(h, io, &self.info.read().unwrap()).expect("Session creation error");
io.update_registration(token).expect("Error updating session registration");
Some(Arc::new(Mutex::new(ConnectionEntry::Session(session))))
},
_ => { None } // handshake expired
}
}).ok();
}
fn connection_timeout(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) {
self.kill_connection(token, io)
}
fn kill_connection<'s>(&'s mut self, token: StreamToken, io: &mut IoContext<'s, NetworkIoMessage<Message>>) {
fn kill_connection(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) {
let mut to_disconnect: Vec<ProtocolId> = Vec::new();
let mut remove = true;
match self.connections.get_mut(token) {
Some(&mut ConnectionEntry::Handshake(_)) => (), // just abandon handshake
Some(&mut ConnectionEntry::Session(ref mut s)) if s.is_ready() => {
for (p, _) in self.handlers.iter_mut() {
if s.have_capability(p) {
to_disconnect.push(p);
}
{
let mut connections = self.connections.write().unwrap();
if let Some(connection) = connections.get(token).cloned() {
match *connection.lock().unwrap().deref_mut() {
ConnectionEntry::Handshake(_) => {
connections.remove(token);
},
ConnectionEntry::Session(ref mut s) if s.is_ready() => {
for (p, _) in self.handlers.read().unwrap().iter() {
if s.have_capability(p) {
to_disconnect.push(p);
}
}
connections.remove(token);
},
_ => {},
}
},
_ => {
remove = false;
},
}
}
for p in to_disconnect {
let mut h = self.handlers.get_mut(p).unwrap();
h.disconnected(&mut NetworkContext::new(io, p, Some(token), &mut self.connections, &mut self.timers), &token);
}
if remove {
self.connections.remove(token);
let h = self.handlers.read().unwrap().get(p).unwrap().clone();
h.disconnected(&NetworkContext::new(io, p, Some(token), self.connections.clone()), &token);
}
}
}
impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Message: Send + 'static {
impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Message: Send + Sync + Clone + 'static {
/// Initialize networking
fn initialize(&mut self, io: &mut IoContext<NetworkIoMessage<Message>>) {
/*
match ::ifaces::Interface::get_all().unwrap().into_iter().filter(|x| x.kind == ::ifaces::Kind::Packet && x.addr.is_some()).next() {
Some(iface) => config.public_address = iface.addr.unwrap(),
None => warn!("No public network interface"),
*/
// Start listening for incoming connections
io.event_loop.register(&self.listener, Token(TCP_ACCEPT), EventSet::readable(), PollOpt::edge()).unwrap();
io.event_loop.timeout_ms(Token(IDLE), MAINTENANCE_TIMEOUT).unwrap();
// open the udp socket
io.event_loop.register(&self.udp_socket, Token(NODETABLE_RECEIVE), EventSet::readable(), PollOpt::edge()).unwrap();
io.event_loop.timeout_ms(Token(NODETABLE_MAINTAIN), 7200).unwrap();
let port = self.info.config.listen_address.port();
self.info.listen_port = port;
fn initialize(&self, io: &IoContext<NetworkIoMessage<Message>>) {
io.register_stream(TCP_ACCEPT).expect("Error registering TCP listener");
io.register_stream(NODETABLE_RECEIVE).expect("Error registering UDP listener");
io.register_timer(IDLE, MAINTENANCE_TIMEOUT).expect("Error registering Network idle timer");
//io.register_timer(NODETABLE_MAINTAIN, 7200);
}
fn stream_hup<'s>(&'s mut self, io: &mut IoContext<'s, NetworkIoMessage<Message>>, stream: StreamToken) {
fn stream_hup(&self, io: &IoContext<NetworkIoMessage<Message>>, stream: StreamToken) {
trace!(target: "net", "Hup: {}", stream);
match stream {
FIRST_CONNECTION ... LAST_CONNECTION => self.connection_closed(stream, io),
@@ -570,7 +569,7 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
};
}
fn stream_readable<'s>(&'s mut self, io: &mut IoContext<'s, NetworkIoMessage<Message>>, stream: StreamToken) {
fn stream_readable(&self, io: &IoContext<NetworkIoMessage<Message>>, stream: StreamToken) {
match stream {
FIRST_CONNECTION ... LAST_CONNECTION => self.connection_readable(stream, io),
NODETABLE_RECEIVE => {},
@@ -579,65 +578,97 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
}
}
fn stream_writable<'s>(&'s mut self, io: &mut IoContext<'s, NetworkIoMessage<Message>>, stream: StreamToken) {
fn stream_writable(&self, io: &IoContext<NetworkIoMessage<Message>>, stream: StreamToken) {
match stream {
FIRST_CONNECTION ... LAST_CONNECTION => self.connection_writable(stream, io),
NODETABLE_RECEIVE => {},
_ => panic!("Received unknown writable token"),
}
}
fn timeout<'s>(&'s mut self, io: &mut IoContext<'s, NetworkIoMessage<Message>>, token: TimerToken) {
fn timeout(&self, io: &IoContext<NetworkIoMessage<Message>>, token: TimerToken) {
match token {
IDLE => self.maintain_network(io),
FIRST_CONNECTION ... LAST_CONNECTION => self.connection_timeout(token, io),
NODETABLE_DISCOVERY => {},
NODETABLE_MAINTAIN => {},
_ => match self.timers.get_mut(&token).map(|p| *p) {
Some(protocol) => match self.handlers.get_mut(protocol) {
None => { warn!(target: "net", "No handler found for protocol: {:?}", protocol) },
Some(h) => { h.timeout(&mut NetworkContext::new(io, protocol, Some(token), &mut self.connections, &mut self.timers), token); }
},
None => {} // time not registerd through us
_ => match self.timers.read().unwrap().get(&token).cloned() {
Some(timer) => match self.handlers.read().unwrap().get(timer.protocol).cloned() {
None => { warn!(target: "net", "No handler found for protocol: {:?}", timer.protocol) },
Some(h) => { h.timeout(&NetworkContext::new(io, timer.protocol, None, self.connections.clone()), timer.token); }
},
None => { warn!("Unknown timer token: {}", token); } // timer is not registerd through us
}
}
}
fn message<'s>(&'s mut self, io: &mut IoContext<'s, NetworkIoMessage<Message>>, message: &'s mut NetworkIoMessage<Message>) {
match message {
&mut NetworkIoMessage::AddHandler {
ref mut handler,
fn message(&self, io: &IoContext<NetworkIoMessage<Message>>, message: &NetworkIoMessage<Message>) {
match *message {
NetworkIoMessage::AddHandler {
ref handler,
ref protocol,
ref versions
} => {
let mut h = mem::replace(handler, None).unwrap();
h.initialize(&mut NetworkContext::new(io, protocol, None, &mut self.connections, &mut self.timers));
self.handlers.insert(protocol, h);
let h = handler.clone();
h.initialize(&NetworkContext::new(io, protocol, None, self.connections.clone()));
self.handlers.write().unwrap().insert(protocol, h);
let mut info = self.info.write().unwrap();
for v in versions {
self.info.capabilities.push(CapabilityInfo { protocol: protocol, version: *v, packet_count:0 });
info.capabilities.push(CapabilityInfo { protocol: protocol, version: *v, packet_count:0 });
}
},
&mut NetworkIoMessage::Send {
ref peer,
ref packet_id,
NetworkIoMessage::AddTimer {
ref protocol,
ref data,
ref delay,
ref token,
} => {
match self.connections.get_mut(*peer as usize) {
Some(&mut ConnectionEntry::Session(ref mut s)) => {
s.send_packet(protocol, *packet_id as u8, &data).unwrap_or_else(|e| {
warn!(target: "net", "Send error: {:?}", e);
}); //TODO: don't copy vector data
},
_ => {
warn!(target: "net", "Send: Peer does not exist");
}
}
let handler_token = {
let mut timer_counter = self.timer_counter.write().unwrap();
let counter = timer_counter.deref_mut();
let handler_token = *counter;
*counter += 1;
handler_token
};
self.timers.write().unwrap().insert(handler_token, ProtocolTimer { protocol: protocol, token: *token });
io.register_timer(handler_token, *delay).expect("Error registering timer");
},
&mut NetworkIoMessage::User(ref message) => {
for (p, h) in self.handlers.iter_mut() {
h.message(&mut NetworkContext::new(io, p, None, &mut self.connections, &mut self.timers), &message);
NetworkIoMessage::User(ref message) => {
for (p, h) in self.handlers.read().unwrap().iter() {
h.message(&NetworkContext::new(io, p, None, self.connections.clone()), &message);
}
}
}
}
fn register_stream(&self, stream: StreamToken, reg: Token, event_loop: &mut EventLoop<IoManager<NetworkIoMessage<Message>>>) {
match stream {
FIRST_CONNECTION ... LAST_CONNECTION => {
if let Some(connection) = self.connections.read().unwrap().get(stream).cloned() {
match *connection.lock().unwrap().deref() {
ConnectionEntry::Handshake(ref h) => h.register_socket(reg, event_loop).expect("Error registering socket"),
ConnectionEntry::Session(_) => warn!("Unexpected session stream registration")
}
} else {} // expired
}
NODETABLE_RECEIVE => event_loop.register(self.udp_socket.lock().unwrap().deref(), Token(NODETABLE_RECEIVE), EventSet::all(), PollOpt::edge()).expect("Error registering stream"),
TCP_ACCEPT => event_loop.register(self.tcp_listener.lock().unwrap().deref(), Token(TCP_ACCEPT), EventSet::all(), PollOpt::edge()).expect("Error registering stream"),
_ => warn!("Unexpected stream registration")
}
}
fn update_stream(&self, stream: StreamToken, reg: Token, event_loop: &mut EventLoop<IoManager<NetworkIoMessage<Message>>>) {
match stream {
FIRST_CONNECTION ... LAST_CONNECTION => {
if let Some(connection) = self.connections.read().unwrap().get(stream).cloned() {
match *connection.lock().unwrap().deref() {
ConnectionEntry::Handshake(ref h) => h.update_socket(reg, event_loop).expect("Error updating socket"),
ConnectionEntry::Session(ref s) => s.update_socket(reg, event_loop).expect("Error updating socket"),
}
} else {} // expired
}
NODETABLE_RECEIVE => event_loop.reregister(self.udp_socket.lock().unwrap().deref(), Token(NODETABLE_RECEIVE), EventSet::all(), PollOpt::edge()).expect("Error reregistering stream"),
TCP_ACCEPT => event_loop.reregister(self.tcp_listener.lock().unwrap().deref(), Token(TCP_ACCEPT), EventSet::all(), PollOpt::edge()).expect("Error reregistering stream"),
_ => warn!("Unexpected stream update")
}
}
}

87
util/src/network/mod.rs
View File

@@ -8,39 +8,40 @@
///
/// struct MyHandler;
///
/// #[derive(Clone)]
/// struct MyMessage {
/// data: u32
/// }
///
/// impl NetworkProtocolHandler<MyMessage> for MyHandler {
/// fn initialize(&mut self, io: &mut NetworkContext<MyMessage>) {
/// io.register_timer(1000);
/// fn initialize(&self, io: &NetworkContext<MyMessage>) {
/// io.register_timer(0, 1000);
/// }
///
/// fn read(&mut self, io: &mut NetworkContext<MyMessage>, peer: &PeerId, packet_id: u8, data: &[u8]) {
/// fn read(&self, io: &NetworkContext<MyMessage>, peer: &PeerId, packet_id: u8, data: &[u8]) {
/// println!("Received {} ({} bytes) from {}", packet_id, data.len(), peer);
/// }
///
/// fn connected(&mut self, io: &mut NetworkContext<MyMessage>, peer: &PeerId) {
/// fn connected(&self, io: &NetworkContext<MyMessage>, peer: &PeerId) {
/// println!("Connected {}", peer);
/// }
///
/// fn disconnected(&mut self, io: &mut NetworkContext<MyMessage>, peer: &PeerId) {
/// fn disconnected(&self, io: &NetworkContext<MyMessage>, peer: &PeerId) {
/// println!("Disconnected {}", peer);
/// }
///
/// fn timeout(&mut self, io: &mut NetworkContext<MyMessage>, timer: TimerToken) {
/// fn timeout(&self, io: &NetworkContext<MyMessage>, timer: TimerToken) {
/// println!("Timeout {}", timer);
/// }
///
/// fn message(&mut self, io: &mut NetworkContext<MyMessage>, message: &MyMessage) {
/// fn message(&self, io: &NetworkContext<MyMessage>, message: &MyMessage) {
/// println!("Message {}", message.data);
/// }
/// }
///
/// fn main () {
/// let mut service = NetworkService::<MyMessage>::start().expect("Error creating network service");
/// service.register_protocol(Box::new(MyHandler), "myproto", &[1u8]);
/// service.register_protocol(Arc::new(MyHandler), "myproto", &[1u8]);
///
/// // Wait for quit condition
/// // ...
@@ -57,36 +58,78 @@ mod error;
mod node;
/// TODO [arkpar] Please document me
pub type PeerId = host::PeerId;
pub use network::host::PeerId;
/// TODO [arkpar] Please document me
pub type PacketId = host::PacketId;
pub use network::host::PacketId;
/// TODO [arkpar] Please document me
pub type NetworkContext<'s,'io, Message> = host::NetworkContext<'s, 'io, Message>;
pub use network::host::NetworkContext;
/// TODO [arkpar] Please document me
pub type NetworkService<Message> = service::NetworkService<Message>;
pub use network::service::NetworkService;
/// TODO [arkpar] Please document me
pub use network::host::NetworkIoMessage;
/// TODO [arkpar] Please document me
pub type NetworkIoMessage<Message> = host::NetworkIoMessage<Message>;
pub use network::host::NetworkIoMessage::User as UserMessage;
/// TODO [arkpar] Please document me
pub type NetworkError = error::NetworkError;
pub use network::error::NetworkError;
use io::*;
use io::TimerToken;
/// Network IO protocol handler. This needs to be implemented for each new subprotocol.
/// All the handler function are called from within IO event loop.
/// `Message` is the type for message data.
pub trait NetworkProtocolHandler<Message>: Send where Message: Send {
pub trait NetworkProtocolHandler<Message>: Sync + Send where Message: Send + Sync + Clone {
/// Initialize the handler
fn initialize(&mut self, _io: &mut NetworkContext<Message>) {}
fn initialize(&self, _io: &NetworkContext<Message>) {}
/// Called when new network packet received.
fn read(&mut self, io: &mut NetworkContext<Message>, peer: &PeerId, packet_id: u8, data: &[u8]);
fn read(&self, io: &NetworkContext<Message>, peer: &PeerId, packet_id: u8, data: &[u8]);
/// Called when new peer is connected. Only called when peer supports the same protocol.
fn connected(&mut self, io: &mut NetworkContext<Message>, peer: &PeerId);
fn connected(&self, io: &NetworkContext<Message>, peer: &PeerId);
/// Called when a previously connected peer disconnects.
fn disconnected(&mut self, io: &mut NetworkContext<Message>, peer: &PeerId);
fn disconnected(&self, io: &NetworkContext<Message>, peer: &PeerId);
/// Timer function called after a timeout created with `NetworkContext::timeout`.
fn timeout(&mut self, _io: &mut NetworkContext<Message>, _timer: TimerToken) {}
fn timeout(&self, _io: &NetworkContext<Message>, _timer: TimerToken) {}
/// Called when a broadcasted message is received. The message can only be sent from a different IO handler.
fn message(&mut self, _io: &mut NetworkContext<Message>, _message: &Message) {}
fn message(&self, _io: &NetworkContext<Message>, _message: &Message) {}
}
#[test]
fn test_net_service() {
use std::sync::Arc;
struct MyHandler;
#[derive(Clone)]
struct MyMessage {
data: u32
}
impl NetworkProtocolHandler<MyMessage> for MyHandler {
fn initialize(&self, io: &NetworkContext<MyMessage>) {
io.register_timer(0, 1000).unwrap();
}
fn read(&self, _io: &NetworkContext<MyMessage>, peer: &PeerId, packet_id: u8, data: &[u8]) {
println!("Received {} ({} bytes) from {}", packet_id, data.len(), peer);
}
fn connected(&self, _io: &NetworkContext<MyMessage>, peer: &PeerId) {
println!("Connected {}", peer);
}
fn disconnected(&self, _io: &NetworkContext<MyMessage>, peer: &PeerId) {
println!("Disconnected {}", peer);
}
fn timeout(&self, _io: &NetworkContext<MyMessage>, timer: TimerToken) {
println!("Timeout {}", timer);
}
fn message(&self, _io: &NetworkContext<MyMessage>, message: &MyMessage) {
println!("Message {}", message.data);
}
}
let mut service = NetworkService::<MyMessage>::start().expect("Error creating network service");
service.register_protocol(Arc::new(MyHandler), "myproto", &[1u8]).unwrap();
}

6
util/src/network/node.rs
View File

@@ -20,14 +20,16 @@ pub struct NodeEndpoint {
pub udp_port: u16
}
impl NodeEndpoint {
impl FromStr for NodeEndpoint {
type Err = UtilError;
/// Create endpoint from string. Performs name resolution if given a host name.
fn from_str(s: &str) -> Result<NodeEndpoint, UtilError> {
let address = s.to_socket_addrs().map(|mut i| i.next());
match address {
Ok(Some(a)) => Ok(NodeEndpoint {
address: a,
address_str: s.to_string(),
address_str: s.to_owned(),
udp_port: a.port()
}),
Ok(_) => Err(UtilError::AddressResolve(None)),

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

@@ -1,24 +1,26 @@
use std::sync::*;
use error::*;
use network::{NetworkProtocolHandler};
use network::error::{NetworkError};
use network::host::{Host, NetworkIoMessage, PeerId, PacketId, ProtocolId};
use network::host::{Host, NetworkIoMessage, ProtocolId};
use io::*;
/// IO Service with networking
/// `Message` defines a notification data type.
pub struct NetworkService<Message> where Message: Send + 'static {
pub struct NetworkService<Message> where Message: Send + Sync + Clone + 'static {
io_service: IoService<NetworkIoMessage<Message>>,
host_info: String,
}
impl<Message> NetworkService<Message> where Message: Send + 'static {
impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'static {
/// Starts IO event loop
pub fn start(init_nodes: &Vec<String>) -> Result<NetworkService<Message>, UtilError> {
pub fn start(init_nodes: &[String]) -> Result<NetworkService<Message>, UtilError> {
let mut io_service = try!(IoService::<NetworkIoMessage<Message>>::start());
let mut host = Box::new(Host::new());
let mut host = Host::new();
for n in init_nodes { host.add_node(&n); }
let host_info = host.info.client_version.clone();
info!("NetworkService::start(): id={:?}", host.info.id());
let host = Arc::new(host);
let host_info = host.client_version();
info!("NetworkService::start(): id={:?}", host.client_id());
try!(io_service.register_handler(host));
Ok(NetworkService {
io_service: io_service,
@@ -26,21 +28,10 @@ impl<Message> NetworkService<Message> where Message: Send + 'static {
})
}
/// Send a message over the network. Normaly `HostIo::send` should be used. This can be used from non-io threads.
pub fn send(&mut self, peer: &PeerId, packet_id: PacketId, protocol: ProtocolId, data: &[u8]) -> Result<(), NetworkError> {
try!(self.io_service.send_message(NetworkIoMessage::Send {
peer: *peer,
packet_id: packet_id,
protocol: protocol,
data: data.to_vec()
}));
Ok(())
}
/// Regiter a new protocol handler with the event loop.
pub fn register_protocol(&mut self, handler: Box<NetworkProtocolHandler<Message>+Send>, protocol: ProtocolId, versions: &[u8]) -> Result<(), NetworkError> {
pub fn register_protocol(&mut self, handler: Arc<NetworkProtocolHandler<Message>+Send + Sync>, protocol: ProtocolId, versions: &[u8]) -> Result<(), NetworkError> {
try!(self.io_service.send_message(NetworkIoMessage::AddHandler {
handler: Some(handler),
handler: handler,
protocol: protocol,
versions: versions.to_vec(),
}));

20
util/src/network/session.rs
View File

@@ -4,6 +4,7 @@ use rlp::*;
use network::connection::{EncryptedConnection, Packet};
use network::handshake::Handshake;
use error::*;
use io::{IoContext};
use network::error::{NetworkError, DisconnectReason};
use network::host::*;
use network::node::NodeId;
@@ -84,7 +85,7 @@ const PACKET_LAST: u8 = 0x7f;
impl Session {
/// Create a new session out of comepleted handshake. Consumes handshake object.
pub fn new<Host:Handler<Timeout=Token>>(h: Handshake, event_loop: &mut EventLoop<Host>, host: &HostInfo) -> Result<Session, UtilError> {
pub fn new<Message>(h: Handshake, _io: &IoContext<Message>, host: &HostInfo) -> Result<Session, UtilError> where Message: Send + Sync + Clone {
let id = h.id.clone();
let connection = try!(EncryptedConnection::new(h));
let mut session = Session {
@@ -99,7 +100,6 @@ impl Session {
};
try!(session.write_hello(host));
try!(session.write_ping());
try!(session.connection.register(event_loop));
Ok(session)
}
@@ -109,16 +109,16 @@ impl Session {
}
/// Readable IO handler. Returns packet data if available.
pub fn readable<Host:Handler>(&mut self, event_loop: &mut EventLoop<Host>, host: &HostInfo) -> Result<SessionData, UtilError> {
match try!(self.connection.readable(event_loop)) {
pub fn readable<Message>(&mut self, io: &IoContext<Message>, host: &HostInfo) -> Result<SessionData, UtilError> where Message: Send + Sync + Clone {
match try!(self.connection.readable(io)) {
Some(data) => Ok(try!(self.read_packet(data, host))),
None => Ok(SessionData::None)
}
}
/// Writable IO handler. Sends pending packets.
pub fn writable<Host:Handler>(&mut self, event_loop: &mut EventLoop<Host>, _host: &HostInfo) -> Result<(), UtilError> {
self.connection.writable(event_loop)
pub fn writable<Message>(&mut self, io: &IoContext<Message>, _host: &HostInfo) -> Result<(), UtilError> where Message: Send + Sync + Clone {
self.connection.writable(io)
}
/// Checks if peer supports given capability
@@ -127,8 +127,8 @@ impl Session {
}
/// Update registration with the event loop. Should be called at the end of the IO handler.
pub fn reregister<Host:Handler>(&mut self, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
self.connection.reregister(event_loop)
pub fn update_socket<Host:Handler>(&self, reg:Token, event_loop: &mut EventLoop<Host>) -> Result<(), UtilError> {
self.connection.update_socket(reg, event_loop)
}
/// Send a protocol packet to peer.
@@ -182,7 +182,7 @@ impl Session {
// map to protocol
let protocol = self.info.capabilities[i].protocol;
let pid = packet_id - self.info.capabilities[i].id_offset;
return Ok(SessionData::Packet { data: packet.data, protocol: protocol, packet_id: pid } )
Ok(SessionData::Packet { data: packet.data, protocol: protocol, packet_id: pid } )
},
_ => {
debug!(target: "net", "Unkown packet: {:?}", packet_id);
@@ -212,7 +212,7 @@ impl Session {
// Intersect with host capabilities
// Leave only highset mutually supported capability version
let mut caps: Vec<SessionCapabilityInfo> = Vec::new();
for hc in host.capabilities.iter() {
for hc in &host.capabilities {
if peer_caps.iter().any(|c| c.protocol == hc.protocol && c.version == hc.version) {
caps.push(SessionCapabilityInfo {
protocol: hc.protocol,

11
util/src/overlaydb.rs
View File

@@ -169,7 +169,7 @@ impl HashDB for OverlayDB {
match k {
Some(&(ref d, rc)) if rc > 0 => Some(d),
_ => {
let memrc = k.map(|&(_, rc)| rc).unwrap_or(0);
let memrc = k.map_or(0, |&(_, rc)| rc);
match self.payload(key) {
Some(x) => {
let (d, rc) = x;
@@ -194,16 +194,11 @@ impl HashDB for OverlayDB {
match k {
Some(&(_, rc)) if rc > 0 => true,
_ => {
let memrc = k.map(|&(_, rc)| rc).unwrap_or(0);
let memrc = k.map_or(0, |&(_, rc)| rc);
match self.payload(key) {
Some(x) => {
let (_, rc) = x;
if rc as i32 + memrc > 0 {
true
}
else {
false
}
rc as i32 + memrc > 0
}
// Replace above match arm with this once https://github.com/rust-lang/rust/issues/15287 is done.
//Some((d, rc)) if rc + memrc > 0 => true,

6
util/src/rlp/rlpstream.rs
View File

@@ -41,7 +41,7 @@ impl Stream for RlpStream {
stream
}
fn append<'a, E>(&'a mut self, object: &E) -> &'a mut RlpStream where E: Encodable {
fn append<E>(&mut self, object: &E) -> &mut RlpStream where E: Encodable {
// encode given value and add it at the end of the stream
object.encode(&mut self.encoder);
@@ -52,7 +52,7 @@ impl Stream for RlpStream {
self
}
fn append_list<'a>(&'a mut self, len: usize) -> &'a mut RlpStream {
fn append_list(&mut self, len: usize) -> &mut RlpStream {
match len {
0 => {
// we may finish, if the appended list len is equal 0
@@ -69,7 +69,7 @@ impl Stream for RlpStream {
self
}
fn append_empty_data<'a>(&'a mut self) -> &'a mut RlpStream {
fn append_empty_data(&mut self) -> &mut RlpStream {
// self push raw item
self.encoder.bytes.push(0x80);

6
util/src/rlp/rlptraits.rs
View File

@@ -9,7 +9,7 @@ pub trait Decoder: Sized {
/// TODO [arkpar] Please document me
fn as_list(&self) -> Result<Vec<Self>, DecoderError>;
/// TODO [Gav Wood] Please document me
fn as_rlp<'a>(&'a self) -> &'a UntrustedRlp<'a>;
fn as_rlp(&self) -> &UntrustedRlp;
/// TODO [debris] Please document me
fn as_raw(&self) -> &[u8];
}
@@ -255,7 +255,7 @@ pub trait Stream: Sized {
/// assert_eq!(out, vec![0xca, 0xc8, 0x83, b'c', b'a', b't', 0x83, b'd', b'o', b'g', 0x80]);
/// }
/// ```
fn append_list<'a>(&'a mut self, len: usize) -> &'a mut Self;
fn append_list(&mut self, len: usize) -> &mut Self;
/// Apends null to the end of stream, chainable.
///
@@ -270,7 +270,7 @@ pub trait Stream: Sized {
/// assert_eq!(out, vec![0xc2, 0x80, 0x80]);
/// }
/// ```
fn append_empty_data<'a>(&'a mut self) -> &'a mut Self;
fn append_empty_data(&mut self) -> &mut Self;
/// Appends raw (pre-serialised) RLP data. Use with caution. Chainable.
fn append_raw<'a>(&'a mut self, bytes: &[u8], item_count: usize) -> &'a mut Self;

28
util/src/rlp/tests.rs
View File

@@ -15,25 +15,25 @@ fn rlp_at() {
assert!(rlp.is_list());
//let animals = <Vec<String> as rlp::Decodable>::decode_untrusted(&rlp).unwrap();
let animals: Vec<String> = rlp.as_val().unwrap();
assert_eq!(animals, vec!["cat".to_string(), "dog".to_string()]);
assert_eq!(animals, vec!["cat".to_owned(), "dog".to_owned()]);
let cat = rlp.at(0).unwrap();
assert!(cat.is_data());
assert_eq!(cat.as_raw(), &[0x83, b'c', b'a', b't']);
//assert_eq!(String::decode_untrusted(&cat).unwrap(), "cat".to_string());
assert_eq!(cat.as_val::<String>().unwrap(), "cat".to_string());
//assert_eq!(String::decode_untrusted(&cat).unwrap(), "cat".to_owned());
assert_eq!(cat.as_val::<String>().unwrap(), "cat".to_owned());
let dog = rlp.at(1).unwrap();
assert!(dog.is_data());
assert_eq!(dog.as_raw(), &[0x83, b'd', b'o', b'g']);
//assert_eq!(String::decode_untrusted(&dog).unwrap(), "dog".to_string());
assert_eq!(dog.as_val::<String>().unwrap(), "dog".to_string());
//assert_eq!(String::decode_untrusted(&dog).unwrap(), "dog".to_owned());
assert_eq!(dog.as_val::<String>().unwrap(), "dog".to_owned());
let cat_again = rlp.at(0).unwrap();
assert!(cat_again.is_data());
assert_eq!(cat_again.as_raw(), &[0x83, b'c', b'a', b't']);
//assert_eq!(String::decode_untrusted(&cat_again).unwrap(), "cat".to_string());
assert_eq!(cat_again.as_val::<String>().unwrap(), "cat".to_string());
//assert_eq!(String::decode_untrusted(&cat_again).unwrap(), "cat".to_owned());
assert_eq!(cat_again.as_val::<String>().unwrap(), "cat".to_owned());
}
}
@@ -268,13 +268,13 @@ fn decode_untrusted_u256() {
#[test]
fn decode_untrusted_str() {
let tests = vec![DTestPair("cat".to_string(), vec![0x83, b'c', b'a', b't']),
DTestPair("dog".to_string(), vec![0x83, b'd', b'o', b'g']),
DTestPair("Marek".to_string(),
let tests = vec![DTestPair("cat".to_owned(), vec![0x83, b'c', b'a', b't']),
DTestPair("dog".to_owned(), vec![0x83, b'd', b'o', b'g']),
DTestPair("Marek".to_owned(),
vec![0x85, b'M', b'a', b'r', b'e', b'k']),
DTestPair("".to_string(), vec![0x80]),
DTestPair("".to_owned(), vec![0x80]),
DTestPair("Lorem ipsum dolor sit amet, consectetur adipisicing elit"
.to_string(),
.to_owned(),
vec![0xb8, 0x38, b'L', b'o', b'r', b'e', b'm', b' ', b'i',
b'p', b's', b'u', b'm', b' ', b'd', b'o', b'l', b'o',
b'r', b' ', b's', b'i', b't', b' ', b'a', b'm', b'e',
@@ -311,14 +311,14 @@ fn decode_untrusted_vector_u64() {
#[test]
fn decode_untrusted_vector_str() {
let tests = vec![DTestPair(vec!["cat".to_string(), "dog".to_string()],
let tests = vec![DTestPair(vec!["cat".to_owned(), "dog".to_owned()],
vec![0xc8, 0x83, b'c', b'a', b't', 0x83, b'd', b'o', b'g'])];
run_decode_tests(tests);
}
#[test]
fn decode_untrusted_vector_of_vectors_str() {
let tests = vec![DTestPair(vec![vec!["cat".to_string()]],
let tests = vec![DTestPair(vec![vec!["cat".to_owned()]],
vec![0xc5, 0xc4, 0x83, b'c', b'a', b't'])];
run_decode_tests(tests);
}

9
util/src/rlp/untrusted_rlp.rs
View File

@@ -288,7 +288,7 @@ impl<'a> BasicDecoder<'a> {
/// Return first item info
fn payload_info(bytes: &[u8]) -> Result<PayloadInfo, DecoderError> {
let item = match bytes.first().map(|&x| x) {
let item = match bytes.first().cloned() {
None => return Err(DecoderError::RlpIsTooShort),
Some(0...0x7f) => PayloadInfo::new(0, 1),
Some(l @ 0x80...0xb7) => PayloadInfo::new(1, l as usize - 0x80),
@@ -324,7 +324,7 @@ impl<'a> Decoder for BasicDecoder<'a> {
let bytes = self.rlp.as_raw();
match bytes.first().map(|&x| x) {
match bytes.first().cloned() {
// rlp is too short
None => Err(DecoderError::RlpIsTooShort),
// single byt value
@@ -355,12 +355,12 @@ impl<'a> Decoder for BasicDecoder<'a> {
fn as_list(&self) -> Result<Vec<Self>, DecoderError> {
let v: Vec<BasicDecoder<'a>> = self.rlp.iter()
.map(| i | BasicDecoder::new(i))
.map(BasicDecoder::new)
.collect();
Ok(v)
}
fn as_rlp<'s>(&'s self) -> &'s UntrustedRlp<'s> {
fn as_rlp(&self) -> &UntrustedRlp {
&self.rlp
}
}
@@ -405,6 +405,7 @@ impl<T> Decodable for Option<T> where T: Decodable {
macro_rules! impl_array_decodable {
($index_type:ty, $len:expr ) => (
impl<T> Decodable for [T; $len] where T: Decodable {
#[allow(len_zero)]
fn decode<D>(decoder: &D) -> Result<Self, DecoderError> where D: Decoder {
let decoders = try!(decoder.as_list());

4
util/src/squeeze.rs
View File

@@ -42,7 +42,7 @@ pub trait Squeeze {
impl<K, T> Squeeze for HashMap<K, T> where K: Eq + Hash + Clone + HeapSizeOf, T: HeapSizeOf {
fn squeeze(&mut self, size: usize) {
if self.len() == 0 {
if self.is_empty() {
return
}
@@ -50,7 +50,7 @@ impl<K, T> Squeeze for HashMap<K, T> where K: Eq + Hash + Clone + HeapSizeOf, T:
let all_entries = size_of_entry * self.len();
let mut shrinked_size = all_entries;
while self.len() > 0 && shrinked_size > size {
while !self.is_empty() && shrinked_size > size {
// could be optimized
let key = self.keys().next().unwrap().clone();
self.remove(&key);

12
util/src/trie/triedb.rs
View File

@@ -38,6 +38,7 @@ pub struct TrieDB<'db> {
pub hash_count: usize,
}
#[allow(wrong_self_convention)]
impl<'db> TrieDB<'db> {
/// Create a new trie with the backing database `db` and `root`
/// Panics, if `root` does not exist
@@ -103,7 +104,7 @@ impl<'db> TrieDB<'db> {
match node {
Node::Extension(_, payload) => handle_payload(payload),
Node::Branch(payloads, _) => for payload in payloads.iter() { handle_payload(payload) },
Node::Branch(payloads, _) => for payload in &payloads { handle_payload(payload) },
_ => {},
}
}
@@ -141,12 +142,9 @@ impl<'db> TrieDB<'db> {
},
Node::Branch(ref nodes, ref value) => {
try!(writeln!(f, ""));
match value {
&Some(v) => {
try!(self.fmt_indent(f, deepness + 1));
try!(writeln!(f, "=: {:?}", v.pretty()))
},
&None => {}
if let Some(v) = *value {
try!(self.fmt_indent(f, deepness + 1));
try!(writeln!(f, "=: {:?}", v.pretty()))
}
for i in 0..16 {
match self.get_node(nodes[i]) {

27
util/src/trie/triedbmut.rs
View File

@@ -50,6 +50,7 @@ enum MaybeChanged<'a> {
Changed(Bytes),
}
#[allow(wrong_self_convention)]
impl<'db> TrieDBMut<'db> {
/// Create a new trie with the backing database `db` and empty `root`
/// Initialise to the state entailed by the genesis block.
@@ -145,7 +146,7 @@ impl<'db> TrieDBMut<'db> {
match node {
Node::Extension(_, payload) => handle_payload(payload),
Node::Branch(payloads, _) => for payload in payloads.iter() { handle_payload(payload) },
Node::Branch(payloads, _) => for payload in &payloads { handle_payload(payload) },
_ => {},
}
}
@@ -178,12 +179,9 @@ impl<'db> TrieDBMut<'db> {
},
Node::Branch(ref nodes, ref value) => {
try!(writeln!(f, ""));
match value {
&Some(v) => {
try!(self.fmt_indent(f, deepness + 1));
try!(writeln!(f, "=: {:?}", v.pretty()))
},
&None => {}
if let Some(v) = *value {
try!(self.fmt_indent(f, deepness + 1));
try!(writeln!(f, "=: {:?}", v.pretty()))
}
for i in 0..16 {
match self.get_node(nodes[i]) {
@@ -331,6 +329,7 @@ impl<'db> TrieDBMut<'db> {
}
}
#[allow(cyclomatic_complexity)]
/// Determine the RLP of the node, assuming we're inserting `partial` into the
/// node currently of data `old`. This will *not* delete any hash of `old` from the database;
/// it will just return the new RLP that includes the new node.
@@ -694,7 +693,7 @@ mod tests {
}
}
fn populate_trie<'db>(db: &'db mut HashDB, root: &'db mut H256, v: &Vec<(Vec<u8>, Vec<u8>)>) -> TrieDBMut<'db> {
fn populate_trie<'db>(db: &'db mut HashDB, root: &'db mut H256, v: &[(Vec<u8>, Vec<u8>)]) -> TrieDBMut<'db> {
let mut t = TrieDBMut::new(db, root);
for i in 0..v.len() {
let key: &[u8]= &v[i].0;
@@ -704,8 +703,8 @@ mod tests {
t
}
fn unpopulate_trie<'a, 'db>(t: &mut TrieDBMut<'db>, v: &Vec<(Vec<u8>, Vec<u8>)>) {
for i in v.iter() {
fn unpopulate_trie<'db>(t: &mut TrieDBMut<'db>, v: &[(Vec<u8>, Vec<u8>)]) {
for i in v {
let key: &[u8]= &i.0;
t.remove(&key);
}
@@ -761,7 +760,7 @@ mod tests {
println!("TRIE MISMATCH");
println!("");
println!("{:?} vs {:?}", memtrie.root(), real);
for i in x.iter() {
for i in &x {
println!("{:?} -> {:?}", i.0.pretty(), i.1.pretty());
}
println!("{:?}", memtrie);
@@ -774,7 +773,7 @@ mod tests {
println!("");
println!("remaining: {:?}", memtrie.db_items_remaining());
println!("{:?} vs {:?}", memtrie.root(), real);
for i in x.iter() {
for i in &x {
println!("{:?} -> {:?}", i.0.pretty(), i.1.pretty());
}
println!("{:?}", memtrie);
@@ -1051,12 +1050,12 @@ mod tests {
println!("TRIE MISMATCH");
println!("");
println!("ORIGINAL... {:?}", memtrie.root());
for i in x.iter() {
for i in &x {
println!("{:?} -> {:?}", i.0.pretty(), i.1.pretty());
}
println!("{:?}", memtrie);
println!("SORTED... {:?}", memtrie_sorted.root());
for i in y.iter() {
for i in &y {
println!("{:?} -> {:?}", i.0.pretty(), i.1.pretty());
}
println!("{:?}", memtrie_sorted);

19
util/src/uint.rs
View File

@@ -200,7 +200,7 @@ macro_rules! construct_uint {
#[inline]
fn byte(&self, index: usize) -> u8 {
let &$name(ref arr) = self;
(arr[index / 8] >> ((index % 8)) * 8) as u8
(arr[index / 8] >> (((index % 8)) * 8)) as u8
}
fn to_bytes(&self, bytes: &mut[u8]) {
@@ -446,16 +446,16 @@ macro_rules! construct_uint {
impl FromJson for $name {
fn from_json(json: &Json) -> Self {
match json {
&Json::String(ref s) => {
match *json {
Json::String(ref s) => {
if s.len() >= 2 && &s[0..2] == "0x" {
FromStr::from_str(&s[2..]).unwrap_or(Default::default())
FromStr::from_str(&s[2..]).unwrap_or_else(|_| Default::default())
} else {
Uint::from_dec_str(s).unwrap_or(Default::default())
Uint::from_dec_str(s).unwrap_or_else(|_| Default::default())
}
},
&Json::U64(u) => From::from(u),
&Json::I64(i) => From::from(i as u64),
Json::U64(u) => From::from(u),
Json::I64(i) => From::from(i as u64),
_ => Uint::zero(),
}
}
@@ -488,7 +488,7 @@ macro_rules! construct_uint {
for i in 0..bytes.len() {
let rev = bytes.len() - 1 - i;
let pos = rev / 8;
ret[pos] += (bytes[i] as u64) << (rev % 8) * 8;
ret[pos] += (bytes[i] as u64) << ((rev % 8) * 8);
}
$name(ret)
}
@@ -500,7 +500,7 @@ macro_rules! construct_uint {
fn from_str(value: &str) -> Result<$name, Self::Err> {
let bytes: Vec<u8> = match value.len() % 2 == 0 {
true => try!(value.from_hex()),
false => try!(("0".to_string() + value).from_hex())
false => try!(("0".to_owned() + value).from_hex())
};
let bytes_ref: &[u8] = &bytes;
@@ -1061,6 +1061,7 @@ mod tests {
}
#[test]
#[allow(eq_op)]
pub fn uint256_comp_test() {
let small = U256([10u64, 0, 0, 0]);
let big = U256([0x8C8C3EE70C644118u64, 0x0209E7378231E632, 0, 0]);