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

Remove network stats (#8225)

Browse Source
This commit is contained in:
Pierre Krieger 2018-03-28 08:45:36 +02:00 committed by Marek Kotewicz
parent 06a7ca221c
commit c4dd156113
8 changed files with 10 additions and 145 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
util/network-devp2p/src/connection.rs
View File

@ -14,7 +14,6 @@
// 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::sync::Arc;
use std::collections::VecDeque;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering as AtomicOrdering};
@ -28,7 +27,6 @@ use rlp::*;
use std::io::{self, Cursor, Read, Write};
use io::{IoContext, StreamToken};
use handshake::Handshake;
use stats::NetworkStats;
use rcrypto::blockmodes::*;
use rcrypto::aessafe::*;
use rcrypto::symmetriccipher::*;
@ -61,8 +59,6 @@ pub struct GenericConnection<Socket: GenericSocket> {
send_queue: VecDeque<Cursor<Bytes>>,
/// Event flags this connection expects
interest: Ready,
/// Shared network statistics
stats: Arc<NetworkStats>,
/// Registered flag
registered: AtomicBool,
}
@ -87,7 +83,6 @@ impl<Socket: GenericSocket> GenericConnection<Socket> {
match sock_ref.take(max as u64).try_read(unsafe { self.rec_buf.bytes_mut() }) {
Ok(Some(size)) if size != 0 => {
unsafe { self.rec_buf.advance_mut(size); }
self.stats.inc_recv(size);
trace!(target:"network", "{}: Read {} of {} bytes", self.token, self.rec_buf.len(), self.rec_size);
if self.rec_size != 0 && self.rec_buf.len() == self.rec_size {
self.rec_size = 0;
@ -141,11 +136,9 @@ impl<Socket: GenericSocket> GenericConnection<Socket> {
match self.socket.try_write(Buf::bytes(&buf)) {
Ok(Some(size)) if (pos + size) < send_size => {
buf.advance(size);
self.stats.inc_send(size);
Ok(WriteStatus::Ongoing)
},
Ok(Some(size)) if (pos + size) == send_size => {
self.stats.inc_send(size);
trace!(target:"network", "{}: Wrote {} bytes", self.token, send_size);
Ok(WriteStatus::Complete)
},
@ -171,7 +164,7 @@ pub type Connection = GenericConnection<TcpStream>;
impl Connection {
/// Create a new connection with given id and socket.
pub fn new(token: StreamToken, socket: TcpStream, stats: Arc<NetworkStats>) -> Connection {
pub fn new(token: StreamToken, socket: TcpStream) -> Connection {
Connection {
token: token,
socket: socket,
@ -179,7 +172,6 @@ impl Connection {
rec_buf: Bytes::new(),
rec_size: 0,
interest: Ready::hup() | Ready::readable(),
stats: stats,
registered: AtomicBool::new(false),
}
}
@ -213,7 +205,6 @@ impl Connection {
rec_size: 0,
send_queue: self.send_queue.clone(),
interest: Ready::hup(),
stats: self.stats.clone(),
registered: AtomicBool::new(false),
})
}
@ -507,13 +498,11 @@ mod tests {
use std::cmp;
use std::collections::VecDeque;
use std::io::{Read, Write, Cursor, ErrorKind, Result, Error};
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use mio::{Ready};
use ethcore_bytes::Bytes;
use io::*;
use super::super::stats::*;
use super::*;
pub struct TestSocket {
@ -625,7 +614,6 @@ mod tests {
rec_buf: Bytes::new(),
rec_size: 0,
interest: Ready::hup() | Ready::readable(),
stats: Arc::<NetworkStats>::new(NetworkStats::new()),
registered: AtomicBool::new(false),
}
}
@ -648,7 +636,6 @@ mod tests {
rec_buf: Bytes::new(),
rec_size: 0,
interest: Ready::hup() | Ready::readable(),
stats: Arc::<NetworkStats>::new(NetworkStats::new()),
registered: AtomicBool::new(false),
}
}

10
util/network-devp2p/src/handshake.rs
View File

@ -14,7 +14,6 @@
// 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::sync::Arc;
use rand::random;
use hash::write_keccak;
use mio::tcp::*;
@ -23,7 +22,6 @@ use ethcore_bytes::Bytes;
use rlp::*;
use connection::{Connection};
use node_table::NodeId;
use stats::NetworkStats;
use io::{IoContext, StreamToken};
use ethkey::{KeyPair, Public, Secret, recover, sign, Generator, Random};
use crypto::{ecdh, ecies};
@ -82,10 +80,10 @@ const ECIES_OVERHEAD: usize = 113;
impl Handshake {
/// Create a new handshake object
pub fn new(token: StreamToken, id: Option<&NodeId>, socket: TcpStream, nonce: &H256, stats: Arc<NetworkStats>) -> Result<Handshake, Error> {
pub fn new(token: StreamToken, id: Option<&NodeId>, socket: TcpStream, nonce: &H256) -> Result<Handshake, Error> {
Ok(Handshake {
id: if let Some(id) = id { id.clone()} else { NodeId::new() },
connection: Connection::new(token, socket, stats),
connection: Connection::new(token, socket),
originated: false,
state: HandshakeState::New,
ecdhe: Random.generate()?,
@ -329,13 +327,11 @@ impl Handshake {
#[cfg(test)]
mod test {
use std::sync::Arc;
use rustc_hex::FromHex;
use super::*;
use ethereum_types::H256;
use io::*;
use mio::tcp::TcpStream;
use stats::NetworkStats;
use ethkey::Public;
fn check_auth(h: &Handshake, version: u64) {
@ -355,7 +351,7 @@ mod test {
let addr = "127.0.0.1:50556".parse().unwrap();
let socket = TcpStream::connect(&addr).unwrap();
let nonce = H256::new();
Handshake::new(0, to, socket, &nonce, Arc::new(NetworkStats::new())).unwrap()
Handshake::new(0, to, socket, &nonce).unwrap()
}
fn test_io() -> IoContext<i32> {

10
util/network-devp2p/src/host.rs
View File

@ -39,7 +39,6 @@ use network::{NetworkConfiguration, NetworkIoMessage, ProtocolId, PeerId, Packet
use network::{NonReservedPeerMode, NetworkContext as NetworkContextTrait};
use network::HostInfo as HostInfoTrait;
use network::{SessionInfo, Error, ErrorKind, DisconnectReason, NetworkProtocolHandler};
use stats::NetworkStats;
use discovery::{Discovery, TableUpdates, NodeEntry};
use ip_utils::{map_external_address, select_public_address};
use path::restrict_permissions_owner;
@ -245,7 +244,6 @@ pub struct Host {
handlers: RwLock<HashMap<ProtocolId, Arc<NetworkProtocolHandler + Sync>>>,
timers: RwLock<HashMap<TimerToken, ProtocolTimer>>,
timer_counter: RwLock<usize>,
stats: Arc<NetworkStats>,
reserved_nodes: RwLock<HashSet<NodeId>>,
stopping: AtomicBool,
filter: Option<Arc<ConnectionFilter>>,
@ -253,7 +251,7 @@ pub struct Host {
impl Host {
/// Create a new instance
pub fn new(mut config: NetworkConfiguration, stats: Arc<NetworkStats>, filter: Option<Arc<ConnectionFilter>>) -> Result<Host, Error> {
pub fn new(mut config: NetworkConfiguration, filter: Option<Arc<ConnectionFilter>>) -> Result<Host, Error> {
let mut listen_address = match config.listen_address {
None => SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), DEFAULT_PORT)),
Some(addr) => addr,
@ -301,7 +299,6 @@ impl Host {
handlers: RwLock::new(HashMap::new()),
timers: RwLock::new(HashMap::new()),
timer_counter: RwLock::new(USER_TIMER),
stats: stats,
reserved_nodes: RwLock::new(HashSet::new()),
stopping: AtomicBool::new(false),
filter: filter,
@ -616,7 +613,7 @@ impl Host {
let mut sessions = self.sessions.write();
let token = sessions.insert_with_opt(|token| {
match Session::new(io, socket, token, id, &nonce, self.stats.clone(), &self.info.read()) {
match Session::new(io, socket, token, id, &nonce, &self.info.read()) {
Ok(s) => Some(Arc::new(Mutex::new(s))),
Err(e) => {
debug!(target: "network", "Session create error: {:?}", e);
@ -793,7 +790,6 @@ impl Host {
return;
}
for p in ready_data {
self.stats.inc_sessions();
let reserved = self.reserved_nodes.read();
if let Some(h) = handlers.get(&p).clone() {
h.connected(&NetworkContext::new(io, p, Some(session.clone()), self.sessions.clone(), &reserved), &token);
@ -1150,6 +1146,6 @@ fn host_client_url() {
let mut config = NetworkConfiguration::new_local();
let key = "6f7b0d801bc7b5ce7bbd930b84fd0369b3eb25d09be58d64ba811091046f3aa2".parse().unwrap();
config.use_secret = Some(key);
let host: Host = Host::new(config, Arc::new(NetworkStats::new()), None).unwrap();
let host: Host = Host::new(config, None).unwrap();
assert!(host.local_url().starts_with("enode://101b3ef5a4ea7a1c7928e24c4c75fd053c235d7b80c22ae5c03d145d0ac7396e2a4ffff9adee3133a7b05044a5cee08115fd65145e5165d646bde371010d803c@"));
}

2
util/network-devp2p/src/lib.rs
View File

@ -102,12 +102,10 @@ mod session;
mod discovery;
mod service;
mod node_table;
mod stats;
mod ip_utils;
mod connection_filter;
pub use service::NetworkService;
pub use stats::NetworkStats;
pub use connection_filter::{ConnectionFilter, ConnectionDirection};
pub use host::NetworkContext;

11
util/network-devp2p/src/service.rs
View File

@ -17,7 +17,6 @@
use network::{Error, NetworkConfiguration, NetworkProtocolHandler, NonReservedPeerMode};
use network::{NetworkContext, PeerId, ProtocolId, NetworkIoMessage};
use host::Host;
use stats::NetworkStats;
use io::*;
use parking_lot::RwLock;
use std::sync::Arc;
@ -46,7 +45,6 @@ pub struct NetworkService {
io_service: IoService<NetworkIoMessage>,
host_info: String,
host: RwLock<Option<Arc<Host>>>,
stats: Arc<NetworkStats>,
host_handler: Arc<HostHandler>,
config: NetworkConfiguration,
filter: Option<Arc<ConnectionFilter>>,
@ -58,11 +56,9 @@ impl NetworkService {
let host_handler = Arc::new(HostHandler { public_url: RwLock::new(None) });
let io_service = IoService::<NetworkIoMessage>::start()?;
let stats = Arc::new(NetworkStats::new());
Ok(NetworkService {
io_service: io_service,
host_info: config.client_version.clone(),
stats: stats,
host: RwLock::new(None),
config: config,
host_handler: host_handler,
@ -91,11 +87,6 @@ impl NetworkService {
&self.io_service
}
/// Returns network statistics.
pub fn stats(&self) -> &NetworkStats {
&self.stats
}
/// Returns network configuration.
pub fn config(&self) -> &NetworkConfiguration {
&self.config
@ -117,7 +108,7 @@ impl NetworkService {
pub fn start(&self) -> Result<(), Error> {
let mut host = self.host.write();
if host.is_none() {
let h = Arc::new(Host::new(self.config.clone(), self.stats.clone(), self.filter.clone())?);
let h = Arc::new(Host::new(self.config.clone(), self.filter.clone())?);
self.io_service.register_handler(h.clone())?;
*host = Some(h);
}

6
util/network-devp2p/src/session.rs
View File

@ -16,7 +16,6 @@
use std::{str, io};
use std::net::SocketAddr;
use std::sync::*;
use std::collections::HashMap;
use std::time::{Duration, Instant};
@ -32,7 +31,6 @@ use network::{Error, ErrorKind, DisconnectReason, SessionInfo, ProtocolId, PeerC
use network::{SessionCapabilityInfo, HostInfo as HostInfoTrait};
use host::*;
use node_table::NodeId;
use stats::NetworkStats;
use snappy;
// Timeout must be less than (interval - 1).
@ -103,10 +101,10 @@ impl Session {
/// Create a new session out of comepleted handshake. This clones the handshake connection object
/// and leaves the handhsake in limbo to be deregistered from the event loop.
pub fn new<Message>(io: &IoContext<Message>, socket: TcpStream, token: StreamToken, id: Option<&NodeId>,
nonce: &H256, stats: Arc<NetworkStats>, host: &HostInfo) -> Result<Session, Error>
nonce: &H256, host: &HostInfo) -> Result<Session, Error>
where Message: Send + Clone + Sync + 'static {
let originated = id.is_some();
let mut handshake = Handshake::new(token, id, socket, nonce, stats).expect("Can't create handshake");
let mut handshake = Handshake::new(token, id, socket, nonce).expect("Can't create handshake");
let local_addr = handshake.connection.local_addr_str();
handshake.start(io, host, originated)?;
Ok(Session {

76
util/network-devp2p/src/stats.rs
View File

@ -1,76 +0,0 @@
// Copyright 2015-2017 Parity Technologies (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/>.
//! Network Statistics
use std::sync::atomic::*;
/// Network statistics structure
#[derive(Default, Debug)]
pub struct NetworkStats {
/// Bytes received
recv: AtomicUsize,
/// Bytes sent
send: AtomicUsize,
/// Total number of sessions created
sessions: AtomicUsize,
}
impl NetworkStats {
/// Increase bytes received.
#[inline]
pub fn inc_recv(&self, size: usize) {
self.recv.fetch_add(size, Ordering::Relaxed);
}
/// Increase bytes sent.
#[inline]
pub fn inc_send(&self, size: usize) {
self.send.fetch_add(size, Ordering::Relaxed);
}
/// Increase number of sessions.
#[inline]
pub fn inc_sessions(&self) {
self.sessions.fetch_add(1, Ordering::Relaxed);
}
/// Get bytes sent.
#[inline]
pub fn send(&self) -> usize {
self.send.load(Ordering::Relaxed)
}
/// Get bytes received.
#[inline]
pub fn recv(&self) -> usize {
self.recv.load(Ordering::Relaxed)
}
/// Get total number of sessions created.
#[inline]
pub fn sessions(&self) -> usize {
self.sessions.load(Ordering::Relaxed)
}
/// Create a new empty instance.
pub fn new() -> NetworkStats {
NetworkStats {
recv: AtomicUsize::new(0),
send: AtomicUsize::new(0),
sessions: AtomicUsize::new(0),
}
}
}

25
util/network-devp2p/tests/tests.rs
View File

@ -14,8 +14,6 @@
// You should have received a copy of the GNU General Public License
// along with Parity. If not, see <http://www.gnu.org/licenses/>.
#[macro_use]
extern crate log;
extern crate parking_lot;
extern crate ethcore_bytes;
extern crate ethcore_io as io;
@ -109,29 +107,6 @@ fn net_service() {
service.register_protocol(Arc::new(TestProtocol::new(false)), *b"myp", 1, &[1u8]).unwrap();
}
#[test]
fn net_connect() {
::ethcore_logger::init_log();
let key1 = Random.generate().unwrap();
let mut config1 = NetworkConfiguration::new_local();
config1.use_secret = Some(key1.secret().clone());
config1.boot_nodes = vec![ ];
let mut service1 = NetworkService::new(config1, None).unwrap();
service1.start().unwrap();
let handler1 = TestProtocol::register(&mut service1, false);
let mut config2 = NetworkConfiguration::new_local();
info!("net_connect: local URL: {}", service1.local_url().unwrap());
config2.boot_nodes = vec![ service1.local_url().unwrap() ];
let mut service2 = NetworkService::new(config2, None).unwrap();
service2.start().unwrap();
let handler2 = TestProtocol::register(&mut service2, false);
while !handler1.got_packet() && !handler2.got_packet() && (service1.stats().sessions() == 0 || service2.stats().sessions() == 0) {
thread::sleep(Duration::from_millis(50));
}
assert!(service1.stats().sessions() >= 1);
assert!(service2.stats().sessions() >= 1);
}
#[test]
fn net_start_stop() {
let config = NetworkConfiguration::new_local();