From b953f9b66aefc9c65c9202b6320d6627de4fa00a Mon Sep 17 00:00:00 2001 From: Robert Habermeier Date: Thu, 24 Aug 2017 15:17:48 +0200 Subject: [PATCH] glue for fetching epoch proofs from network --- ethcore/light/src/client/fetch.rs | 5 +- ethcore/light/src/client/mod.rs | 9 ++- ethcore/src/header.rs | 5 ++ parity/blockchain.rs | 4 +- parity/dapps.rs | 17 +++-- parity/informant.rs | 2 +- parity/light_helpers/epoch_fetch.rs | 89 +++++++++++++++++++++++++++ parity/light_helpers/mod.rs | 2 + parity/light_helpers/queue_cull.rs | 8 +-- parity/rpc_apis.rs | 9 +-- parity/run.rs | 17 +++-- rpc/src/v1/impls/light/eth.rs | 16 ++--- sync/src/chain.rs | 2 +- sync/src/light_sync/tests/test_net.rs | 13 +++- sync/src/tests/consensus.rs | 8 +-- 15 files changed, 168 insertions(+), 38 deletions(-) create mode 100644 parity/light_helpers/epoch_fetch.rs diff --git a/ethcore/light/src/client/fetch.rs b/ethcore/light/src/client/fetch.rs index 2e125bb42..9f4a38dc4 100644 --- a/ethcore/light/src/client/fetch.rs +++ b/ethcore/light/src/client/fetch.rs @@ -18,6 +18,7 @@ use std::sync::Arc; +use ethcore::encoded; use ethcore::engines::{Engine, StateDependentProof}; use ethcore::header::Header; use ethcore::receipt::Receipt; @@ -30,7 +31,7 @@ pub trait ChainDataFetcher: Send + Sync + 'static { type Error: ::std::fmt::Debug; /// Future for fetching block body. - type Body: IntoFuture,Error=Self::Error>; + type Body: IntoFuture; /// Future for fetching block receipts. type Receipts: IntoFuture,Error=Self::Error>; /// Future for fetching epoch transition @@ -55,7 +56,7 @@ pub fn unavailable() -> Unavailable { Unavailable } impl ChainDataFetcher for Unavailable { type Error = &'static str; - type Body = Result, &'static str>; + type Body = Result; type Receipts = Result, &'static str>; type Transition = Result, &'static str>; diff --git a/ethcore/light/src/client/mod.rs b/ethcore/light/src/client/mod.rs index 9c8e1da50..569acecc6 100644 --- a/ethcore/light/src/client/mod.rs +++ b/ethcore/light/src/client/mod.rs @@ -83,6 +83,9 @@ impl Default for Config { /// Trait for interacting with the header chain abstractly. pub trait LightChainClient: Send + Sync { + /// Adds a new `LightChainNotify` listener. + fn add_listener(&self, listener: Weak); + /// Get chain info. fn chain_info(&self) -> BlockChainInfo; @@ -481,7 +484,7 @@ impl Client { }; if let Some(b) = b { - block = Some(b.into_future().wait()?); + block = Some(b.into_future().wait()?.into_inner()); } if let Some(r) = r { @@ -526,6 +529,10 @@ impl Client { } impl LightChainClient for Client { + fn add_listener(&self, listener: Weak) { + Client::add_listener(self, listener) + } + fn chain_info(&self) -> BlockChainInfo { Client::chain_info(self) } fn queue_header(&self, header: Header) -> Result { diff --git a/ethcore/src/header.rs b/ethcore/src/header.rs index a9a4f948d..a256b5c87 100644 --- a/ethcore/src/header.rs +++ b/ethcore/src/header.rs @@ -259,6 +259,11 @@ impl Header { /// Get the SHA3 (Keccak) of this header, optionally `with_seal`. pub fn rlp_sha3(&self, with_seal: Seal) -> H256 { self.rlp(with_seal).sha3() } + + /// Encode the header, getting a type-safe wrapper around the RLP. + pub fn encoded(&self) -> ::encoded::Header { + ::encoded::Header::new(self.rlp(Seal::With)) + } } impl Decodable for Header { diff --git a/parity/blockchain.rs b/parity/blockchain.rs index 426c8d447..d9821c5bb 100644 --- a/parity/blockchain.rs +++ b/parity/blockchain.rs @@ -204,7 +204,9 @@ fn execute_import_light(cmd: ImportBlockchain) -> Result<(), String> { config.queue.max_mem_use = cmd.cache_config.queue() as usize * 1024 * 1024; config.queue.verifier_settings = cmd.verifier_settings; - let service = LightClientService::start(config, &spec, &client_path, cache) + // TODO: could epoch signals be avilable at the end of the file? + let fetch = ::light::client::fetch::unavailable(); + let service = LightClientService::start(config, &spec, fetch, &client_path, cache) .map_err(|e| format!("Failed to start client: {}", e))?; // free up the spec in memory. diff --git a/parity/dapps.rs b/parity/dapps.rs index fddd050f5..b7ca739ec 100644 --- a/parity/dapps.rs +++ b/parity/dapps.rs @@ -26,7 +26,7 @@ use futures_cpupool::CpuPool; use hash_fetch::fetch::Client as FetchClient; use hash_fetch::urlhint::ContractClient; use helpers::replace_home; -use light::client::Client as LightClient; +use light::client::LightChainClient; use light::on_demand::{self, OnDemand}; use rpc; use rpc_apis::SignerService; @@ -94,16 +94,16 @@ impl ContractClient for FullRegistrar { } /// Registrar implementation for the light client. -pub struct LightRegistrar { +pub struct LightRegistrar { /// The light client. - pub client: Arc, + pub client: Arc, /// Handle to the on-demand service. pub on_demand: Arc, /// Handle to the light network service. pub sync: Arc, } -impl ContractClient for LightRegistrar { +impl ContractClient for LightRegistrar { fn registrar(&self) -> Result { self.client.engine().additional_params().get("registrar") .ok_or_else(|| "Registrar not defined.".into()) @@ -113,7 +113,14 @@ impl ContractClient for LightRegistrar { } fn call(&self, address: Address, data: Bytes) -> BoxFuture { - let (header, env_info) = (self.client.best_block_header(), self.client.latest_env_info()); + let header = self.client.best_block_header(); + let env_info = self.client.env_info(BlockId::Hash(header.hash())) + .ok_or_else(|| format!("Cannot fetch env info for header {}", header.hash())); + + let env_info = match env_info { + Ok(x) => x, + Err(e) => return future::err(e).boxed(), + }; let maybe_future = self.sync.with_context(move |ctx| { self.on_demand diff --git a/parity/informant.rs b/parity/informant.rs index 1935ec9b6..2be83d566 100644 --- a/parity/informant.rs +++ b/parity/informant.rs @@ -22,7 +22,7 @@ use std::sync::{Arc}; use std::sync::atomic::{AtomicUsize, AtomicBool, Ordering as AtomicOrdering}; use std::time::{Instant, Duration}; -use ethcore::client::*; +use ethcore::client::{BlockId, BlockChainClient, BlockChainInfo, BlockQueueInfo, ChainNotify, ClientReport, Client}; use ethcore::header::BlockNumber; use ethcore::service::ClientIoMessage; use ethcore::snapshot::{RestorationStatus, SnapshotService as SS}; diff --git a/parity/light_helpers/epoch_fetch.rs b/parity/light_helpers/epoch_fetch.rs new file mode 100644 index 000000000..d710abb9f --- /dev/null +++ b/parity/light_helpers/epoch_fetch.rs @@ -0,0 +1,89 @@ +// 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 . + +use std::sync::{Arc, Weak}; + +use ethcore::encoded; +use ethcore::engines::{Engine, StateDependentProof}; +use ethcore::header::Header; +use ethcore::receipt::Receipt; +use ethsync::LightSync; + +use futures::{future, Future, BoxFuture}; + +use light::client::fetch::ChainDataFetcher; +use light::on_demand::{request, OnDemand}; + +use util::{RwLock, H256}; + +const ALL_VALID_BACKREFS: &str = "no back-references, therefore all back-references valid; qed"; + +/// Allows on-demand fetch of data useful for the light client. +pub struct EpochFetch { + /// A handle to the sync service. + pub sync: Arc>>, + /// The on-demand request service. + pub on_demand: Arc, +} + +impl EpochFetch { + fn request(&self, req: T) -> BoxFuture + where T: Send + request::RequestAdapter + 'static, T::Out: Send + 'static + { + match self.sync.read().upgrade() { + Some(sync) => { + let on_demand = &self.on_demand; + let maybe_future = sync.with_context(move |ctx| { + on_demand.request(ctx, req).expect(ALL_VALID_BACKREFS) + }); + + match maybe_future { + Some(x) => x.map_err(|_| "Request canceled").boxed(), + None => future::err("Unable to access network.").boxed(), + } + } + None => future::err("Unable to access network").boxed(), + } + } +} + +impl ChainDataFetcher for EpochFetch { + type Error = &'static str; + + type Body = BoxFuture; + type Receipts = BoxFuture, &'static str>; + type Transition = BoxFuture, &'static str>; + + fn block_body(&self, header: &Header) -> Self::Body { + self.request(request::Body(header.encoded().into())) + } + + /// Fetch block receipts. + fn block_receipts(&self, header: &Header) -> Self::Receipts { + self.request(request::BlockReceipts(header.encoded().into())) + } + + /// Fetch epoch transition proof at given header. + fn epoch_transition(&self, hash: H256, engine: Arc, checker: Arc) + -> Self::Transition + { + self.request(request::Signal { + hash: hash, + engine: engine, + proof_check: checker, + }) + } +} diff --git a/parity/light_helpers/mod.rs b/parity/light_helpers/mod.rs index 488f970c2..5fc9c516b 100644 --- a/parity/light_helpers/mod.rs +++ b/parity/light_helpers/mod.rs @@ -16,6 +16,8 @@ //! Utilities and helpers for the light client. +mod epoch_fetch; mod queue_cull; +pub use self::epoch_fetch::EpochFetch; pub use self::queue_cull::QueueCull; diff --git a/parity/light_helpers/queue_cull.rs b/parity/light_helpers/queue_cull.rs index 982e69e48..34769b864 100644 --- a/parity/light_helpers/queue_cull.rs +++ b/parity/light_helpers/queue_cull.rs @@ -23,7 +23,7 @@ use ethcore::service::ClientIoMessage; use ethsync::LightSync; use io::{IoContext, IoHandler, TimerToken}; -use light::client::Client; +use light::client::LightChainClient; use light::on_demand::{request, OnDemand}; use light::TransactionQueue; @@ -41,9 +41,9 @@ const TIMEOUT_MS: u64 = 1000 * 60 * 10; const PURGE_TIMEOUT_MS: u64 = 1000 * 60 * 9; /// Periodically culls the transaction queue of mined transactions. -pub struct QueueCull { +pub struct QueueCull { /// A handle to the client, for getting the latest block header. - pub client: Arc, + pub client: Arc, /// A handle to the sync service. pub sync: Arc, /// The on-demand request service. @@ -54,7 +54,7 @@ pub struct QueueCull { pub remote: Remote, } -impl IoHandler for QueueCull { +impl IoHandler for QueueCull { fn initialize(&self, io: &IoContext) { io.register_timer(TOKEN, TIMEOUT_MS).expect("Error registering timer"); } diff --git a/parity/rpc_apis.rs b/parity/rpc_apis.rs index dba69ae7e..bb1e1c377 100644 --- a/parity/rpc_apis.rs +++ b/parity/rpc_apis.rs @@ -34,6 +34,7 @@ use ethsync::{ManageNetwork, SyncProvider, LightSync}; use hash_fetch::fetch::Client as FetchClient; use jsonrpc_core::{self as core, MetaIoHandler}; use light::{TransactionQueue as LightTransactionQueue, Cache as LightDataCache}; +use light::client::LightChainClient; use updater::Updater; use util::{Mutex, RwLock}; use ethcore_logger::RotatingLogger; @@ -395,9 +396,9 @@ impl ActivityNotifier for LightClientNotifier { } /// RPC dependencies for a light client. -pub struct LightDependencies { +pub struct LightDependencies { pub signer_service: Arc, - pub client: Arc<::light::client::Client>, + pub client: Arc, pub sync: Arc, pub net: Arc, pub secret_store: Arc, @@ -415,7 +416,7 @@ pub struct LightDependencies { pub whisper_rpc: Option<::whisper::RpcFactory>, } -impl LightDependencies { +impl LightDependencies { fn extend_api>( &self, handler: &mut MetaIoHandler, @@ -563,7 +564,7 @@ impl LightDependencies { } } -impl Dependencies for LightDependencies { +impl Dependencies for LightDependencies { type Notifier = LightClientNotifier; fn activity_notifier(&self) -> Self::Notifier { LightClientNotifier } diff --git a/parity/run.rs b/parity/run.rs index 49624480c..277ee9025 100644 --- a/parity/run.rs +++ b/parity/run.rs @@ -14,7 +14,7 @@ // You should have received a copy of the GNU General Public License // along with Parity. If not, see . -use std::sync::Arc; +use std::sync::{Arc, Weak}; use std::net::{TcpListener}; use ctrlc::CtrlC; @@ -217,7 +217,16 @@ fn execute_light(cmd: RunCmd, can_restart: bool, logger: Arc) -> config.queue.max_mem_use = cmd.cache_config.queue() as usize * 1024 * 1024; config.queue.verifier_settings = cmd.verifier_settings; - let service = light_client::Service::start(config, &spec, &db_dirs.client_path(algorithm), cache.clone()) + // start on_demand service. + let on_demand = Arc::new(::light::on_demand::OnDemand::new(cache.clone())); + + let sync_handle = Arc::new(RwLock::new(Weak::new())); + let fetch = ::light_helpers::EpochFetch { + on_demand: on_demand.clone(), + sync: sync_handle.clone(), + }; + + let service = light_client::Service::start(config, &spec, fetch, &db_dirs.client_path(algorithm), cache.clone()) .map_err(|e| format!("Error starting light client: {}", e))?; let txq = Arc::new(RwLock::new(::light::transaction_queue::TransactionQueue::default())); let provider = ::light::provider::LightProvider::new(service.client().clone(), txq.clone()); @@ -229,9 +238,6 @@ fn execute_light(cmd: RunCmd, can_restart: bool, logger: Arc) -> net_conf.boot_nodes = spec.nodes.clone(); } - // start on_demand service. - let on_demand = Arc::new(::light::on_demand::OnDemand::new(cache.clone())); - let mut attached_protos = Vec::new(); let whisper_factory = if cmd.whisper.enabled { let (whisper_net, whisper_factory) = ::whisper::setup(cmd.whisper.target_message_pool_size) @@ -255,6 +261,7 @@ fn execute_light(cmd: RunCmd, can_restart: bool, logger: Arc) -> }; let light_sync = LightSync::new(sync_params).map_err(|e| format!("Error starting network: {}", e))?; let light_sync = Arc::new(light_sync); + *sync_handle.write() = Arc::downgrade(&light_sync); // spin up event loop let event_loop = EventLoop::spawn(); diff --git a/rpc/src/v1/impls/light/eth.rs b/rpc/src/v1/impls/light/eth.rs index d33b5a06f..b515d7961 100644 --- a/rpc/src/v1/impls/light/eth.rs +++ b/rpc/src/v1/impls/light/eth.rs @@ -25,7 +25,7 @@ use jsonrpc_core::Error; use jsonrpc_macros::Trailing; use light::cache::Cache as LightDataCache; -use light::client::{Client as LightClient, LightChainClient}; +use light::client::LightChainClient; use light::{cht, TransactionQueue}; use light::on_demand::{request, OnDemand}; @@ -62,9 +62,9 @@ use util::Address; const NO_INVALID_BACK_REFS: &'static str = "Fails only on invalid back-references; back-references here known to be valid; qed"; /// Light client `ETH` (and filter) RPC. -pub struct EthClient { +pub struct EthClient { sync: Arc, - client: Arc, + client: Arc, on_demand: Arc, transaction_queue: Arc>, accounts: Arc, @@ -72,7 +72,7 @@ pub struct EthClient { polls: Mutex>, } -impl Clone for EthClient { +impl Clone for EthClient { fn clone(&self) -> Self { // each instance should have its own poll manager. EthClient { @@ -88,12 +88,12 @@ impl Clone for EthClient { } -impl EthClient { +impl EthClient { /// Create a new `EthClient` with a handle to the light sync instance, client, /// and on-demand request service, which is assumed to be attached as a handler. pub fn new( sync: Arc, - client: Arc, + client: Arc, on_demand: Arc, transaction_queue: Arc>, accounts: Arc, @@ -208,7 +208,7 @@ impl EthClient { } } -impl Eth for EthClient { +impl Eth for EthClient { type Metadata = Metadata; fn protocol_version(&self) -> Result { @@ -465,7 +465,7 @@ impl Eth for EthClient { } // This trait implementation triggers a blanked impl of `EthFilter`. -impl Filterable for EthClient { +impl Filterable for EthClient { fn best_block_number(&self) -> u64 { self.client.chain_info().best_block_number } fn block_hash(&self, id: BlockId) -> Option { diff --git a/sync/src/chain.rs b/sync/src/chain.rs index 3a472c88f..a4708a12d 100644 --- a/sync/src/chain.rs +++ b/sync/src/chain.rs @@ -2238,7 +2238,7 @@ mod tests { use super::{PeerInfo, PeerAsking}; use ethkey; use ethcore::header::*; - use ethcore::client::*; + use ethcore::client::{BlockChainClient, EachBlockWith, TestBlockChainClient}; use ethcore::transaction::UnverifiedTransaction; use ethcore::miner::MinerService; diff --git a/sync/src/light_sync/tests/test_net.rs b/sync/src/light_sync/tests/test_net.rs index f6d5eddf0..51cb796de 100644 --- a/sync/src/light_sync/tests/test_net.rs +++ b/sync/src/light_sync/tests/test_net.rs @@ -25,7 +25,7 @@ use tests::helpers::{TestNet, Peer as PeerLike, TestPacket}; use ethcore::client::TestBlockChainClient; use ethcore::spec::Spec; use io::IoChannel; -use light::client::Client as LightClient; +use light::client::fetch::{self, Unavailable}; use light::net::{LightProtocol, IoContext, Capabilities, Params as LightParams}; use light::provider::LightProvider; use network::{NodeId, PeerId}; @@ -36,6 +36,8 @@ use light::cache::Cache; const NETWORK_ID: u64 = 0xcafebabe; +pub type LightClient = ::light::client::Client; + struct TestIoContext<'a> { queue: &'a RwLock>, sender: Option, @@ -216,7 +218,14 @@ impl TestNet { // skip full verification because the blocks are bad. config.verify_full = false; let cache = Arc::new(Mutex::new(Cache::new(Default::default(), Duration::hours(6)))); - let client = LightClient::in_memory(config, &Spec::new_test(), IoChannel::disconnected(), cache); + let client = LightClient::in_memory( + config, + &Spec::new_test(), + fetch::unavailable(), // TODO: allow fetch from full nodes. + IoChannel::disconnected(), + cache + ); + peers.push(Arc::new(Peer::new_light(Arc::new(client)))) } diff --git a/sync/src/tests/consensus.rs b/sync/src/tests/consensus.rs index 9b01156b7..c31ea692f 100644 --- a/sync/src/tests/consensus.rs +++ b/sync/src/tests/consensus.rs @@ -69,8 +69,8 @@ fn authority_round() { // Push transaction to both clients. Only one of them gets lucky to produce a block. net.peer(0).chain.miner().set_engine_signer(s0.address(), "".to_owned()).unwrap(); net.peer(1).chain.miner().set_engine_signer(s1.address(), "".to_owned()).unwrap(); - net.peer(0).chain.engine().register_client(Arc::downgrade(&net.peer(0).chain)); - net.peer(1).chain.engine().register_client(Arc::downgrade(&net.peer(1).chain)); + net.peer(0).chain.engine().register_client(Arc::downgrade(&net.peer(0).chain) as _); + net.peer(1).chain.engine().register_client(Arc::downgrade(&net.peer(1).chain) as _); net.peer(0).chain.set_io_channel(IoChannel::to_handler(Arc::downgrade(&io_handler1))); net.peer(1).chain.set_io_channel(IoChannel::to_handler(Arc::downgrade(&io_handler0))); // exchange statuses @@ -158,8 +158,8 @@ fn tendermint() { trace!(target: "poa", "Peer 0 is {}.", s0.address()); net.peer(1).chain.miner().set_engine_signer(s1.address(), "".to_owned()).unwrap(); trace!(target: "poa", "Peer 1 is {}.", s1.address()); - net.peer(0).chain.engine().register_client(Arc::downgrade(&net.peer(0).chain)); - net.peer(1).chain.engine().register_client(Arc::downgrade(&net.peer(1).chain)); + net.peer(0).chain.engine().register_client(Arc::downgrade(&net.peer(0).chain) as _); + net.peer(1).chain.engine().register_client(Arc::downgrade(&net.peer(1).chain) as _); net.peer(0).chain.set_io_channel(IoChannel::to_handler(Arc::downgrade(&io_handler0))); net.peer(1).chain.set_io_channel(IoChannel::to_handler(Arc::downgrade(&io_handler1))); // Exhange statuses