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

[Beta] Backports (#8136)

Browse Source 
* Support parity protocol. (#8035)

* updater: apply exponential backoff after download failure (#8059)

* updater: apply exponential backoff after download failure

* updater: reset backoff on new release

* Max code size on Kovan (#8067)

* Enable code size limit on kovan

* Fix formatting.

* Limit incoming connections.  (#8060)

* Limit ingress connections
* Optimized handshakes logging

* WASM libraries bump (#7970)

* update wasmi, parity-wasm, wasm-utils to latest version

* Update to new wasmi & error handling

* also utilize new stack limiter

* fix typo

* replace dependency url

* Cargo.lock update

* add some dos protection (#8084)

* revert removing blooms (#8066)

* Revert "fix traces, removed bloomchain crate, closes #7228, closes #7167"

This reverts commit 1bf62038678295e5586f02a38a0c5aab9a9efe62.

* Revert "fixed broken logs (#7934)"

This reverts commit f8a2e53f3e.

* fixed broken logs

* bring back old lock order

* remove migration v13

* revert CURRENT_VERSION to 12 in migration.rs

* more dos protection (#8104)

* Const time comparison (#8113)

* Use `subtle::slices_equal` for constant time comparison.

Also update the existing version of subtle in `ethcrypto` from
0.1 to 0.5

* Test specifically for InvalidPassword error.

* fix trace filter returning returning unrelated reward calls, closes #8070 (#8098)

* network: init discovery using healthy nodes (#8061)

* network: init discovery using healthy nodes

* network: fix style grumble

* network: fix typo

* Postpone Kovan hard fork (#8137)

* ethcore: postpone Kovan hard fork

* util: update version fork metadata

* Disable UI by default. (#8105)

* dapps: update parity-ui dependencies (#8160)
This commit is contained in:
André Silva
2018-03-19 20:40:56 +00:00
committed by Tomasz Drwięga
parent 2403fc52c1
commit 610f6f1425
82 changed files with 2698 additions and 364 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -18,9 +18,9 @@ use std::net::{SocketAddr, SocketAddrV4, Ipv4Addr};
use std::collections::{HashMap, HashSet};
use std::str::FromStr;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, AtomicBool, Ordering as AtomicOrdering};
use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
use std::ops::*;
use std::cmp::min;
use std::cmp::{min, max};
use std::path::{Path, PathBuf};
use std::io::{Read, Write, self};
use std::fs;
@@ -383,7 +383,6 @@ pub struct Host {
timer_counter: RwLock<usize>,
stats: Arc<NetworkStats>,
reserved_nodes: RwLock<HashSet<NodeId>>,
num_sessions: AtomicUsize,
stopping: AtomicBool,
filter: Option<Arc<ConnectionFilter>>,
}
@@ -440,7 +439,6 @@ impl Host {
timer_counter: RwLock::new(USER_TIMER),
stats: stats,
reserved_nodes: RwLock::new(HashSet::new()),
num_sessions: AtomicUsize::new(0),
stopping: AtomicBool::new(false),
filter: filter,
};
@@ -495,7 +493,7 @@ impl Host {
// disconnect all non-reserved peers here.
let reserved: HashSet<NodeId> = self.reserved_nodes.read().clone();
let mut to_kill = Vec::new();
for e in self.sessions.write().iter_mut() {
for e in self.sessions.read().iter() {
let mut s = e.lock();
{
let id = s.id();
@@ -535,7 +533,7 @@ impl Host {
pub fn stop(&self, io: &IoContext<NetworkIoMessage>) -> Result<(), Error> {
self.stopping.store(true, AtomicOrdering::Release);
let mut to_kill = Vec::new();
for e in self.sessions.write().iter_mut() {
for e in self.sessions.read().iter() {
let mut s = e.lock();
s.disconnect(io, DisconnectReason::ClientQuit);
to_kill.push(s.token());
@@ -605,8 +603,8 @@ impl Host {
};
if let Some(mut discovery) = discovery {
discovery.init_node_list(self.nodes.read().unordered_entries());
discovery.add_node_list(self.nodes.read().unordered_entries());
discovery.init_node_list(self.nodes.read().entries());
discovery.add_node_list(self.nodes.read().entries());
*self.discovery.lock() = Some(discovery);
io.register_stream(DISCOVERY)?;
io.register_timer(DISCOVERY_REFRESH, DISCOVERY_REFRESH_TIMEOUT)?;
@@ -626,22 +624,28 @@ impl Host {
self.sessions.read().iter().any(|e| e.lock().info.id == Some(id.clone()))
}
fn session_count(&self) -> usize {
self.num_sessions.load(AtomicOrdering::Relaxed)
// returns (handshakes, egress, ingress)
fn session_count(&self) -> (usize, usize, usize) {
let mut handshakes = 0;
let mut egress = 0;
let mut ingress = 0;
for s in self.sessions.read().iter() {
match s.try_lock() {
Some(ref s) if s.is_ready() && s.info.originated => egress += 1,
Some(ref s) if s.is_ready() && !s.info.originated => ingress += 1,
_ => handshakes +=1,
}
}
(handshakes, egress, ingress)
}
fn connecting_to(&self, id: &NodeId) -> bool {
self.sessions.read().iter().any(|e| e.lock().id() == Some(id))
}
fn handshake_count(&self) -> usize {
// session_count < total_count is possible because of the data race.
self.sessions.read().count().saturating_sub(self.session_count())
}
fn keep_alive(&self, io: &IoContext<NetworkIoMessage>) {
let mut to_kill = Vec::new();
for e in self.sessions.write().iter_mut() {
for e in self.sessions.read().iter() {
let mut s = e.lock();
if !s.keep_alive(io) {
s.disconnect(io, DisconnectReason::PingTimeout);
@@ -665,9 +669,9 @@ impl Host {
(config.min_peers, config.non_reserved_mode == NonReservedPeerMode::Deny, config.max_handshakes as usize, config.ip_filter.clone(), info.id().clone())
};
let session_count = self.session_count();
let (handshake_count, egress_count, ingress_count) = self.session_count();
let reserved_nodes = self.reserved_nodes.read();
if session_count >= min_peers as usize + reserved_nodes.len() {
if egress_count + ingress_count >= min_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;
@@ -677,7 +681,6 @@ impl Host {
pin = true;
}
let handshake_count = self.handshake_count();
// allow 16 slots for incoming connections
if handshake_count >= max_handshakes {
return;
@@ -702,7 +705,7 @@ impl Host {
self.connect_peer(&id, io);
started += 1;
}
debug!(target: "network", "Connecting peers: {} sessions, {} pending, {} started", self.session_count(), self.handshake_count(), started);
debug!(target: "network", "Connecting peers: {} sessions, {} pending + {} started", egress_count + ingress_count, handshake_count, started);
}
fn connect_peer(&self, id: &NodeId, io: &IoContext<NetworkIoMessage>) {
@@ -812,11 +815,11 @@ impl Host {
let mut ready_id = None;
if let Some(session) = session.clone() {
{
let mut s = session.lock();
loop {
let session_result = s.readable(io, &self.info.read());
let session_result = session.lock().readable(io, &self.info.read());
match session_result {
Err(e) => {
let s = session.lock();
trace!(target: "network", "Session read error: {}:{:?} ({:?}) {:?}", token, s.id(), s.remote_addr(), e);
if let ErrorKind::Disconnect(DisconnectReason::IncompatibleProtocol) = *e.kind() {
if let Some(id) = s.id() {
@@ -829,9 +832,9 @@ impl Host {
break;
},
Ok(SessionData::Ready) => {
self.num_sessions.fetch_add(1, AtomicOrdering::SeqCst);
let session_count = self.session_count();
let (min_peers, max_peers, reserved_only, self_id) = {
let (_, egress_count, ingress_count) = self.session_count();
let mut s = session.lock();
let (min_peers, mut max_peers, reserved_only, self_id) = {
let info = self.info.read();
let mut max_peers = info.config.max_peers;
for cap in s.info.capabilities.iter() {
@@ -843,12 +846,17 @@ impl Host {
(info.config.min_peers as usize, max_peers as usize, info.config.non_reserved_mode == NonReservedPeerMode::Deny, info.id().clone())
};
max_peers = max(max_peers, min_peers);
let id = s.id().expect("Ready session always has id").clone();
// Check for the session limit. session_counts accounts for the new session.
// Check for the session limit.
// Outgoing connections are allowed as long as their count is <= min_peers
// Incoming connections are allowed to take all of the max_peers reserve, or at most half of the slots.
let max_ingress = max(max_peers - min_peers, min_peers / 2);
if reserved_only ||
(s.info.originated && session_count > min_peers) ||
(!s.info.originated && session_count > max_peers) {
(s.info.originated && egress_count > min_peers) ||
(!s.info.originated && ingress_count > max_ingress) {
// only proceed if the connecting peer is reserved.
if !self.reserved_nodes.read().contains(&id) {
s.disconnect(io, DisconnectReason::TooManyPeers);
@@ -952,13 +960,12 @@ impl Host {
let mut deregister = false;
let mut expired_session = None;
if let FIRST_SESSION ... LAST_SESSION = token {
let sessions = self.sessions.write();
let sessions = self.sessions.read();
if let Some(session) = sessions.get(token).cloned() {
expired_session = Some(session.clone());
let mut s = session.lock();
if !s.expired() {
if s.is_ready() {
self.num_sessions.fetch_sub(1, AtomicOrdering::SeqCst);
for (p, _) in self.handlers.read().iter() {
if s.have_capability(*p) {
to_disconnect.push(*p);
@@ -990,7 +997,7 @@ impl Host {
fn update_nodes(&self, _io: &IoContext<NetworkIoMessage>, node_changes: TableUpdates) {
let mut to_remove: Vec<PeerId> = Vec::new();
{
let sessions = self.sessions.write();
let sessions = self.sessions.read();
for c in sessions.iter() {
let s = c.lock();
if let Some(id) = s.id() {

25
util/network/src/node_table.rs
View File

@@ -241,24 +241,33 @@ impl NodeTable {
self.nodes.insert(node.id.clone(), node);
}
/// Returns node ids sorted by failure percentage, for nodes with the same failure percentage the absolute number of
/// failures is considered.
pub fn nodes(&self, filter: IpFilter) -> Vec<NodeId> {
fn ordered_entries(&self) -> Vec<&Node> {
let mut refs: Vec<&Node> = self.nodes.values()
.filter(|n| !self.useless_nodes.contains(&n.id))
.filter(|n| n.endpoint.is_allowed(&filter))
.collect();
refs.sort_by(|a, b| {
a.failure_percentage().cmp(&b.failure_percentage())
.then_with(|| a.failures.cmp(&b.failures))
.then_with(|| b.attempts.cmp(&a.attempts)) // we use reverse ordering for number of attempts
});
refs.into_iter().map(|n| n.id).collect()
refs
}
/// Unordered list of all entries
pub fn unordered_entries(&self) -> Vec<NodeEntry> {
self.nodes.values().map(|n| NodeEntry {
/// Returns node ids sorted by failure percentage, for nodes with the same failure percentage the absolute number of
/// failures is considered.
pub fn nodes(&self, filter: IpFilter) -> Vec<NodeId> {
self.ordered_entries().iter()
.filter(|n| n.endpoint.is_allowed(&filter))
.map(|n| n.id)
.collect()
}
/// Ordered list of all entries by failure percentage, for nodes with the same failure percentage the absolute
/// number of failures is considered.
pub fn entries(&self) -> Vec<NodeEntry> {
self.ordered_entries().iter().map(|n| NodeEntry {
endpoint: n.endpoint.clone(),
id: n.id.clone(),
}).collect()