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

Merge branch 'master' into rocksdb-flush-limit

Browse Source
This commit is contained in:
NikVolf
2016-06-21 17:32:07 +03:00
parent 9132895d0e b2891fcdda
commit de079ebe31
43 changed files with 727 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
util/res/geth_keystore/UTC--2016-02-17T09-20-45.721400158Z--3f49624084b67849c7b4e805c5988c21a430f9d9
View File

@@ -1,21 +0,0 @@
{
"address": "3f49624084b67849c7b4e805c5988c21a430f9d9",
"Crypto": {
"cipher": "aes-128-ctr",
"ciphertext": "9f27e3dd4fc73e7103ed61e5493662189a3eb52223ae49e3d1deacc04c889eae",
"cipherparams": {
"iv": "457494bf05f2618c397dc74dbb5181c0"
},
"kdf": "scrypt",
"kdfparams": {
"dklen": 32,
"n": 262144,
"p": 1,
"r": 8,
"salt": "db14edb18c41ee7f5ec4397df89c3a2ae4d0af60884c52bb54ce490574f8df33"
},
"mac": "572d24532438d31fdf513c744a3ff26c933ffda5744ee42bc71661cbe3f2112e"
},
"id": "62a0ad73-556d-496a-8e1c-0783d30d3ace",
"version": 3
}

21
util/res/geth_keystore/UTC--2016-02-20T09-33-03.984382741Z--5ba4dcf897e97c2bdf8315b9ef26c13c085988cf
View File

@@ -1,21 +0,0 @@
{
"address": "5ba4dcf897e97c2bdf8315b9ef26c13c085988cf",
"Crypto": {
"cipher": "aes-128-ctr",
"ciphertext": "d4a08ec930163778273920f6ad1d49b71836337be6fd9863993ac700a612fddd",
"cipherparams": {
"iv": "89ce5ec129fc27cd5bcbeb8c92bdad50"
},
"kdf": "scrypt",
"kdfparams": {
"dklen": 32,
"n": 262144,
"p": 1,
"r": 8,
"salt": "612ab108dc37e69ee8af37a7b24bf7f2234086d7bbf945bacdeccce331f7f84a"
},
"mac": "4152caa7444e06784223d735cea80cd2690b4c587ad8db3d5529442227b25695"
},
"id": "35086353-fb12-4029-b56b-033cd61ce35b",
"version": 3
}

1
util/res/geth_keystore/UTC--2016-04-03T08-58-49.834202900Z--63121b431a52f8043c16fcf0d1df9cb7b5f66649
View File

@@ -1 +0,0 @@
{"address":"63121b431a52f8043c16fcf0d1df9cb7b5f66649","crypto":{"cipher":"aes-128-ctr","ciphertext":"1dd21926c644b9983916d646f3a4f2c7f9362f7e1c9fb1abcb42494dae06fa01","cipherparams":{"iv":"c52c6ee66d89a7aa8c6839f4b6ed29c8"},"kdf":"scrypt","kdfparams":{"dklen":32,"n":262144,"p":1,"r":8,"salt":"96f17c17bbf48db2dc4da00b3e7decce8e21f44a5d7963dadeeff70e1d38ad75"},"mac":"f279f3444585c2817701225e2196c1176386ad549ebaec2bcc4f94f309727fe6"},"id":"15e49cd2-51fb-4316-ba46-c3cf8db4ae44","version":3}

21
util/res/pat/p1.json
View File

@@ -1,21 +0,0 @@
{
"address": "3f49624084b67849c7b4e805c5988c21a430f9d9",
"Crypto": {
"cipher": "aes-128-ctr",
"ciphertext": "9f27e3dd4fc73e7103ed61e5493662189a3eb52223ae49e3d1deacc04c889eae",
"cipherparams": {
"iv": "457494bf05f2618c397dc74dbb5181c0"
},
"kdf": "scrypt",
"kdfparams": {
"dklen": 32,
"n": 262144,
"p": 1,
"r": 8,
"salt": "db14edb18c41ee7f5ec4397df89c3a2ae4d0af60884c52bb54ce490574f8df33"
},
"mac": "572d24532438d31fdf513c744a3ff26c933ffda5744ee42bc71661cbe3f2112e"
},
"id": "62a0ad73-556d-496a-8e1c-0783d30d3ace",
"version": 3
}

