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

Proper light client informant and more verification of imported headers (#5897)

Browse Source 
* do more validation of imported headers in light client

* generalize informant with traits

* informant implementation for light client

* make comment into TODO

* fix broken test

* disable full checking of headers in light client in sync tests
This commit is contained in:
Robert Habermeier
2017-07-10 13:21:11 +02:00
committed by Arkadiy Paronyan
parent f0a6b5d401
commit 67c1f71b6e
8 changed files with 374 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
parity/blockchain.rs
View File

@@ -29,7 +29,7 @@ use ethcore::error::ImportError;
use ethcore::miner::Miner;
use ethcore::verification::queue::VerifierSettings;
use cache::CacheConfig;
use informant::{Informant, MillisecondDuration};
use informant::{Informant, FullNodeInformantData, MillisecondDuration};
use params::{SpecType, Pruning, Switch, tracing_switch_to_bool, fatdb_switch_to_bool};
use helpers::{to_client_config, execute_upgrades};
use dir::Directories;
@@ -238,7 +238,17 @@ fn execute_import(cmd: ImportBlockchain) -> Result<(), String> {
}
};
let informant = Arc::new(Informant::new(client.clone(), None, None, None, None, cmd.with_color));
let informant = Arc::new(Informant::new(
FullNodeInformantData {
client: client.clone(),
sync: None,
net: None,
},
None,
None,
cmd.with_color,
));
service.register_io_handler(informant).map_err(|_| "Unable to register informant handler".to_owned())?;
let do_import = |bytes| {

279
parity/informant.rs
View File

@@ -21,32 +21,21 @@ use self::ansi_term::Style;
use std::sync::{Arc};
use std::sync::atomic::{AtomicUsize, AtomicBool, Ordering as AtomicOrdering};
use std::time::{Instant, Duration};
use ethcore::client::*;
use ethcore::header::BlockNumber;
use ethcore::service::ClientIoMessage;
use ethcore::snapshot::{RestorationStatus, SnapshotService as SS};
use ethcore::snapshot::service::Service as SnapshotService;
use ethsync::{LightSyncProvider, LightSync, SyncProvider, ManageNetwork};
use io::{TimerToken, IoContext, IoHandler};
use isatty::{stdout_isatty};
use ethsync::{SyncProvider, ManageNetwork};
use util::{RwLock, Mutex, H256, Colour, Bytes};
use ethcore::client::*;
use ethcore::service::ClientIoMessage;
use ethcore::snapshot::service::Service as SnapshotService;
use ethcore::snapshot::{RestorationStatus, SnapshotService as SS};
use light::Cache as LightDataCache;
use light::client::LightChainClient;
use number_prefix::{binary_prefix, Standalone, Prefixed};
use parity_rpc::{is_major_importing};
use parity_rpc::informant::RpcStats;
pub struct Informant {
report: RwLock<Option<ClientReport>>,
last_tick: RwLock<Instant>,
with_color: bool,
client: Arc<Client>,
snapshot: Option<Arc<SnapshotService>>,
sync: Option<Arc<SyncProvider>>,
net: Option<Arc<ManageNetwork>>,
rpc_stats: Option<Arc<RpcStats>>,
last_import: Mutex<Instant>,
skipped: AtomicUsize,
skipped_txs: AtomicUsize,
in_shutdown: AtomicBool,
}
use util::{RwLock, Mutex, H256, Colour, Bytes};
/// Format byte counts to standard denominations.
pub fn format_bytes(b: usize) -> String {
@@ -68,29 +57,188 @@ impl MillisecondDuration for Duration {
}
}
impl Informant {
#[derive(Default)]
struct CacheSizes {
sizes: ::std::collections::BTreeMap<&'static str, usize>,
}
impl CacheSizes {
fn insert(&mut self, key: &'static str, bytes: usize) {
self.sizes.insert(key, bytes);
}
fn display<F>(&self, style: Style, paint: F) -> String
where F: Fn(Style, String) -> String
{
use std::fmt::Write;
let mut buf = String::new();
for (name, &size) in &self.sizes {
write!(buf, " {:>8} {}", paint(style, format_bytes(size)), name)
.expect("writing to string won't fail unless OOM; qed")
}
buf
}
}
pub struct SyncInfo {
last_imported_block_number: BlockNumber,
last_imported_old_block_number: Option<BlockNumber>,
num_peers: usize,
max_peers: u32,
}
pub struct Report {
importing: bool,
chain_info: BlockChainInfo,
client_report: ClientReport,
queue_info: BlockQueueInfo,
cache_sizes: CacheSizes,
sync_info: Option<SyncInfo>,
}
/// Something which can provide data to the informant.
pub trait InformantData: Send + Sync {
/// Whether it executes transactions
fn executes_transactions(&self) -> bool;
/// Whether it is currently importing (also included in `Report`)
fn is_major_importing(&self) -> bool;
/// Generate a report of blockchain status, memory usage, and sync info.
fn report(&self) -> Report;
}
/// Informant data for a full node.
pub struct FullNodeInformantData {
pub client: Arc<Client>,
pub sync: Option<Arc<SyncProvider>>,
pub net: Option<Arc<ManageNetwork>>,
}
impl InformantData for FullNodeInformantData {
fn executes_transactions(&self) -> bool { true }
fn is_major_importing(&self) -> bool {
let state = self.sync.as_ref().map(|sync| sync.status().state);
is_major_importing(state, self.client.queue_info())
}
fn report(&self) -> Report {
let (client_report, queue_info, blockchain_cache_info) =
(self.client.report(), self.client.queue_info(), self.client.blockchain_cache_info());
let chain_info = self.client.chain_info();
let mut cache_sizes = CacheSizes::default();
cache_sizes.insert("db", client_report.state_db_mem);
cache_sizes.insert("queue", queue_info.mem_used);
cache_sizes.insert("chain", blockchain_cache_info.total());
let (importing, sync_info) = match (self.sync.as_ref(), self.net.as_ref()) {
(Some(sync), Some(net)) => {
let status = sync.status();
let net_config = net.network_config();
cache_sizes.insert("sync", status.mem_used);
let importing = is_major_importing(Some(status.state), queue_info.clone());
(importing, Some(SyncInfo {
last_imported_block_number: status.last_imported_block_number.unwrap_or(chain_info.best_block_number),
last_imported_old_block_number: status.last_imported_old_block_number,
num_peers: status.num_peers,
max_peers: status.current_max_peers(net_config.min_peers, net_config.max_peers),
}))
}
_ => (is_major_importing(None, queue_info.clone()), None),
};
Report {
importing,
chain_info,
client_report,
queue_info,
cache_sizes,
sync_info,
}
}
}
/// Informant data for a light node -- note that the network is required.
pub struct LightNodeInformantData {
pub client: Arc<LightChainClient>,
pub sync: Arc<LightSync>,
pub cache: Arc<Mutex<LightDataCache>>,
}
impl InformantData for LightNodeInformantData {
fn executes_transactions(&self) -> bool { false }
fn is_major_importing(&self) -> bool {
self.sync.is_major_importing()
}
fn report(&self) -> Report {
let (client_report, queue_info, chain_info) =
(self.client.report(), self.client.queue_info(), self.client.chain_info());
let mut cache_sizes = CacheSizes::default();
cache_sizes.insert("queue", queue_info.mem_used);
cache_sizes.insert("cache", self.cache.lock().mem_used());
let peer_numbers = self.sync.peer_numbers();
let sync_info = Some(SyncInfo {
last_imported_block_number: chain_info.best_block_number,
last_imported_old_block_number: None,
num_peers: peer_numbers.connected,
max_peers: peer_numbers.max as u32,
});
Report {
importing: self.sync.is_major_importing(),
chain_info,
client_report,
queue_info,
cache_sizes,
sync_info,
}
}
}
pub struct Informant<T> {
last_tick: RwLock<Instant>,
with_color: bool,
target: T,
snapshot: Option<Arc<SnapshotService>>,
rpc_stats: Option<Arc<RpcStats>>,
last_import: Mutex<Instant>,
skipped: AtomicUsize,
skipped_txs: AtomicUsize,
in_shutdown: AtomicBool,
last_report: Mutex<ClientReport>,
}
impl<T: InformantData> Informant<T> {
/// Make a new instance potentially `with_color` output.
pub fn new(
client: Arc<Client>,
sync: Option<Arc<SyncProvider>>,
net: Option<Arc<ManageNetwork>>,
target: T,
snapshot: Option<Arc<SnapshotService>>,
rpc_stats: Option<Arc<RpcStats>>,
with_color: bool,
) -> Self {
Informant {
report: RwLock::new(None),
last_tick: RwLock::new(Instant::now()),
with_color: with_color,
client: client,
target: target,
snapshot: snapshot,
sync: sync,
net: net,
rpc_stats: rpc_stats,
last_import: Mutex::new(Instant::now()),
skipped: AtomicUsize::new(0),
skipped_txs: AtomicUsize::new(0),
in_shutdown: AtomicBool::new(false),
last_report: Mutex::new(Default::default()),
}
}
@@ -106,14 +254,24 @@ impl Informant {
return;
}
let chain_info = self.client.chain_info();
let queue_info = self.client.queue_info();
let cache_info = self.client.blockchain_cache_info();
let network_config = self.net.as_ref().map(|n| n.network_config());
let sync_status = self.sync.as_ref().map(|s| s.status());
let Report {
importing,
chain_info,
client_report,
queue_info,
cache_sizes,
sync_info,
} = self.target.report();
let client_report = {
let mut last_report = self.last_report.lock();
let diffed = client_report.clone() - &*last_report;
*last_report = client_report.clone();
diffed
};
let rpc_stats = self.rpc_stats.as_ref();
let importing = is_major_importing(sync_status.map(|s| s.state), self.client.queue_info());
let (snapshot_sync, snapshot_current, snapshot_total) = self.snapshot.as_ref().map_or((false, 0, 0), |s|
match s.status() {
RestorationStatus::Ongoing { state_chunks, block_chunks, state_chunks_done, block_chunks_done } =>
@@ -128,9 +286,6 @@ impl Informant {
*self.last_tick.write() = Instant::now();
let mut write_report = self.report.write();
let report = self.client.report();
let paint = |c: Style, t: String| match self.with_color && stdout_isatty() {
true => format!("{}", c.paint(t)),
false => t,
@@ -142,13 +297,16 @@ impl Informant {
false => format!("Syncing {} {} {} {}+{} Qed",
paint(White.bold(), format!("{:>8}", format!("#{}", chain_info.best_block_number))),
paint(White.bold(), format!("{}", chain_info.best_block_hash)),
{
let last_report = match *write_report { Some(ref last_report) => last_report.clone(), _ => ClientReport::default() };
if self.target.executes_transactions() {
format!("{} blk/s {} tx/s {} Mgas/s",
paint(Yellow.bold(), format!("{:4}", ((report.blocks_imported - last_report.blocks_imported) * 1000) as u64 / elapsed.as_milliseconds())),
paint(Yellow.bold(), format!("{:4}", ((report.transactions_applied - last_report.transactions_applied) * 1000) as u64 / elapsed.as_milliseconds())),
paint(Yellow.bold(), format!("{:3}", ((report.gas_processed - last_report.gas_processed) / From::from(elapsed.as_milliseconds() * 1000)).low_u64()))
)
paint(Yellow.bold(), format!("{:4}", (client_report.blocks_imported * 1000) as u64 / elapsed.as_milliseconds())),
paint(Yellow.bold(), format!("{:4}", (client_report.transactions_applied * 1000) as u64 / elapsed.as_milliseconds())),
paint(Yellow.bold(), format!("{:3}", (client_report.gas_processed / From::from(elapsed.as_milliseconds() * 1000)).low_u64()))
)
} else {
format!("{} hdr/s",
paint(Yellow.bold(), format!("{:4}", (client_report.blocks_imported * 1000) as u64 / elapsed.as_milliseconds()))
)
},
paint(Green.bold(), format!("{:5}", queue_info.unverified_queue_size)),
paint(Green.bold(), format!("{:5}", queue_info.verified_queue_size))
@@ -157,29 +315,21 @@ impl Informant {
},
false => String::new(),
},
match (&sync_status, &network_config) {
(&Some(ref sync_info), &Some(ref net_config)) => format!("{}{}/{} peers",
match sync_info.as_ref() {
Some(ref sync_info) => format!("{}{}/{} peers",
match importing {
true => format!("{} ", paint(Green.bold(), format!("{:>8}", format!("#{}", sync_info.last_imported_block_number.unwrap_or(chain_info.best_block_number))))),
true => format!("{} ", paint(Green.bold(), format!("{:>8}", format!("#{}", sync_info.last_imported_block_number)))),
false => match sync_info.last_imported_old_block_number {
Some(number) => format!("{} ", paint(Yellow.bold(), format!("{:>8}", format!("#{}", number)))),
None => String::new(),
}
},
paint(Cyan.bold(), format!("{:2}", sync_info.num_peers)),
paint(Cyan.bold(), format!("{:2}", sync_info.current_max_peers(net_config.min_peers, net_config.max_peers))),
paint(Cyan.bold(), format!("{:2}", sync_info.max_peers)),
),
_ => String::new(),
},
format!("{} db {} chain {} queue{}",
paint(Blue.bold(), format!("{:>8}", format_bytes(report.state_db_mem))),
paint(Blue.bold(), format!("{:>8}", format_bytes(cache_info.total()))),
paint(Blue.bold(), format!("{:>8}", format_bytes(queue_info.mem_used))),
match sync_status {
Some(ref sync_info) => format!(" {} sync", paint(Blue.bold(), format!("{:>8}", format_bytes(sync_info.mem_used)))),
_ => String::new(),
}
),
cache_sizes.display(Blue.bold(), &paint),
match rpc_stats {
Some(ref rpc_stats) => format!(
"RPC: {} conn, {} req/s, {} µs",
@@ -190,25 +340,24 @@ impl Informant {
_ => String::new(),
},
);
*write_report = Some(report);
}
}
impl ChainNotify for Informant {
impl ChainNotify for Informant<FullNodeInformantData> {
fn new_blocks(&self, imported: Vec<H256>, _invalid: Vec<H256>, _enacted: Vec<H256>, _retracted: Vec<H256>, _sealed: Vec<H256>, _proposed: Vec<Bytes>, duration: u64) {
let mut last_import = self.last_import.lock();
let sync_state = self.sync.as_ref().map(|s| s.status().state);
let importing = is_major_importing(sync_state, self.client.queue_info());
let client = &self.target.client;
let importing = self.target.is_major_importing();
let ripe = Instant::now() > *last_import + Duration::from_secs(1) && !importing;
let txs_imported = imported.iter()
.take(imported.len().saturating_sub(if ripe { 1 } else { 0 }))
.filter_map(|h| self.client.block(BlockId::Hash(*h)))
.filter_map(|h| client.block(BlockId::Hash(*h)))
.map(|b| b.transactions_count())
.sum();
if ripe {
if let Some(block) = imported.last().and_then(|h| self.client.block(BlockId::Hash(*h))) {
if let Some(block) = imported.last().and_then(|h| client.block(BlockId::Hash(*h))) {
let header_view = block.header_view();
let size = block.rlp().as_raw().len();
let (skipped, skipped_txs) = (self.skipped.load(AtomicOrdering::Relaxed) + imported.len() - 1, self.skipped_txs.load(AtomicOrdering::Relaxed) + txs_imported);
@@ -241,7 +390,7 @@ impl ChainNotify for Informant {
const INFO_TIMER: TimerToken = 0;
impl IoHandler<ClientIoMessage> for Informant {
impl<T: InformantData> IoHandler<ClientIoMessage> for Informant<T> {
fn initialize(&self, io: &IoContext<ClientIoMessage>) {
io.register_timer(INFO_TIMER, 5000).expect("Error registering timer");
}

63
parity/run.rs
View File

@@ -16,26 +16,27 @@
use std::sync::Arc;
use std::net::{TcpListener};
use ctrlc::CtrlC;
use fdlimit::raise_fd_limit;
use parity_rpc::{NetworkSettings, informant, is_major_importing};
use ethsync::NetworkConfiguration;
use util::{Colour, version, Mutex, Condvar};
use ethcore_logger::{Config as LogConfig, RotatingLogger};
use ethcore::miner::{StratumOptions, Stratum};
use ethcore::client::{Client, Mode, DatabaseCompactionProfile, VMType, BlockChainClient};
use ethcore::service::ClientService;
use ethcore::account_provider::{AccountProvider, AccountProviderSettings};
use ethcore::client::{Client, Mode, DatabaseCompactionProfile, VMType, BlockChainClient};
use ethcore::ethstore::ethkey;
use ethcore::miner::{Miner, MinerService, ExternalMiner, MinerOptions};
use ethcore::miner::{StratumOptions, Stratum};
use ethcore::service::ClientService;
use ethcore::snapshot;
use ethcore::verification::queue::VerifierSettings;
use ethcore::ethstore::ethkey;
use light::Cache as LightDataCache;
use ethsync::NetworkConfiguration;
use ethsync::SyncConfig;
use informant::Informant;
use updater::{UpdatePolicy, Updater};
use parity_reactor::EventLoop;
use fdlimit::raise_fd_limit;
use hash_fetch::fetch::{Fetch, Client as FetchClient};
use informant::{Informant, LightNodeInformantData, FullNodeInformantData};
use light::Cache as LightDataCache;
use parity_reactor::EventLoop;
use parity_rpc::{NetworkSettings, informant, is_major_importing};
use updater::{UpdatePolicy, Updater};
use util::{Colour, version, Mutex, Condvar};
use params::{
SpecType, Pruning, AccountsConfig, GasPricerConfig, MinerExtras, Switch,
@@ -209,6 +210,7 @@ fn execute_light(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) ->
db_cache_size: Some(cmd.cache_config.blockchain() as usize * 1024 * 1024),
db_compaction: compaction,
db_wal: cmd.wal,
verify_full: true,
};
config.queue.max_mem_use = cmd.cache_config.queue() as usize * 1024 * 1024;
@@ -300,7 +302,7 @@ fn execute_light(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) ->
logger: logger,
settings: Arc::new(cmd.net_settings),
on_demand: on_demand,
cache: cache,
cache: cache.clone(),
transaction_queue: txq,
dapps_service: dapps_service,
dapps_address: cmd.dapps_conf.address(cmd.http_conf.address()),
@@ -322,16 +324,25 @@ fn execute_light(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) ->
let _ipc_server = rpc::new_ipc(cmd.ipc_conf, &dependencies)?;
let _ui_server = rpc::new_http("Parity Wallet (UI)", "ui", cmd.ui_conf.clone().into(), &dependencies, ui_middleware)?;
// minimal informant thread. Just prints block number every 5 seconds.
// TODO: integrate with informant.rs
let informant_client = service.client().clone();
::std::thread::spawn(move || loop {
info!("#{}", informant_client.best_block_header().number());
::std::thread::sleep(::std::time::Duration::from_secs(5));
});
// the informant
let informant = Arc::new(Informant::new(
LightNodeInformantData {
client: service.client().clone(),
sync: light_sync.clone(),
cache: cache,
},
None,
Some(rpc_stats),
cmd.logger_config.color,
));
// wait for ctrl-c.
Ok(wait_for_exit(None, None, can_restart))
service.register_handler(informant.clone()).map_err(|_| "Unable to register informant handler".to_owned())?;
// wait for ctrl-c and then shut down the informant.
let res = wait_for_exit(None, None, can_restart);
informant.shutdown();
Ok(res)
}
pub fn execute(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) -> Result<(bool, Option<String>), String> {
@@ -672,9 +683,11 @@ pub fn execute(cmd: RunCmd, can_restart: bool, logger: Arc<RotatingLogger>) -> R
// the informant
let informant = Arc::new(Informant::new(
service.client(),
Some(sync_provider.clone()),
Some(manage_network.clone()),
FullNodeInformantData {
client: service.client(),
sync: Some(sync_provider.clone()),
net: Some(manage_network.clone()),
},
Some(snapshot_service.clone()),
Some(rpc_stats.clone()),
cmd.logger_config.color,