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

merge accountdb_migration

Browse Source
This commit is contained in:
Robert Habermeier
2016-07-11 11:08:41 +02:00
parent 36dfa4743e c8b565805e
commit 3dd7ce5566
67 changed files with 1112 additions and 674 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
util/bigint/src/uint.rs
View File

@@ -524,9 +524,8 @@ pub trait Uint: Sized + Default + FromStr + From<u64> + fmt::Debug + fmt::Displa
fn bit(&self, index: usize) -> bool;
/// Return single byte
fn byte(&self, index: usize) -> u8;
/// Get this Uint as slice of bytes
fn to_raw_bytes(&self, bytes: &mut[u8]);
/// Convert U256 to the sequence of bytes with a big endian
fn to_big_endian(&self, bytes: &mut[u8]);
/// Create `Uint(10**n)`
fn exp10(n: usize) -> Self;
/// Return eponentation `self**other`. Panic on overflow.
@@ -551,6 +550,9 @@ pub trait Uint: Sized + Default + FromStr + From<u64> + fmt::Debug + fmt::Displa
/// Returns negation of this `Uint` and overflow (always true)
fn overflowing_neg(self) -> (Self, bool);
/// Returns
fn is_zero(&self) -> bool;
}
macro_rules! construct_uint {
@@ -616,6 +618,13 @@ macro_rules! construct_uint {
arr[0]
}
#[inline]
fn is_zero(&self) -> bool {
let &$name(ref arr) = self;
for i in 0..$n_words { if arr[i] != 0 { return false; } }
return true;
}
/// Return the least number of bits needed to represent the number
#[inline]
fn bits(&self) -> usize {
@@ -638,7 +647,7 @@ macro_rules! construct_uint {
(arr[index / 8] >> (((index % 8)) * 8)) as u8
}
fn to_raw_bytes(&self, bytes: &mut[u8]) {
fn to_big_endian(&self, bytes: &mut[u8]) {
assert!($n_words * 8 == bytes.len());
let &$name(ref arr) = self;
for i in 0..bytes.len() {
@@ -1454,7 +1463,7 @@ mod tests {
let hex = "8090a0b0c0d0e0f00910203040506077583a2cf8264910e1436bda32571012f0";
let uint = U256::from_str(hex).unwrap();
let mut bytes = [0u8; 32];
uint.to_raw_bytes(&mut bytes);
uint.to_big_endian(&mut bytes);
let uint2 = U256::from(&bytes[..]);
assert_eq!(uint, uint2);
}
@@ -2021,6 +2030,44 @@ mod tests {
assert!(overflow);
}
#[test]
fn big_endian() {
let source = U256([1, 0, 0, 0]);
let mut target = vec![0u8; 32];
assert_eq!(source, U256::from(1));
source.to_big_endian(&mut target);
assert_eq!(
vec![0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8,
0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 1u8],
target);
let source = U256([512, 0, 0, 0]);
let mut target = vec![0u8; 32];
source.to_big_endian(&mut target);
assert_eq!(
vec![0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8,
0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 2u8, 0u8],
target);
let source = U256([0, 512, 0, 0]);
let mut target = vec![0u8; 32];
source.to_big_endian(&mut target);
assert_eq!(
vec![0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8,
0u8, 0u8, 0u8, 0u8, 0u8, 2u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8, 0u8],
target);
let source = U256::from_str("0102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f20").unwrap();
source.to_big_endian(&mut target);
assert_eq!(
vec![0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0a, 0x0b, 0x0c, 0x0d, 0x0e, 0x0f, 0x10, 0x11,
0x12, 0x13, 0x14, 0x15, 0x16, 0x17, 0x18, 0x19, 0x1a, 0x1b, 0x1c, 0x1d, 0x1e, 0x1f, 0x20],
target);
}
#[test]
#[cfg_attr(feature="dev", allow(cyclomatic_complexity))]

4
util/src/hash.rs
View File

@@ -445,7 +445,7 @@ macro_rules! impl_hash {
impl From<U256> for H256 {
fn from(value: U256) -> H256 {
let mut ret = H256::new();
value.to_raw_bytes(&mut ret);
value.to_big_endian(&mut ret);
ret
}
}
@@ -453,7 +453,7 @@ impl From<U256> for H256 {
impl<'a> From<&'a U256> for H256 {
fn from(value: &'a U256) -> H256 {
let mut ret: H256 = H256::new();
value.to_raw_bytes(&mut ret);
value.to_big_endian(&mut ret);
ret
}
}

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

@@ -21,6 +21,7 @@ use mio::*;
use crossbeam::sync::chase_lev;
use slab::Slab;
use error::*;
use misc::*;
use io::{IoError, IoHandler};
use io::worker::{Worker, Work, WorkType};
use panics::*;
@@ -227,7 +228,7 @@ impl<Message> Handler for IoManager<Message> where Message: Send + Clone + Sync
let handler_index = token.as_usize() / TOKENS_PER_HANDLER;
let token_id = token.as_usize() % TOKENS_PER_HANDLER;
if let Some(handler) = self.handlers.get(handler_index) {
if let Some(timer) = self.timers.read().unwrap().get(&token.as_usize()) {
if let Some(timer) = self.timers.unwrapped_read().get(&token.as_usize()) {
event_loop.timeout_ms(token, timer.delay).expect("Error re-registering user timer");
self.worker_channel.push(Work { work_type: WorkType::Timeout, token: token_id, handler: handler.clone(), handler_id: handler_index });
self.work_ready.notify_all();
@@ -249,7 +250,7 @@ impl<Message> Handler for IoManager<Message> where Message: Send + Clone + Sync
// TODO: flush event loop
self.handlers.remove(handler_id);
// unregister timers
let mut timers = self.timers.write().unwrap();
let mut timers = self.timers.unwrapped_write();
let to_remove: Vec<_> = timers.keys().cloned().filter(|timer_id| timer_id / TOKENS_PER_HANDLER == handler_id).collect();
for timer_id in to_remove {
let timer = timers.remove(&timer_id).expect("to_remove only contains keys from timers; qed");
@@ -259,11 +260,11 @@ impl<Message> Handler for IoManager<Message> where Message: Send + Clone + Sync
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 });
self.timers.unwrapped_write().insert(timer_id, UserTimer { delay: delay, timeout: timeout });
},
IoMessage::RemoveTimer { handler_id, token } => {
let timer_id = token + handler_id * TOKENS_PER_HANDLER;
if let Some(timer) = self.timers.write().unwrap().remove(&timer_id) {
if let Some(timer) = self.timers.unwrapped_write().remove(&timer_id) {
event_loop.clear_timeout(timer.timeout);
}
},
@@ -277,7 +278,7 @@ impl<Message> Handler for IoManager<Message> where Message: Send + Clone + Sync
handler.deregister_stream(token, event_loop);
// unregister a timer associated with the token (if any)
let timer_id = token + handler_id * TOKENS_PER_HANDLER;
if let Some(timer) = self.timers.write().unwrap().remove(&timer_id) {
if let Some(timer) = self.timers.unwrapped_write().remove(&timer_id) {
event_loop.clear_timeout(timer.timeout);
}
}

3
util/src/io/worker.rs
View File

@@ -22,6 +22,7 @@ use crossbeam::sync::chase_lev;
use io::service::{HandlerId, IoChannel, IoContext};
use io::{IoHandler};
use panics::*;
use misc::Lockable;
pub enum WorkType<Message> {
Readable,
@@ -81,7 +82,7 @@ impl Worker {
where Message: Send + Sync + Clone + 'static {
loop {
{
let lock = wait_mutex.lock().unwrap();
let lock = wait_mutex.locked();
if deleting.load(AtomicOrdering::Acquire) {
return;
}

7
util/src/journaldb/earlymergedb.rs
View File

@@ -20,6 +20,7 @@ use common::*;
use rlp::*;
use hashdb::*;
use memorydb::*;
use misc::RwLockable;
use super::{DB_PREFIX_LEN, LATEST_ERA_KEY, VERSION_KEY};
use super::traits::JournalDB;
use kvdb::{Database, DBTransaction, DatabaseConfig};
@@ -225,7 +226,7 @@ impl EarlyMergeDB {
#[cfg(test)]
fn can_reconstruct_refs(&self) -> bool {
let (latest_era, reconstructed) = Self::read_refs(&self.backing);
let refs = self.refs.as_ref().unwrap().write().unwrap();
let refs = self.refs.as_ref().unwrap().unwrapped_write();
if *refs != reconstructed || latest_era != self.latest_era {
let clean_refs = refs.iter().filter_map(|(k, v)| if reconstructed.get(k) == Some(v) {None} else {Some((k.clone(), v.clone()))}).collect::<HashMap<_, _>>();
let clean_recon = reconstructed.into_iter().filter_map(|(k, v)| if refs.get(&k) == Some(&v) {None} else {Some((k.clone(), v.clone()))}).collect::<HashMap<_, _>>();
@@ -333,7 +334,7 @@ impl JournalDB for EarlyMergeDB {
fn mem_used(&self) -> usize {
self.overlay.mem_used() + match self.refs {
Some(ref c) => c.read().unwrap().heap_size_of_children(),
Some(ref c) => c.unwrapped_read().heap_size_of_children(),
None => 0
}
}
@@ -385,7 +386,7 @@ impl JournalDB for EarlyMergeDB {
//
// record new commit's details.
let mut refs = self.refs.as_ref().unwrap().write().unwrap();
let mut refs = self.refs.as_ref().unwrap().unwrapped_write();
let batch = DBTransaction::new();
let trace = false;
{

11
util/src/journaldb/overlayrecentdb.rs
View File

@@ -20,6 +20,7 @@ use common::*;
use rlp::*;
use hashdb::*;
use memorydb::*;
use misc::RwLockable;
use super::{DB_PREFIX_LEN, LATEST_ERA_KEY, VERSION_KEY};
use kvdb::{Database, DBTransaction, DatabaseConfig};
#[cfg(test)]
@@ -136,7 +137,7 @@ impl OverlayRecentDB {
#[cfg(test)]
fn can_reconstruct_refs(&self) -> bool {
let reconstructed = Self::read_overlay(&self.backing);
let journal_overlay = self.journal_overlay.read().unwrap();
let journal_overlay = self.journal_overlay.unwrapped_read();
*journal_overlay == reconstructed
}
@@ -199,7 +200,7 @@ impl JournalDB for OverlayRecentDB {
fn mem_used(&self) -> usize {
let mut mem = self.transaction_overlay.mem_used();
let overlay = self.journal_overlay.read().unwrap();
let overlay = self.journal_overlay.unwrapped_read();
mem += overlay.backing_overlay.mem_used();
mem += overlay.journal.heap_size_of_children();
mem
@@ -209,12 +210,12 @@ impl JournalDB for OverlayRecentDB {
self.backing.get(&LATEST_ERA_KEY).expect("Low level database error").is_none()
}
fn latest_era(&self) -> Option<u64> { self.journal_overlay.read().unwrap().latest_era }
fn latest_era(&self) -> Option<u64> { self.journal_overlay.unwrapped_read().latest_era }
fn commit(&mut self, now: u64, id: &H256, end: Option<(u64, H256)>) -> Result<u32, UtilError> {
// record new commit's details.
trace!("commit: #{} ({}), end era: {:?}", now, id, end);
let mut journal_overlay = self.journal_overlay.write().unwrap();
let mut journal_overlay = self.journal_overlay.unwrapped_write();
let batch = DBTransaction::new();
{
let mut r = RlpStream::new_list(3);
@@ -321,7 +322,7 @@ impl HashDB for OverlayRecentDB {
match k {
Some(&(ref d, rc)) if rc > 0 => Some(d),
_ => {
let v = self.journal_overlay.read().unwrap().backing_overlay.get(key).map(|v| v.to_vec());
let v = self.journal_overlay.unwrapped_read().backing_overlay.get(key).map(|v| v.to_vec());
match v {
Some(x) => {
Some(&self.transaction_overlay.denote(key, x).0)

7
util/src/log.rs
View File

@@ -23,6 +23,7 @@ use env_logger::LogBuilder;
use std::sync::{RwLock, RwLockReadGuard};
use std::sync::atomic::{Ordering, AtomicBool};
use arrayvec::ArrayVec;
use misc::RwLockable;
pub use ansi_term::{Colour, Style};
lazy_static! {
@@ -54,7 +55,7 @@ lazy_static! {
builder.parse(&log);
}
if let Ok(_) = builder.init() {
if builder.init().is_ok() {
println!("logger initialized");
}
true
@@ -90,7 +91,7 @@ impl RotatingLogger {
/// Append new log entry
pub fn append(&self, log: String) {
self.logs.write().unwrap().insert(0, log);
self.logs.unwrapped_write().insert(0, log);
}
/// Return levels
@@ -100,7 +101,7 @@ impl RotatingLogger {
/// Return logs
pub fn logs(&self) -> RwLockReadGuard<ArrayVec<[String; LOG_SIZE]>> {
self.logs.read().unwrap()
self.logs.unwrapped_read()
}
}

4
util/src/migration/mod.rs
View File

@@ -209,12 +209,12 @@ impl Manager {
// start with the old db.
let old_path_str = try!(old_path.to_str().ok_or(Error::MigrationImpossible));
let mut cur_db = try!(Database::open(&db_config, old_path_str).map_err(|s| Error::Custom(s)));
let mut cur_db = try!(Database::open(&db_config, old_path_str).map_err(Error::Custom));
for migration in migrations {
// open the target temporary database.
temp_path = temp_idx.path(&db_root);
let temp_path_str = try!(temp_path.to_str().ok_or(Error::MigrationImpossible));
let mut new_db = try!(Database::open(&db_config, temp_path_str).map_err(|s| Error::Custom(s)));
let mut new_db = try!(Database::open(&db_config, temp_path_str).map_err(Error::Custom));
// perform the migration from cur_db to new_db.
try!(migration.migrate(&cur_db, &config, &mut new_db));

25
util/src/misc.rs
View File

@@ -65,3 +65,28 @@ pub fn version_data() -> Bytes {
s.append(&&Target::os()[0..2]);
s.out()
}
/// Object can be locked directly into a `MutexGuard`.
pub trait Lockable<T> {
/// Lock object directly into a `MutexGuard`.
fn locked(&self) -> MutexGuard<T>;
}
impl<T> Lockable<T> for Mutex<T> {
fn locked(&self) -> MutexGuard<T> { self.lock().unwrap() }
}
/// Object can be read or write locked directly into a guard.
pub trait RwLockable<T> {
/// Read-lock object directly into a `ReadGuard`.
fn unwrapped_read(&self) -> RwLockReadGuard<T>;
/// Write-lock object directly into a `WriteGuard`.
fn unwrapped_write(&self) -> RwLockWriteGuard<T>;
}
impl<T> RwLockable<T> for RwLock<T> {
fn unwrapped_read(&self) -> RwLockReadGuard<T> { self.read().unwrap() }
fn unwrapped_write(&self) -> RwLockWriteGuard<T> { self.write().unwrap() }
}

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

@@ -28,7 +28,7 @@ use std::fs;
use mio::*;
use mio::tcp::*;
use hash::*;
use misc::version;
use misc::*;
use crypto::*;
use sha3::Hashable;
use rlp::*;
@@ -123,8 +123,7 @@ const IDLE: usize = SYS_TIMER + 2;
const DISCOVERY: usize = SYS_TIMER + 3;
const DISCOVERY_REFRESH: usize = SYS_TIMER + 4;
const DISCOVERY_ROUND: usize = SYS_TIMER + 5;
const INIT_PUBLIC: usize = SYS_TIMER + 6;
const NODE_TABLE: usize = SYS_TIMER + 7;
const NODE_TABLE: usize = SYS_TIMER + 6;
const FIRST_SESSION: usize = 0;
const LAST_SESSION: usize = FIRST_SESSION + MAX_SESSIONS - 1;
const USER_TIMER: usize = LAST_SESSION + 256;
@@ -156,6 +155,8 @@ pub enum NetworkIoMessage<Message> where Message: Send + Sync + Clone {
/// Timer delay in milliseconds.
delay: u64,
},
/// Initliaze public interface.
InitPublicInterface,
/// Disconnect a peer.
Disconnect(PeerId),
/// Disconnect and temporary disable peer.
@@ -202,7 +203,7 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
protocol: ProtocolId,
session: Option<SharedSession>, sessions: Arc<RwLock<Slab<SharedSession>>>,
reserved_peers: &'s HashSet<NodeId>) -> NetworkContext<'s, Message> {
let id = session.as_ref().map(|s| s.lock().unwrap().token());
let id = session.as_ref().map(|s| s.locked().token());
NetworkContext {
io: io,
protocol: protocol,
@@ -216,7 +217,7 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
fn resolve_session(&self, peer: PeerId) -> Option<SharedSession> {
match self.session_id {
Some(id) if id == peer => self.session.clone(),
_ => self.sessions.read().unwrap().get(peer).cloned(),
_ => self.sessions.unwrapped_read().get(peer).cloned(),
}
}
@@ -224,7 +225,7 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
pub fn send(&self, peer: PeerId, packet_id: PacketId, data: Vec<u8>) -> Result<(), UtilError> {
let session = self.resolve_session(peer);
if let Some(session) = session {
try!(session.lock().unwrap().send_packet(self.io, self.protocol, packet_id as u8, &data));
try!(session.locked().send_packet(self.io, self.protocol, packet_id as u8, &data));
} else {
trace!(target: "network", "Send: Peer no longer exist")
}
@@ -262,7 +263,7 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
/// Check if the session is still active.
pub fn is_expired(&self) -> bool {
self.session.as_ref().map_or(false, |s| s.lock().unwrap().expired())
self.session.as_ref().map_or(false, |s| s.locked().expired())
}
/// Register a new IO timer. 'IoHandler::timeout' will be called with the token.
@@ -279,7 +280,7 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
pub fn peer_info(&self, peer: PeerId) -> String {
let session = self.resolve_session(peer);
if let Some(session) = session {
return session.lock().unwrap().info.client_version.clone()
return session.locked().info.client_version.clone()
}
"unknown".to_owned()
}
@@ -422,8 +423,8 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
Ok(n) => {
let entry = NodeEntry { endpoint: n.endpoint.clone(), id: n.id.clone() };
self.nodes.write().unwrap().add_node(n);
if let Some(ref mut discovery) = *self.discovery.lock().unwrap() {
self.nodes.unwrapped_write().add_node(n);
if let Some(ref mut discovery) = *self.discovery.locked() {
discovery.add_node(entry);
}
}
@@ -434,9 +435,9 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
let n = try!(Node::from_str(id));
let entry = NodeEntry { endpoint: n.endpoint.clone(), id: n.id.clone() };
self.reserved_nodes.write().unwrap().insert(n.id.clone());
self.reserved_nodes.unwrapped_write().insert(n.id.clone());
if let Some(ref mut discovery) = *self.discovery.lock().unwrap() {
if let Some(ref mut discovery) = *self.discovery.locked() {
discovery.add_node(entry);
}
@@ -444,17 +445,17 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
}
pub fn set_non_reserved_mode(&self, mode: NonReservedPeerMode, io: &IoContext<NetworkIoMessage<Message>>) {
let mut info = self.info.write().unwrap();
let mut info = self.info.unwrapped_write();
if info.config.non_reserved_mode != mode {
info.config.non_reserved_mode = mode.clone();
drop(info);
if let NonReservedPeerMode::Deny = mode {
// disconnect all non-reserved peers here.
let reserved: HashSet<NodeId> = self.reserved_nodes.read().unwrap().clone();
let reserved: HashSet<NodeId> = self.reserved_nodes.unwrapped_read().clone();
let mut to_kill = Vec::new();
for e in self.sessions.write().unwrap().iter_mut() {
let mut s = e.lock().unwrap();
for e in self.sessions.unwrapped_write().iter_mut() {
let mut s = e.locked();
{
let id = s.id();
if id.is_some() && reserved.contains(id.unwrap()) {
@@ -475,7 +476,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
pub fn remove_reserved_node(&self, id: &str) -> Result<(), UtilError> {
let n = try!(Node::from_str(id));
self.reserved_nodes.write().unwrap().remove(&n.id);
self.reserved_nodes.unwrapped_write().remove(&n.id);
Ok(())
}
@@ -485,11 +486,11 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
}
pub fn external_url(&self) -> Option<String> {
self.info.read().unwrap().public_endpoint.as_ref().map(|e| format!("{}", Node::new(self.info.read().unwrap().id().clone(), e.clone())))
self.info.unwrapped_read().public_endpoint.as_ref().map(|e| format!("{}", Node::new(self.info.unwrapped_read().id().clone(), e.clone())))
}
pub fn local_url(&self) -> String {
let r = format!("{}", Node::new(self.info.read().unwrap().id().clone(), self.info.read().unwrap().local_endpoint.clone()));
let r = format!("{}", Node::new(self.info.unwrapped_read().id().clone(), self.info.unwrapped_read().local_endpoint.clone()));
println!("{}", r);
r
}
@@ -497,8 +498,8 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
pub fn stop(&self, io: &IoContext<NetworkIoMessage<Message>>) -> Result<(), UtilError> {
self.stopping.store(true, AtomicOrdering::Release);
let mut to_kill = Vec::new();
for e in self.sessions.write().unwrap().iter_mut() {
let mut s = e.lock().unwrap();
for e in self.sessions.unwrapped_write().iter_mut() {
let mut s = e.locked();
s.disconnect(io, DisconnectReason::ClientQuit);
to_kill.push(s.token());
}
@@ -511,17 +512,16 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
}
fn init_public_interface(&self, io: &IoContext<NetworkIoMessage<Message>>) -> Result<(), UtilError> {
io.clear_timer(INIT_PUBLIC).unwrap();
if self.info.read().unwrap().public_endpoint.is_some() {
if self.info.unwrapped_read().public_endpoint.is_some() {
return Ok(());
}
let local_endpoint = self.info.read().unwrap().local_endpoint.clone();
let public_address = self.info.read().unwrap().config.public_address.clone();
let local_endpoint = self.info.unwrapped_read().local_endpoint.clone();
let public_address = self.info.unwrapped_read().config.public_address.clone();
let public_endpoint = match public_address {
None => {
let public_address = select_public_address(local_endpoint.address.port());
let public_endpoint = NodeEndpoint { address: public_address, udp_port: local_endpoint.udp_port };
if self.info.read().unwrap().config.nat_enabled {
if self.info.unwrapped_read().config.nat_enabled {
match map_external_address(&local_endpoint) {
Some(endpoint) => {
info!("NAT mapped to external address {}", endpoint.address);
@@ -536,7 +536,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
Some(addr) => NodeEndpoint { address: addr, udp_port: local_endpoint.udp_port }
};
self.info.write().unwrap().public_endpoint = Some(public_endpoint.clone());
self.info.unwrapped_write().public_endpoint = Some(public_endpoint.clone());
if let Some(url) = self.external_url() {
io.message(NetworkIoMessage::NetworkStarted(url)).unwrap_or_else(|e| warn!("Error sending IO notification: {:?}", e));
@@ -544,18 +544,20 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
// Initialize discovery.
let discovery = {
let info = self.info.read().unwrap();
let info = self.info.unwrapped_read();
if info.config.discovery_enabled && info.config.non_reserved_mode == NonReservedPeerMode::Accept {
Some(Discovery::new(&info.keys, public_endpoint.address.clone(), public_endpoint, DISCOVERY))
let mut udp_addr = local_endpoint.address.clone();
udp_addr.set_port(local_endpoint.udp_port);
Some(Discovery::new(&info.keys, udp_addr, public_endpoint, DISCOVERY))
} else { None }
};
if let Some(mut discovery) = discovery {
discovery.init_node_list(self.nodes.read().unwrap().unordered_entries());
for n in self.nodes.read().unwrap().unordered_entries() {
discovery.init_node_list(self.nodes.unwrapped_read().unordered_entries());
for n in self.nodes.unwrapped_read().unordered_entries() {
discovery.add_node(n.clone());
}
*self.discovery.lock().unwrap() = Some(discovery);
*self.discovery.locked() = Some(discovery);
io.register_stream(DISCOVERY).expect("Error registering UDP listener");
io.register_timer(DISCOVERY_REFRESH, 7200).expect("Error registering discovery timer");
io.register_timer(DISCOVERY_ROUND, 300).expect("Error registering discovery timer");
@@ -571,7 +573,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
}
fn have_session(&self, id: &NodeId) -> bool {
self.sessions.read().unwrap().iter().any(|e| e.lock().unwrap().info.id == Some(id.clone()))
self.sessions.unwrapped_read().iter().any(|e| e.locked().info.id == Some(id.clone()))
}
fn session_count(&self) -> usize {
@@ -579,17 +581,17 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
}
fn connecting_to(&self, id: &NodeId) -> bool {
self.sessions.read().unwrap().iter().any(|e| e.lock().unwrap().id() == Some(id))
self.sessions.unwrapped_read().iter().any(|e| e.locked().id() == Some(id))
}
fn handshake_count(&self) -> usize {
self.sessions.read().unwrap().count() - self.session_count()
self.sessions.unwrapped_read().count() - self.session_count()
}
fn keep_alive(&self, io: &IoContext<NetworkIoMessage<Message>>) {
let mut to_kill = Vec::new();
for e in self.sessions.write().unwrap().iter_mut() {
let mut s = e.lock().unwrap();
for e in self.sessions.unwrapped_write().iter_mut() {
let mut s = e.locked();
if !s.keep_alive(io) {
s.disconnect(io, DisconnectReason::PingTimeout);
to_kill.push(s.token());
@@ -603,7 +605,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
fn connect_peers(&self, io: &IoContext<NetworkIoMessage<Message>>) {
let (ideal_peers, mut pin) = {
let info = self.info.read().unwrap();
let info = self.info.unwrapped_read();
if info.capabilities.is_empty() {
return;
}
@@ -613,7 +615,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
};
let session_count = self.session_count();
let reserved_nodes = self.reserved_nodes.read().unwrap();
let reserved_nodes = self.reserved_nodes.unwrapped_read();
if session_count >= ideal_peers as usize + reserved_nodes.len() {
// check if all pinned nodes are connected.
if reserved_nodes.iter().all(|n| self.have_session(n) && self.connecting_to(n)) {
@@ -634,7 +636,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
// iterate over all nodes, reserved ones coming first.
// if we are pinned to only reserved nodes, ignore all others.
let nodes = reserved_nodes.iter().cloned().chain(if !pin {
self.nodes.read().unwrap().nodes()
self.nodes.unwrapped_read().nodes()
} else {
Vec::new()
});
@@ -662,7 +664,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
let socket = {
let address = {
let mut nodes = self.nodes.write().unwrap();
let mut nodes = self.nodes.unwrapped_write();
if let Some(node) = nodes.get_mut(id) {
node.last_attempted = Some(::time::now());
node.endpoint.address
@@ -687,10 +689,10 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
#[cfg_attr(feature="dev", allow(block_in_if_condition_stmt))]
fn create_connection(&self, socket: TcpStream, id: Option<&NodeId>, io: &IoContext<NetworkIoMessage<Message>>) -> Result<(), UtilError> {
let nonce = self.info.write().unwrap().next_nonce();
let mut sessions = self.sessions.write().unwrap();
let nonce = self.info.unwrapped_write().next_nonce();
let mut sessions = self.sessions.unwrapped_write();
let token = sessions.insert_with_opt(|token| {
match Session::new(io, socket, token, id, &nonce, self.stats.clone(), &self.info.read().unwrap()) {
match Session::new(io, socket, token, id, &nonce, self.stats.clone(), &self.info.unwrapped_read()) {
Ok(s) => Some(Arc::new(Mutex::new(s))),
Err(e) => {
debug!(target: "network", "Session create error: {:?}", e);
@@ -711,7 +713,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
fn accept(&self, io: &IoContext<NetworkIoMessage<Message>>) {
trace!(target: "network", "Accepting incoming connection");
loop {
let socket = match self.tcp_listener.lock().unwrap().accept() {
let socket = match self.tcp_listener.locked().accept() {
Ok(None) => break,
Ok(Some((sock, _addr))) => sock,
Err(e) => {
@@ -726,10 +728,10 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
}
fn session_writable(&self, token: StreamToken, io: &IoContext<NetworkIoMessage<Message>>) {
let session = { self.sessions.read().unwrap().get(token).cloned() };
let session = { self.sessions.unwrapped_read().get(token).cloned() };
if let Some(session) = session {
let mut s = session.lock().unwrap();
if let Err(e) = s.writable(io, &self.info.read().unwrap()) {
let mut s = session.locked();
if let Err(e) = s.writable(io, &self.info.unwrapped_read()) {
trace!(target: "network", "Session write error: {}: {:?}", token, e);
}
if s.done() {
@@ -748,16 +750,16 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
let mut ready_data: Vec<ProtocolId> = Vec::new();
let mut packet_data: Vec<(ProtocolId, PacketId, Vec<u8>)> = Vec::new();
let mut kill = false;
let session = { self.sessions.read().unwrap().get(token).cloned() };
let session = { self.sessions.unwrapped_read().get(token).cloned() };
if let Some(session) = session.clone() {
let mut s = session.lock().unwrap();
let mut s = session.locked();
loop {
match s.readable(io, &self.info.read().unwrap()) {
match s.readable(io, &self.info.unwrapped_read()) {
Err(e) => {
trace!(target: "network", "Session read error: {}:{:?} ({:?}) {:?}", token, s.id(), s.remote_addr(), e);
if let UtilError::Network(NetworkError::Disconnect(DisconnectReason::IncompatibleProtocol)) = e {
if let Some(id) = s.id() {
self.nodes.write().unwrap().mark_as_useless(id);
self.nodes.unwrapped_write().mark_as_useless(id);
}
}
kill = true;
@@ -767,9 +769,9 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
self.num_sessions.fetch_add(1, AtomicOrdering::SeqCst);
if !s.info.originated {
let session_count = self.session_count();
let reserved_nodes = self.reserved_nodes.read().unwrap();
let reserved_nodes = self.reserved_nodes.unwrapped_read();
let (ideal_peers, reserved_only) = {
let info = self.info.read().unwrap();
let info = self.info.unwrapped_read();
(info.config.ideal_peers, info.config.non_reserved_mode == NonReservedPeerMode::Deny)
};
@@ -784,14 +786,14 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
// Add it no node table
if let Ok(address) = s.remote_addr() {
let entry = NodeEntry { id: s.id().unwrap().clone(), endpoint: NodeEndpoint { address: address, udp_port: address.port() } };
self.nodes.write().unwrap().add_node(Node::new(entry.id.clone(), entry.endpoint.clone()));
let mut discovery = self.discovery.lock().unwrap();
self.nodes.unwrapped_write().add_node(Node::new(entry.id.clone(), entry.endpoint.clone()));
let mut discovery = self.discovery.locked();
if let Some(ref mut discovery) = *discovery.deref_mut() {
discovery.add_node(entry);
}
}
}
for (p, _) in self.handlers.read().unwrap().iter() {
for (p, _) in self.handlers.unwrapped_read().iter() {
if s.have_capability(p) {
ready_data.push(p);
}
@@ -802,7 +804,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
protocol,
packet_id,
}) => {
match self.handlers.read().unwrap().get(protocol) {
match self.handlers.unwrapped_read().get(protocol) {
None => { warn!(target: "network", "No handler found for protocol: {:?}", protocol) },
Some(_) => packet_data.push((protocol, packet_id, data)),
}
@@ -815,16 +817,16 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
if kill {
self.kill_connection(token, io, true);
}
let handlers = self.handlers.read().unwrap();
let handlers = self.handlers.unwrapped_read();
for p in ready_data {
let h = handlers.get(p).unwrap().clone();
self.stats.inc_sessions();
let reserved = self.reserved_nodes.read().unwrap();
let reserved = self.reserved_nodes.unwrapped_read();
h.connected(&NetworkContext::new(io, p, session.clone(), self.sessions.clone(), &reserved), &token);
}
for (p, packet_id, data) in packet_data {
let h = handlers.get(p).unwrap().clone();
let reserved = self.reserved_nodes.read().unwrap();
let reserved = self.reserved_nodes.unwrapped_read();
h.read(&NetworkContext::new(io, p, session.clone(), self.sessions.clone(), &reserved), &token, packet_id, &data[1..]);
}
}
@@ -840,14 +842,14 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
let mut deregister = false;
let mut expired_session = None;
if let FIRST_SESSION ... LAST_SESSION = token {
let sessions = self.sessions.write().unwrap();
let sessions = self.sessions.unwrapped_write();
if let Some(session) = sessions.get(token).cloned() {
expired_session = Some(session.clone());
let mut s = session.lock().unwrap();
let mut s = session.locked();
if !s.expired() {
if s.is_ready() {
self.num_sessions.fetch_sub(1, AtomicOrdering::SeqCst);
for (p, _) in self.handlers.read().unwrap().iter() {
for (p, _) in self.handlers.unwrapped_read().iter() {
if s.have_capability(p) {
to_disconnect.push(p);
}
@@ -861,12 +863,12 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
}
if let Some(id) = failure_id {
if remote {
self.nodes.write().unwrap().note_failure(&id);
self.nodes.unwrapped_write().note_failure(&id);
}
}
for p in to_disconnect {
let h = self.handlers.read().unwrap().get(p).unwrap().clone();
let reserved = self.reserved_nodes.read().unwrap();
let h = self.handlers.unwrapped_read().get(p).unwrap().clone();
let reserved = self.reserved_nodes.unwrapped_read();
h.disconnected(&NetworkContext::new(io, p, expired_session.clone(), self.sessions.clone(), &reserved), &token);
}
if deregister {
@@ -877,9 +879,9 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
fn update_nodes(&self, io: &IoContext<NetworkIoMessage<Message>>, node_changes: TableUpdates) {
let mut to_remove: Vec<PeerId> = Vec::new();
{
let sessions = self.sessions.write().unwrap();
let sessions = self.sessions.unwrapped_write();
for c in sessions.iter() {
let s = c.lock().unwrap();
let s = c.locked();
if let Some(id) = s.id() {
if node_changes.removed.contains(id) {
to_remove.push(s.token());
@@ -891,7 +893,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
trace!(target: "network", "Removed from node table: {}", i);
self.kill_connection(i, io, false);
}
self.nodes.write().unwrap().update(node_changes);
self.nodes.unwrapped_write().update(node_changes);
}
}
@@ -899,7 +901,7 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
/// Initialize networking
fn initialize(&self, io: &IoContext<NetworkIoMessage<Message>>) {
io.register_timer(IDLE, MAINTENANCE_TIMEOUT).expect("Error registering Network idle timer");
io.register_timer(INIT_PUBLIC, 0).expect("Error registering initialization timer");
io.message(NetworkIoMessage::InitPublicInterface).unwrap_or_else(|e| warn!("Error sending IO notification: {:?}", e));
self.maintain_network(io)
}
@@ -918,7 +920,7 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
match stream {
FIRST_SESSION ... LAST_SESSION => self.session_readable(stream, io),
DISCOVERY => {
let node_changes = { self.discovery.lock().unwrap().as_mut().unwrap().readable(io) };
let node_changes = { self.discovery.locked().as_mut().unwrap().readable(io) };
if let Some(node_changes) = node_changes {
self.update_nodes(io, node_changes);
}
@@ -935,7 +937,7 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
match stream {
FIRST_SESSION ... LAST_SESSION => self.session_writable(stream, io),
DISCOVERY => {
self.discovery.lock().unwrap().as_mut().unwrap().writable(io);
self.discovery.locked().as_mut().unwrap().writable(io);
}
_ => panic!("Received unknown writable token"),
}
@@ -947,15 +949,13 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
}
match token {
IDLE => self.maintain_network(io),
INIT_PUBLIC => self.init_public_interface(io).unwrap_or_else(|e|
warn!("Error initializing public interface: {:?}", e)),
FIRST_SESSION ... LAST_SESSION => self.connection_timeout(token, io),
DISCOVERY_REFRESH => {
self.discovery.lock().unwrap().as_mut().unwrap().refresh();
self.discovery.locked().as_mut().unwrap().refresh();
io.update_registration(DISCOVERY).unwrap_or_else(|e| debug!("Error updating discovery registration: {:?}", e));
},
DISCOVERY_ROUND => {
let node_changes = { self.discovery.lock().unwrap().as_mut().unwrap().round() };
let node_changes = { self.discovery.locked().as_mut().unwrap().round() };
if let Some(node_changes) = node_changes {
self.update_nodes(io, node_changes);
}
@@ -963,13 +963,13 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
},
NODE_TABLE => {
trace!(target: "network", "Refreshing node table");
self.nodes.write().unwrap().clear_useless();
self.nodes.unwrapped_write().clear_useless();
},
_ => match self.timers.read().unwrap().get(&token).cloned() {
Some(timer) => match self.handlers.read().unwrap().get(timer.protocol).cloned() {
_ => match self.timers.unwrapped_read().get(&token).cloned() {
Some(timer) => match self.handlers.unwrapped_read().get(timer.protocol).cloned() {
None => { warn!(target: "network", "No handler found for protocol: {:?}", timer.protocol) },
Some(h) => {
let reserved = self.reserved_nodes.read().unwrap();
let reserved = self.reserved_nodes.unwrapped_read();
h.timeout(&NetworkContext::new(io, timer.protocol, None, self.sessions.clone(), &reserved), timer.token);
}
},
@@ -989,10 +989,10 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
ref versions
} => {
let h = handler.clone();
let reserved = self.reserved_nodes.read().unwrap();
let reserved = self.reserved_nodes.unwrapped_read();
h.initialize(&NetworkContext::new(io, protocol, None, self.sessions.clone(), &reserved));
self.handlers.write().unwrap().insert(protocol, h);
let mut info = self.info.write().unwrap();
self.handlers.unwrapped_write().insert(protocol, h);
let mut info = self.info.unwrapped_write();
for v in versions {
info.capabilities.push(CapabilityInfo { protocol: protocol, version: *v, packet_count:0 });
}
@@ -1003,40 +1003,42 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
ref token,
} => {
let handler_token = {
let mut timer_counter = self.timer_counter.write().unwrap();
let mut timer_counter = self.timer_counter.unwrapped_write();
let counter = &mut *timer_counter;
let handler_token = *counter;
*counter += 1;
handler_token
};
self.timers.write().unwrap().insert(handler_token, ProtocolTimer { protocol: protocol, token: *token });
self.timers.unwrapped_write().insert(handler_token, ProtocolTimer { protocol: protocol, token: *token });
io.register_timer(handler_token, *delay).unwrap_or_else(|e| debug!("Error registering timer {}: {:?}", token, e));
},
NetworkIoMessage::Disconnect(ref peer) => {
let session = { self.sessions.read().unwrap().get(*peer).cloned() };
let session = { self.sessions.unwrapped_read().get(*peer).cloned() };
if let Some(session) = session {
session.lock().unwrap().disconnect(io, DisconnectReason::DisconnectRequested);
session.locked().disconnect(io, DisconnectReason::DisconnectRequested);
}
trace!(target: "network", "Disconnect requested {}", peer);
self.kill_connection(*peer, io, false);
},
NetworkIoMessage::DisablePeer(ref peer) => {
let session = { self.sessions.read().unwrap().get(*peer).cloned() };
let session = { self.sessions.unwrapped_read().get(*peer).cloned() };
if let Some(session) = session {
session.lock().unwrap().disconnect(io, DisconnectReason::DisconnectRequested);
if let Some(id) = session.lock().unwrap().id() {
self.nodes.write().unwrap().mark_as_useless(id)
session.locked().disconnect(io, DisconnectReason::DisconnectRequested);
if let Some(id) = session.locked().id() {
self.nodes.unwrapped_write().mark_as_useless(id)
}
}
trace!(target: "network", "Disabling peer {}", peer);
self.kill_connection(*peer, io, false);
},
NetworkIoMessage::User(ref message) => {
let reserved = self.reserved_nodes.read().unwrap();
for (p, h) in self.handlers.read().unwrap().iter() {
let reserved = self.reserved_nodes.unwrapped_read();
for (p, h) in self.handlers.unwrapped_read().iter() {
h.message(&NetworkContext::new(io, p, None, self.sessions.clone(), &reserved), &message);
}
}
},
NetworkIoMessage::InitPublicInterface =>
self.init_public_interface(io).unwrap_or_else(|e| warn!("Error initializing public interface: {:?}", e)),
_ => {} // ignore others.
}
}
@@ -1044,13 +1046,13 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
fn register_stream(&self, stream: StreamToken, reg: Token, event_loop: &mut EventLoop<IoManager<NetworkIoMessage<Message>>>) {
match stream {
FIRST_SESSION ... LAST_SESSION => {
let session = { self.sessions.read().unwrap().get(stream).cloned() };
let session = { self.sessions.unwrapped_read().get(stream).cloned() };
if let Some(session) = session {
session.lock().unwrap().register_socket(reg, event_loop).expect("Error registering socket");
session.locked().register_socket(reg, event_loop).expect("Error registering socket");
}
}
DISCOVERY => self.discovery.lock().unwrap().as_ref().unwrap().register_socket(event_loop).expect("Error registering discovery socket"),
TCP_ACCEPT => event_loop.register(&*self.tcp_listener.lock().unwrap(), Token(TCP_ACCEPT), EventSet::all(), PollOpt::edge()).expect("Error registering stream"),
DISCOVERY => self.discovery.locked().as_ref().unwrap().register_socket(event_loop).expect("Error registering discovery socket"),
TCP_ACCEPT => event_loop.register(&*self.tcp_listener.locked(), Token(TCP_ACCEPT), EventSet::all(), PollOpt::edge()).expect("Error registering stream"),
_ => warn!("Unexpected stream registration")
}
}
@@ -1058,9 +1060,9 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
fn deregister_stream(&self, stream: StreamToken, event_loop: &mut EventLoop<IoManager<NetworkIoMessage<Message>>>) {
match stream {
FIRST_SESSION ... LAST_SESSION => {
let mut connections = self.sessions.write().unwrap();
let mut connections = self.sessions.unwrapped_write();
if let Some(connection) = connections.get(stream).cloned() {
connection.lock().unwrap().deregister_socket(event_loop).expect("Error deregistering socket");
connection.locked().deregister_socket(event_loop).expect("Error deregistering socket");
connections.remove(stream);
}
}
@@ -1072,13 +1074,13 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
fn update_stream(&self, stream: StreamToken, reg: Token, event_loop: &mut EventLoop<IoManager<NetworkIoMessage<Message>>>) {
match stream {
FIRST_SESSION ... LAST_SESSION => {
let connection = { self.sessions.read().unwrap().get(stream).cloned() };
let connection = { self.sessions.unwrapped_read().get(stream).cloned() };
if let Some(connection) = connection {
connection.lock().unwrap().update_socket(reg, event_loop).expect("Error updating socket");
connection.locked().update_socket(reg, event_loop).expect("Error updating socket");
}
}
DISCOVERY => self.discovery.lock().unwrap().as_ref().unwrap().update_registration(event_loop).expect("Error reregistering discovery socket"),
TCP_ACCEPT => event_loop.reregister(&*self.tcp_listener.lock().unwrap(), Token(TCP_ACCEPT), EventSet::all(), PollOpt::edge()).expect("Error reregistering stream"),
DISCOVERY => self.discovery.locked().as_ref().unwrap().update_registration(event_loop).expect("Error reregistering discovery socket"),
TCP_ACCEPT => event_loop.reregister(&*self.tcp_listener.locked(), Token(TCP_ACCEPT), EventSet::all(), PollOpt::edge()).expect("Error reregistering stream"),
_ => warn!("Unexpected stream update")
}
}

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

@@ -17,6 +17,7 @@
use std::sync::*;
use error::*;
use panics::*;
use misc::RwLockable;
use network::{NetworkProtocolHandler, NetworkConfiguration};
use network::error::NetworkError;
use network::host::{Host, NetworkIoMessage, ProtocolId};
@@ -85,19 +86,19 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
/// Returns external url if available.
pub fn external_url(&self) -> Option<String> {
let host = self.host.read().unwrap();
let host = self.host.unwrapped_read();
host.as_ref().and_then(|h| h.external_url())
}
/// Returns external url if available.
pub fn local_url(&self) -> Option<String> {
let host = self.host.read().unwrap();
let host = self.host.unwrapped_read();
host.as_ref().map(|h| h.local_url())
}
/// Start network IO
pub fn start(&self) -> Result<(), UtilError> {
let mut host = self.host.write().unwrap();
let mut host = self.host.unwrapped_write();
if host.is_none() {
let h = Arc::new(try!(Host::new(self.config.clone(), self.stats.clone())));
try!(self.io_service.register_handler(h.clone()));
@@ -108,7 +109,7 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
/// Stop network IO
pub fn stop(&self) -> Result<(), UtilError> {
let mut host = self.host.write().unwrap();
let mut host = self.host.unwrapped_write();
if let Some(ref host) = *host {
let io = IoContext::new(self.io_service.channel(), 0); //TODO: take token id from host
try!(host.stop(&io));
@@ -119,7 +120,7 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
/// Try to add a reserved peer.
pub fn add_reserved_peer(&self, peer: &str) -> Result<(), UtilError> {
let host = self.host.read().unwrap();
let host = self.host.unwrapped_read();
if let Some(ref host) = *host {
host.add_reserved_node(peer)
} else {
@@ -129,7 +130,7 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
/// Try to remove a reserved peer.
pub fn remove_reserved_peer(&self, peer: &str) -> Result<(), UtilError> {
let host = self.host.read().unwrap();
let host = self.host.unwrapped_read();
if let Some(ref host) = *host {
host.remove_reserved_node(peer)
} else {
@@ -139,7 +140,7 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
/// Set the non-reserved peer mode.
pub fn set_non_reserved_mode(&self, mode: ::network::NonReservedPeerMode) {
let host = self.host.read().unwrap();
let host = self.host.unwrapped_read();
if let Some(ref host) = *host {
let io_ctxt = IoContext::new(self.io_service.channel(), 0);
host.set_non_reserved_mode(mode, &io_ctxt);

5
util/src/network/tests.rs
View File

@@ -18,6 +18,7 @@ use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
use std::thread;
use std::time::*;
use common::*;
use misc::*;
use network::*;
use io::TimerToken;
use crypto::KeyPair;
@@ -51,7 +52,7 @@ impl TestProtocol {
}
pub fn got_packet(&self) -> bool {
self.packet.lock().unwrap().deref()[..] == b"hello"[..]
self.packet.locked().deref()[..] == b"hello"[..]
}
pub fn got_timeout(&self) -> bool {
@@ -70,7 +71,7 @@ impl NetworkProtocolHandler<TestProtocolMessage> for TestProtocol {
fn read(&self, _io: &NetworkContext<TestProtocolMessage>, _peer: &PeerId, packet_id: u8, data: &[u8]) {
assert_eq!(packet_id, 33);
self.packet.lock().unwrap().extend(data);
self.packet.locked().extend(data);
}
fn connected(&self, io: &NetworkContext<TestProtocolMessage>, peer: &PeerId) {

42
util/src/nibblevec.rs
View File

@@ -1,16 +1,33 @@
// Copyright 2015, 2016 Ethcore (UK) Ltd.
// This file is part of Parity.
// Parity is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
// Parity is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
// You should have received a copy of the GNU General Public License
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
//! An owning, nibble-oriented byte vector.
use ::NibbleSlice;
#[derive(PartialEq, Eq, PartialOrd, Ord, Debug)]
/// Owning, nibble-oriented byte vector. Counterpart to NibbleSlice.
#[derive(Default, PartialEq, Eq, PartialOrd, Ord, Debug)]
/// Owning, nibble-oriented byte vector. Counterpart to `NibbleSlice`.
pub struct NibbleVec {
inner: Vec<u8>,
len: usize,
}
impl NibbleVec {
/// Make a new NibbleVec
/// Make a new `NibbleVec`
pub fn new() -> Self {
NibbleVec {
inner: Vec::new(),
@@ -18,7 +35,7 @@ impl NibbleVec {
}
}
/// Make a NibbleVec with capacity for `n` nibbles.
/// Make a `NibbleVec` with capacity for `n` nibbles.
pub fn with_capacity(n: usize) -> Self {
NibbleVec {
inner: Vec::with_capacity((n / 2) + (n % 2)),
@@ -26,10 +43,13 @@ impl NibbleVec {
}
}
/// Length of the NibbleVec
/// Length of the `NibbleVec`
pub fn len(&self) -> usize { self.len }
/// Capacity of the NibbleVec.
/// Retrurns true if `NibbleVec` has zero length
pub fn is_empty(&self) -> bool { self.len == 0 }
/// Capacity of the `NibbleVec`.
pub fn capacity(&self) -> usize { self.inner.capacity() * 2 }
/// Try to get the nibble at the given offset.
@@ -41,7 +61,7 @@ impl NibbleVec {
}
}
/// Push a nibble onto the NibbleVec. Ignores the high 4 bits.
/// Push a nibble onto the `NibbleVec`. Ignores the high 4 bits.
pub fn push(&mut self, nibble: u8) {
let nibble = nibble & 0x0F;
@@ -54,9 +74,9 @@ impl NibbleVec {
self.len += 1;
}
/// Try to pop a nibble off the NibbleVec. Fails if len == 0.
/// Try to pop a nibble off the `NibbleVec`. Fails if len == 0.
pub fn pop(&mut self) -> Option<u8> {
if self.len == 0 {
if self.is_empty() {
return None;
}
@@ -72,7 +92,7 @@ impl NibbleVec {
Some(nibble)
}
/// Try to treat this NibbleVec as a NibbleSlice. Works only if len is even.
/// Try to treat this `NibbleVec` as a `NibbleSlice`. Works only if len is even.
pub fn as_nibbleslice(&self) -> Option<NibbleSlice> {
if self.len % 2 == 0 {
Some(NibbleSlice::new(self.inner()))
@@ -127,4 +147,4 @@ mod tests {
let v2: NibbleVec = v.as_nibbleslice().unwrap().into();
assert_eq!(v, v2);
}
}
}

26
util/src/panics.rs
View File

@@ -20,6 +20,7 @@ use std::thread;
use std::ops::DerefMut;
use std::sync::{Arc, Mutex};
use std::default::Default;
use misc::Lockable;
/// Thread-safe closure for handling possible panics
pub trait OnPanicListener: Send + Sync + 'static {
@@ -88,7 +89,7 @@ impl PanicHandler {
/// Notifies all listeners in case there is a panic.
/// You should use `catch_panic` instead of calling this method explicitly.
pub fn notify_all(&self, r: String) {
let mut listeners = self.listeners.lock().unwrap();
let mut listeners = self.listeners.locked();
for listener in listeners.deref_mut() {
listener.call(&r);
}
@@ -97,7 +98,7 @@ impl PanicHandler {
impl MayPanic for PanicHandler {
fn on_panic<F>(&self, closure: F) where F: OnPanicListener {
self.listeners.lock().unwrap().push(Box::new(closure));
self.listeners.locked().push(Box::new(closure));
}
}
@@ -119,46 +120,49 @@ impl<F> OnPanicListener for F
#[ignore] // panic forwarding doesnt work on the same thread in beta
fn should_notify_listeners_about_panic () {
use std::sync::RwLock;
use misc::RwLockable;
// given
let invocations = Arc::new(RwLock::new(vec![]));
let i = invocations.clone();
let p = PanicHandler::new();
p.on_panic(move |t| i.write().unwrap().push(t));
p.on_panic(move |t| i.unwrapped_write().push(t));
// when
p.catch_panic(|| panic!("Panic!")).unwrap_err();
// then
assert!(invocations.read().unwrap()[0] == "Panic!");
assert!(invocations.unwrapped_read()[0] == "Panic!");
}
#[test]
#[ignore] // panic forwarding doesnt work on the same thread in beta
fn should_notify_listeners_about_panic_when_string_is_dynamic () {
use std::sync::RwLock;
use misc::RwLockable;
// given
let invocations = Arc::new(RwLock::new(vec![]));
let i = invocations.clone();
let p = PanicHandler::new();
p.on_panic(move |t| i.write().unwrap().push(t));
p.on_panic(move |t| i.unwrapped_write().push(t));
// when
p.catch_panic(|| panic!("Panic: {}", 1)).unwrap_err();
// then
assert!(invocations.read().unwrap()[0] == "Panic: 1");
assert!(invocations.unwrapped_read()[0] == "Panic: 1");
}
#[test]
fn should_notify_listeners_about_panic_in_other_thread () {
use std::thread;
use std::sync::RwLock;
use misc::RwLockable;
// given
let invocations = Arc::new(RwLock::new(vec![]));
let i = invocations.clone();
let p = PanicHandler::new();
p.on_panic(move |t| i.write().unwrap().push(t));
p.on_panic(move |t| i.unwrapped_write().push(t));
// when
let t = thread::spawn(move ||
@@ -167,18 +171,20 @@ fn should_notify_listeners_about_panic_in_other_thread () {
t.join().unwrap_err();
// then
assert!(invocations.read().unwrap()[0] == "Panic!");
assert!(invocations.unwrapped_read()[0] == "Panic!");
}
#[test]
#[ignore] // panic forwarding doesnt work on the same thread in beta
fn should_forward_panics () {
use std::sync::RwLock;
use misc::RwLockable;
// given
let invocations = Arc::new(RwLock::new(vec![]));
let i = invocations.clone();
let p = PanicHandler::new_in_arc();
p.on_panic(move |t| i.write().unwrap().push(t));
p.on_panic(move |t| i.unwrapped_write().push(t));
let p2 = PanicHandler::new();
p.forward_from(&p2);
@@ -187,5 +193,5 @@ use std::sync::RwLock;
p2.catch_panic(|| panic!("Panic!")).unwrap_err();
// then
assert!(invocations.read().unwrap()[0] == "Panic!");
assert!(invocations.unwrapped_read()[0] == "Panic!");
}