21
util/res/pat/p2.json
View File

@@ -1,21 +0,0 @@
{
"address": "5ba4dcf897e97c2bdf8315b9ef26c13c085988cf",
"Crypto": {
"cipher": "aes-128-ctr",
"ciphertext": "d4a08ec930163778273920f6ad1d49b71836337be6fd9863993ac700a612fddd",
"cipherparams": {
"iv": "89ce5ec129fc27cd5bcbeb8c92bdad50"
},
"kdf": "scrypt",
"kdfparams": {
"dklen": 32,
"n": 262144,
"p": 1,
"r": 8,
"salt": "612ab108dc37e69ee8af37a7b24bf7f2234086d7bbf945bacdeccce331f7f84a"
},
"mac": "4152caa7444e06784223d735cea80cd2690b4c587ad8db3d5529442227b25695"
},
"id": "35086353-fb12-4029-b56b-033cd61ce35b",
"version": 3
}

2
util/src/lib.rs
View File

@@ -154,6 +154,7 @@ pub mod panics;
pub mod table;
pub mod network_settings;
pub mod path;
mod timer;
pub use common::*;
pub use misc::*;
@@ -175,6 +176,7 @@ pub use network::*;
pub use io::*;
pub use log::*;
pub use kvdb::*;
pub use timer::*;
#[cfg(test)]
mod tests {

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

@@ -14,9 +14,9 @@
// You should have received a copy of the GNU General Public License
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
use std::net::{SocketAddr};
use std::collections::{HashMap};
use std::str::{FromStr};
use std::net::SocketAddr;
use std::collections::{HashMap, HashSet};
use std::str::FromStr;
use std::sync::*;
use std::sync::atomic::{AtomicUsize, AtomicBool, Ordering as AtomicOrdering};
use std::ops::*;
@@ -35,7 +35,7 @@ use rlp::*;
use network::session::{Session, SessionData};
use error::*;
use io::*;
use network::{NetworkProtocolHandler, PROTOCOL_VERSION};
use network::{NetworkProtocolHandler, NonReservedPeerMode, PROTOCOL_VERSION};
use network::node_table::*;
use network::stats::NetworkStats;
use network::error::{NetworkError, DisconnectReason};
@@ -65,8 +65,6 @@ pub struct NetworkConfiguration {
pub nat_enabled: bool,
/// Enable discovery
pub discovery_enabled: bool,
/// Pin to reserved nodes only
pub reserved_only: bool,
/// List of initial node addresses
pub boot_nodes: Vec<String>,
/// Use provided node key instead of default
@@ -75,6 +73,8 @@ pub struct NetworkConfiguration {
pub ideal_peers: u32,
/// List of reserved node addresses.
pub reserved_nodes: Vec<String>,
/// The non-reserved peer mode.
pub non_reserved_mode: NonReservedPeerMode,
}
impl Default for NetworkConfiguration {
@@ -93,11 +93,11 @@ impl NetworkConfiguration {
udp_port: None,
nat_enabled: true,
discovery_enabled: true,
reserved_only: false,
boot_nodes: Vec::new(),
use_secret: None,
ideal_peers: 25,
reserved_nodes: Vec::new(),
non_reserved_mode: NonReservedPeerMode::Accept,
}
}
@@ -191,13 +191,15 @@ pub struct NetworkContext<'s, Message> where Message: Send + Sync + Clone + 'sta
sessions: Arc<RwLock<Slab<SharedSession>>>,
session: Option<SharedSession>,
session_id: Option<StreamToken>,
reserved_peers: &'s HashSet<NodeId>,
}
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 IoContext<NetworkIoMessage<Message>>,
protocol: ProtocolId,
session: Option<SharedSession>, sessions: Arc<RwLock<Slab<SharedSession>>>) -> NetworkContext<'s, Message> {
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());
NetworkContext {
io: io,
@@ -205,6 +207,7 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
session_id: id,
session: session,
sessions: sessions,
reserved_peers: reserved_peers,
}
}
@@ -237,7 +240,7 @@ impl<'s, Message> NetworkContext<'s, Message> where Message: Send + Sync + Clone
self.io.message(NetworkIoMessage::User(msg));
}
/// Send an IO message
/// Get an IoChannel.
pub fn io_channel(&self) -> IoChannel<NetworkIoMessage<Message>> {
self.io.channel()
}
@@ -335,7 +338,7 @@ pub struct Host<Message> where Message: Send + Sync + Clone {
timers: RwLock<HashMap<TimerToken, ProtocolTimer>>,
timer_counter: RwLock<usize>,
stats: Arc<NetworkStats>,
pinned_nodes: Vec<NodeId>,
reserved_nodes: RwLock<HashSet<NodeId>>,
num_sessions: AtomicUsize,
stopping: AtomicBool,
}
@@ -390,28 +393,28 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
timers: RwLock::new(HashMap::new()),
timer_counter: RwLock::new(USER_TIMER),
stats: stats,
pinned_nodes: Vec::new(),
reserved_nodes: RwLock::new(HashSet::new()),
num_sessions: AtomicUsize::new(0),
stopping: AtomicBool::new(false),
};
for n in boot_nodes {
// don't pin boot nodes.
host.add_node(&n, false);
host.add_node(&n);
}
for n in reserved_nodes {
host.add_node(&n, true);
if let Err(e) = host.add_reserved_node(&n) {
debug!(target: "network", "Error parsing node id: {}: {:?}", n, e);
}
}
Ok(host)
}
pub fn add_node(&mut self, id: &str, pin: bool) {
pub fn add_node(&mut self, id: &str) {
match Node::from_str(id) {
Err(e) => { debug!(target: "network", "Could not add node {}: {:?}", id, e); },
Ok(n) => {
let entry = NodeEntry { endpoint: n.endpoint.clone(), id: n.id.clone() };
if pin { self.pinned_nodes.push(n.id.clone()) }
self.nodes.write().unwrap().add_node(n);
if let Some(ref mut discovery) = *self.discovery.lock().unwrap() {
@@ -421,6 +424,56 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
}
}
pub fn add_reserved_node(&self, id: &str) -> Result<(), UtilError> {
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());
if let Some(ref mut discovery) = *self.discovery.lock().unwrap() {
discovery.add_node(entry);
}
Ok(())
}
pub fn set_non_reserved_mode(&self, mode: NonReservedPeerMode, io: &IoContext<NetworkIoMessage<Message>>) {
let mut info = self.info.write().unwrap();
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 mut to_kill = Vec::new();
for e in self.sessions.write().unwrap().iter_mut() {
let mut s = e.lock().unwrap();
{
let id = s.id();
if id.is_some() && reserved.contains(id.unwrap()) {
continue;
}
}
s.disconnect(io, DisconnectReason::ClientQuit);
to_kill.push(s.token());
}
for p in to_kill {
trace!(target: "network", "Disconnecting on reserved-only mode: {}", p);
self.kill_connection(p, io, false);
}
}
}
}
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);
Ok(())
}
pub fn client_version() -> String {
version()
}
@@ -483,7 +536,7 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
// Initialize discovery.
let discovery = {
let info = self.info.read().unwrap();
if info.config.discovery_enabled && !info.config.reserved_only {
if info.config.discovery_enabled && info.config.non_reserved_mode == NonReservedPeerMode::Accept {
Some(Discovery::new(&info.keys, public_endpoint.address.clone(), public_endpoint, DISCOVERY))
} else { None }
};
@@ -540,17 +593,26 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
}
fn connect_peers(&self, io: &IoContext<NetworkIoMessage<Message>>) {
if self.info.read().unwrap().capabilities.is_empty() {
return;
}
let ideal_peers = { self.info.read().unwrap().config.ideal_peers };
let pin = { self.info.read().unwrap().config.reserved_only };
let session_count = self.session_count();
if session_count >= ideal_peers as usize + self.pinned_nodes.len() {
// check if all pinned nodes are connected.
if self.pinned_nodes.iter().all(|n| self.have_session(n) && self.connecting_to(n)) {
let (ideal_peers, mut pin) = {
let info = self.info.read().unwrap();
if info.capabilities.is_empty() {
return;
}
let config = &info.config;
(config.ideal_peers, config.non_reserved_mode == NonReservedPeerMode::Deny)
};
let session_count = self.session_count();
let reserved_nodes = self.reserved_nodes.read().unwrap();
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)) {
return;
}
// if not, only attempt connect to reserved peers
pin = true;
}
let handshake_count = self.handshake_count();
@@ -562,7 +624,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 = self.pinned_nodes.iter().cloned().chain(if !pin {
let nodes = reserved_nodes.iter().cloned().chain(if !pin {
self.nodes.read().unwrap().nodes()
} else {
Vec::new()
@@ -684,13 +746,10 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
match s.readable(io, &self.info.read().unwrap()) {
Err(e) => {
trace!(target: "network", "Session read error: {}:{:?} ({:?}) {:?}", token, s.id(), s.remote_addr(), e);
match e {
UtilError::Network(NetworkError::Disconnect(DisconnectReason::IncompatibleProtocol)) => {
if let Some(id) = s.id() {
self.nodes.write().unwrap().mark_as_useless(id);
}
if let UtilError::Network(NetworkError::Disconnect(DisconnectReason::IncompatibleProtocol)) = e {
if let Some(id) = s.id() {
self.nodes.write().unwrap().mark_as_useless(id);
}
_ => (),
}
kill = true;
break;
@@ -699,11 +758,20 @@ 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 ideal_peers = { self.info.read().unwrap().config.ideal_peers };
if session_count >= ideal_peers as usize {
s.disconnect(io, DisconnectReason::TooManyPeers);
return;
let reserved_nodes = self.reserved_nodes.read().unwrap();
let (ideal_peers, reserved_only) = {
let info = self.info.read().unwrap();
(info.config.ideal_peers, info.config.non_reserved_mode == NonReservedPeerMode::Deny)
};
if session_count >= ideal_peers as usize || reserved_only {
// only proceed if the connecting peer is reserved.
if !reserved_nodes.contains(s.id().unwrap()) {
s.disconnect(io, DisconnectReason::TooManyPeers);
return;
}
}
// 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() } };
@@ -738,14 +806,17 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
if kill {
self.kill_connection(token, io, true);
}
let handlers = self.handlers.read().unwrap();
for p in ready_data {
let h = self.handlers.read().unwrap().get(p).unwrap().clone();
let h = handlers.get(p).unwrap().clone();
self.stats.inc_sessions();
h.connected(&NetworkContext::new(io, p, session.clone(), self.sessions.clone()), &token);
let reserved = self.reserved_nodes.read().unwrap();
h.connected(&NetworkContext::new(io, p, session.clone(), self.sessions.clone(), &reserved), &token);
}
for (p, packet_id, data) in packet_data {
let h = self.handlers.read().unwrap().get(p).unwrap().clone();
h.read(&NetworkContext::new(io, p, session.clone(), self.sessions.clone()), &token, packet_id, &data[1..]);
let h = handlers.get(p).unwrap().clone();
let reserved = self.reserved_nodes.read().unwrap();
h.read(&NetworkContext::new(io, p, session.clone(), self.sessions.clone(), &reserved), &token, packet_id, &data[1..]);
}
}
@@ -786,7 +857,8 @@ impl<Message> Host<Message> where Message: Send + Sync + Clone {
}
for p in to_disconnect {
let h = self.handlers.read().unwrap().get(p).unwrap().clone();
h.disconnected(&NetworkContext::new(io, p, expired_session.clone(), self.sessions.clone()), &token);
let reserved = self.reserved_nodes.read().unwrap();
h.disconnected(&NetworkContext::new(io, p, expired_session.clone(), self.sessions.clone(), &reserved), &token);
}
if deregister {
io.deregister_stream(token).unwrap_or_else(|e| debug!("Error deregistering stream: {:?}", e));
@@ -889,7 +961,10 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
_ => match self.timers.read().unwrap().get(&token).cloned() {
Some(timer) => match self.handlers.read().unwrap().get(timer.protocol).cloned() {
None => { warn!(target: "network", "No handler found for protocol: {:?}", timer.protocol) },
Some(h) => { h.timeout(&NetworkContext::new(io, timer.protocol, None, self.sessions.clone()), timer.token); }
Some(h) => {
let reserved = self.reserved_nodes.read().unwrap();
h.timeout(&NetworkContext::new(io, timer.protocol, None, self.sessions.clone(), &reserved), timer.token);
}
},
None => { warn!("Unknown timer token: {}", token); } // timer is not registerd through us
}
@@ -907,7 +982,8 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
ref versions
} => {
let h = handler.clone();
h.initialize(&NetworkContext::new(io, protocol, None, self.sessions.clone()));
let reserved = self.reserved_nodes.read().unwrap();
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();
for v in versions {
@@ -949,8 +1025,9 @@ impl<Message> IoHandler<NetworkIoMessage<Message>> for Host<Message> where Messa
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() {
h.message(&NetworkContext::new(io, p, None, self.sessions.clone()), &message);
h.message(&NetworkContext::new(io, p, None, self.sessions.clone(), &reserved), &message);
}
}
}

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

@@ -14,8 +14,8 @@
// You should have received a copy of the GNU General Public License
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
//! Network and general IO module.
//!
//! Network and general IO module.
//!
//! Example usage for craeting a network service and adding an IO handler:
//!
//! ```rust
@@ -112,3 +112,22 @@ pub trait NetworkProtocolHandler<Message>: Sync + Send where Message: Send + Syn
fn message(&self, _io: &NetworkContext<Message>, _message: &Message) {}
}
/// Non-reserved peer modes.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum NonReservedPeerMode {
/// Accept them. This is the default.
Accept,
/// Deny them.
Deny,
}
impl NonReservedPeerMode {
/// Attempt to parse the peer mode from a string.
pub fn parse(s: &str) -> Option<Self> {
match s {
"accept" => Some(NonReservedPeerMode::Accept),
"deny" => Some(NonReservedPeerMode::Deny),
_ => None,
}
}
}

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

@@ -18,9 +18,9 @@ use std::sync::*;
use error::*;
use panics::*;
use network::{NetworkProtocolHandler, NetworkConfiguration};
use network::error::{NetworkError};
use network::error::NetworkError;
use network::host::{Host, NetworkIoMessage, ProtocolId};
use network::stats::{NetworkStats};
use network::stats::NetworkStats;
use io::*;
/// IO Service with networking
@@ -111,6 +111,35 @@ impl<Message> NetworkService<Message> where Message: Send + Sync + Clone + 'stat
*host = None;
Ok(())
}
/// Try to add a reserved peer.
pub fn add_reserved_peer(&self, peer: &str) -> Result<(), UtilError> {
let host = self.host.read().unwrap();
if let Some(ref host) = *host {
host.add_reserved_node(peer)
} else {
Ok(())
}
}
/// Try to remove a reserved peer.
pub fn remove_reserved_peer(&self, peer: &str) -> Result<(), UtilError> {
let host = self.host.read().unwrap();
if let Some(ref host) = *host {
host.remove_reserved_node(peer)
} else {
Ok(())
}
}
/// Set the non-reserved peer mode.
pub fn set_non_reserved_mode(&self, mode: ::network::NonReservedPeerMode) {
let host = self.host.read().unwrap();
if let Some(ref host) = *host {
let io_ctxt = IoContext::new(self.io_service.channel(), 0);
host.set_non_reserved_mode(mode, &io_ctxt);
}
}
}
impl<Message> MayPanic for NetworkService<Message> where Message: Send + Sync + Clone + 'static {

51
util/src/timer.rs Normal file
View File

@@ -0,0 +1,51 @@
// 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/>.
//! Performance timer with logging
use time::precise_time_ns;
/// Performance timer with logging. Starts measuring time in the constructor, prints
/// elapsed time in the destructor or when `stop` is called.
pub struct PerfTimer {
name: &'static str,
start: u64,
stopped: bool,
}
impl PerfTimer {
/// Create an instance with given name.
pub fn new(name: &'static str) -> PerfTimer {
PerfTimer {
name: name,
start: precise_time_ns(),
stopped: false,
}
}
/// Stop the timer and print elapsed time on trace level with `perf` target.
pub fn stop(&mut self) {
if !self.stopped {
trace!(target: "perf", "{}: {:.2}ms", self.name, (precise_time_ns() - self.start) as f32 / 1000_000.0);
self.stopped = true;
}
}
}
impl Drop for PerfTimer {
fn drop(&mut self) {
self.stop()
}
}