fix(light): make OnDemand generic instead of using the concrete type (

#10514) * ethcore: add clique engine (#9981) * fix broken sync * correct seal fields * ethcore: fix comment * parity: remove duplicate params * clique: fix whitespaces * ethcore: fix goerli chain spec * refactor signer_snapshot into pending/finalized state * move close_block_extra_data after seal is applied * refactor most of the logic into the signer_snapshot * clique: refactor locking logic out of the consensus engine interface * Fix jsonspec and add an unittest * Replace space with tabs * Unbroke sync * Fix broken sync * 1/2 state tracking without votes * 2/2 implement vote tracking * ci: use travis for goerli * ci: setup a clique network * ci: sync a görli node * add clique deploy script * ci: fix paths in clique deploy script * ci: use docker compose * ci: fix travis job names * ci: fix build deps * ci: massively reduce tests * Revert "ci: massively reduce tests" This reverts commit 6369f0b. * ci: run cargo test directly * ci: separate build and test stages * ci: cache rust installation * ci: simplify ci stages * ci: make clique deploy script executable * ci: shutdown goerli sync after 20min * ci: remove slow sync stage * ci: use timeout to finish jobs * ci: fix build path * ci: use absolute paths to end this confusion * ci: add geth and parity to path * ci: be more verbose * ci: allow for more relaxed caching timeout * ci: update repositories for custom ppa * ci: fix typo in file name * ci: fix docker compose file * ci: add ethkey to docker * ci: make sure deploy script is up to date with upstream * ci: stop docker container after certain time * ci: force superuser to update permissions on docker files * ci: reduce run time of script to ~30 min * ci: remove duplicate caching in travis * remove trace statements * clique: add more validation involving the recent signer list * ethcore: enable constantinople for rinkeby * ethcore: fix whitespaces in rinkeby spec * ethcore: reformat goerli.json * Revert "ci: remove duplicate caching in travis" This reverts commit a562838. * tmp commit * another tmp commit * it builds! * add sealing capabilities * add seal_header hook to allow separation of block seal/importing code paths * clique: remove populate_from_parent. * add panic * make turn delay random * initialize OpenBlock properly in 'enact' * misc: remove duplicate lines * misc: fix license headers * misc: convert spaces to tabs * misc: fix tabs * Update Cargo.toml * Update Cargo.toml * Update Cargo.toml * clique: ensure validator restores state before trying to seal * clique: make 'state' return an Error. Make some error messages more clear * Fix compile error after rebase & toolchain upgrade * fix a bunch of import warnings * Refactor code * Fix permissions * Refactoring syncing * Implement full validator checks * Refactor util functions to seperate file * mining 1 * ethcore: add chainspec for kotti * ethcore: rename pre-goerli configs * ethcore: load kotti chain spec * cli: add kotti to params * Implement working local sealing * making sealing & syncing work together * Relax timestamp checking * ethcore: prepare for the real goerli to launch * Implement NOTURN wiggle properly & cleanupnup warnings * Implement vote casting * Update docs & skip signing if no signer * Optimize step-service interval * Record state on local sealed block * Fix script filemode * Cleaning up codebase * restore enact trace logging * Delete clique.sh and move sync.sh * remove travis.yml * Remove dead code * Cleanup compile warning * address review comments * adding more comments and removing unwrap() * ci: remove sync script * Address review comments * fix compile error * adding better debugging for timing * Implement an dedicated thread for sealing timing * fix(add helper for timestamp overflows) (#10330) * fix(add helper timestamp overflows) * fix(simplify code) * fix(make helper private) * snap: official image / test (#10168) * official image / test * fix / test * bit more necromancy * fix paths * add source bin/df /test * add source bin/df /test2 * something w paths /test * something w paths /test * add source-type /test * show paths /test * copy plugin /test * plugin -> nil * install rhash * no questions while installing rhash * publish snap only for release * fix(docker): fix not receives SIGINT (#10059) * fix(docker): fix not receives SIGINT * fix: update with reviews * update with review * update * update * Don't add discovery initiators to the node table (#10305) * Don't add discovery initiators to the node table * Use enums for tracking state of the nodes in discovery * Dont try to ping ourselves * Fix minor nits * Update timeouts when observing an outdated node * Extracted update_bucket_record from update_node * Fixed typo * Fix two final nits from @todr * change docker image based on debian instead of ubuntu due to the chan… (#10336) * change docker image based on debian instead of ubuntu due to the changes of the build container * role back docker build image and docker deploy image to ubuntu:xenial based (#10338) * Bundle protocol and packet_id together in chain sync (#10315) Define a new `enum` where devp2p subprotocol packet ids (currently eth and par) are defined. Additionally provide functionality to query id value and protocol of a given id object. * snap: prefix version and populate candidate channel (#10343) * snap: populate candidate releases with beta snaps to avoid stale channel * snap: prefix version with v* * addressing review comments * engine: fix copyright header * scripts: restore permissions on sign command * ethcore: enforce tabs * ethcore: enforce tabs * ethcore: enforce tabs * addressing comments * addressing comments * addressing more comments * addressing more comments * addressing more comments * addressing more comments * addressing more comments * json-spec: fix clique epoch to non-zero u64 * ci: enable travis for parity goerli * ci: don't separate build and test step * ci: don't run c++ tests on travis * ci: simplify cargo test to squeeze into travis timeout * ci: don't run tests on travis at all * style(fixes) * fix(add tests) * fix(recent_signer bug) * fix(complete all tests) * fix(nits) * fix(simplify asserts) * fix(cliqueState): simplify code * fix(nits) * docs(comments what's need to fixed) * fix(revert unintended changes) * fix(tests) * fix(logs): voting logs * fix(readability + more logs) * fix(sync) * docs(add missing licens header) * fix(log): info! -> trace! * docs(fix nits) + fix(remove assert) * perf(use counter instead of vec) * fix(remove needless block in match) * fix(faulty comment) * grumbles(docs for tests) * fix(nits) * fix(revert_vote): only remove vote when votes == 0 * fix(vote counter): checked arithmetics * fix(simplify tests) * fix(nits) * fix(clique): err types * fix(clique utils): make use of errors * fix(cleanup nits) * fix(clique sealing): don't read state no signer * fix(replace Vec<Signers> with BTreeSet<Signers>) * fix(tests): BTreeSet and more generic helpers * fix(nits) * fix(ethcore_block_seal): remove needless `Box` * fix(faulty log): info -> trace * fix(checked SystemTime): prevent SystemTime panics * style(chain cfg): space after `:` * style(fn enact): fix whitespace * docs(clique): StepService * docs(nit): fix faulty comment * docs(fix typo) * style(fix bad indentation) * fix(bad regex match) * grumble(on_seal_block): make `&mut` to avoid clone * docs(on_seal_block): fix faulty documentation * Delete .travis.yml * docs: remove eth hf references in spec * Update client.rs * fix(nits) * fix(clique step): `RwLock` -> `AtomicBool` * fix(clique): use `Duration::as_millis` * Clean up some Clique documentation * Add trace information to eth_estimateGas (#10519) * Add trace information to eth_estimateGas * replace unwrap better version * change vm::Error formatter to more user-friendly * remove extra error format * use map_or instead sequence of map/unwrap_or * fix(light/on_demand): extract as a trait * fix(grumble): OnDemand remove needless trait bound
openethereum · Mar 27, 2019 · 407de5e · 407de5e
1 parent 7d26a82
commit 407de5e
Show file tree

Hide file tree

Showing 9 changed files with 179 additions and 132 deletions.
diff --git a/ethcore/light/src/on_demand/mod.rs b/ethcore/light/src/on_demand/mod.rs
@@ -94,6 +94,24 @@ pub mod error {
 	}
 }
 
+/// Public interface for performing network requests `OnDemand`
+pub trait OnDemandRequester: Send + Sync {
+	/// Submit a strongly-typed batch of requests.
+	///
+	/// Fails if back-reference are not coherent.
+	fn request<T>(&self, ctx: &BasicContext, requests: T) -> Result<OnResponses<T>, basic_request::NoSuchOutput>
+	where
+		T: request::RequestAdapter;
+
+	/// Submit a vector of requests to be processed together.
+	///
+	/// Fails if back-references are not coherent.
+	/// The returned vector of responses will correspond to the requests exactly.
+	fn request_raw(&self, ctx: &BasicContext, requests: Vec<Request>)
+		-> Result<Receiver<PendingResponse>, basic_request::NoSuchOutput>;
+}
+
+
 // relevant peer info.
 #[derive(Debug, Clone, PartialEq, Eq)]
 struct Peer {
@@ -355,71 +373,8 @@ pub struct OnDemand {
 	request_number_of_consecutive_errors: usize
 }
 
-impl OnDemand {
-
-	/// Create a new `OnDemand` service with the given cache.
-	pub fn new(
-		cache: Arc<Mutex<Cache>>,
-		response_time_window: Duration,
-		request_backoff_start: Duration,
-		request_backoff_max: Duration,
-		request_backoff_rounds_max: usize,
-		request_number_of_consecutive_errors: usize,
-	) -> Self {
-
-		Self {
-			pending: RwLock::new(Vec::new()),
-			peers: RwLock::new(HashMap::new()),
-			in_transit: RwLock::new(HashMap::new()),
-			cache,
-			no_immediate_dispatch: false,
-			response_time_window: Self::sanitize_circuit_breaker_input(response_time_window, "Response time window"),
-			request_backoff_start: Self::sanitize_circuit_breaker_input(request_backoff_start, "Request initial backoff time window"),
-			request_backoff_max: Self::sanitize_circuit_breaker_input(request_backoff_max, "Request maximum backoff time window"),
-			request_backoff_rounds_max,
-			request_number_of_consecutive_errors,
-		}
-	}
-
-	fn sanitize_circuit_breaker_input(dur: Duration, name: &'static str) -> Duration {
-		if dur.as_secs() < 1 {
-			warn!(target: "on_demand",
-				"{} is too short must be at least 1 second, configuring it to 1 second", name);
-			Duration::from_secs(1)
-		} else {
-			dur
-		}
-	}
-
-	// make a test version: this doesn't dispatch pending requests
-	// until you trigger it manually.
-	#[cfg(test)]
-	fn new_test(
-		cache: Arc<Mutex<Cache>>,
-		request_ttl: Duration,
-		request_backoff_start: Duration,
-		request_backoff_max: Duration,
-		request_backoff_rounds_max: usize,
-		request_number_of_consecutive_errors: usize,
-	) -> Self {
-		let mut me = OnDemand::new(
-			cache,
-			request_ttl,
-			request_backoff_start,
-			request_backoff_max,
-			request_backoff_rounds_max,
-			request_number_of_consecutive_errors,
-		);
-		me.no_immediate_dispatch = true;
-
-		me
-	}
-
-	/// Submit a vector of requests to be processed together.
-	///
-	/// Fails if back-references are not coherent.
-	/// The returned vector of responses will correspond to the requests exactly.
-	pub fn request_raw(&self, ctx: &BasicContext, requests: Vec<Request>)
+impl OnDemandRequester for OnDemand {
+	fn request_raw(&self, ctx: &BasicContext, requests: Vec<Request>)
 		-> Result<Receiver<PendingResponse>, basic_request::NoSuchOutput>
 	{
 		let (sender, receiver) = oneshot::channel();
@@ -475,10 +430,7 @@ impl OnDemand {
 		Ok(receiver)
 	}
 
-	/// Submit a strongly-typed batch of requests.
-	///
-	/// Fails if back-reference are not coherent.
-	pub fn request<T>(&self, ctx: &BasicContext, requests: T) -> Result<OnResponses<T>, basic_request::NoSuchOutput>
+	fn request<T>(&self, ctx: &BasicContext, requests: T) -> Result<OnResponses<T>, basic_request::NoSuchOutput>
 		where T: request::RequestAdapter
 	{
 		self.request_raw(ctx, requests.make_requests()).map(|recv| OnResponses {
@@ -487,6 +439,69 @@ impl OnDemand {
 		})
 	}
 
+}
+
+impl OnDemand {
+
+	/// Create a new `OnDemand` service with the given cache.
+	pub fn new(
+		cache: Arc<Mutex<Cache>>,
+		response_time_window: Duration,
+		request_backoff_start: Duration,
+		request_backoff_max: Duration,
+		request_backoff_rounds_max: usize,
+		request_number_of_consecutive_errors: usize,
+	) -> Self {
+
+		Self {
+			pending: RwLock::new(Vec::new()),
+			peers: RwLock::new(HashMap::new()),
+			in_transit: RwLock::new(HashMap::new()),
+			cache,
+			no_immediate_dispatch: false,
+			response_time_window: Self::sanitize_circuit_breaker_input(response_time_window, "Response time window"),
+			request_backoff_start: Self::sanitize_circuit_breaker_input(request_backoff_start, "Request initial backoff time window"),
+			request_backoff_max: Self::sanitize_circuit_breaker_input(request_backoff_max, "Request maximum backoff time window"),
+			request_backoff_rounds_max,
+			request_number_of_consecutive_errors,
+		}
+	}
+
+	fn sanitize_circuit_breaker_input(dur: Duration, name: &'static str) -> Duration {
+		if dur.as_secs() < 1 {
+			warn!(target: "on_demand",
+				"{} is too short must be at least 1 second, configuring it to 1 second", name);
+			Duration::from_secs(1)
+		} else {
+			dur
+		}
+	}
+
+	// make a test version: this doesn't dispatch pending requests
+	// until you trigger it manually.
+	#[cfg(test)]
+	fn new_test(
+		cache: Arc<Mutex<Cache>>,
+		request_ttl: Duration,
+		request_backoff_start: Duration,
+		request_backoff_max: Duration,
+		request_backoff_rounds_max: usize,
+		request_number_of_consecutive_errors: usize,
+	) -> Self {
+		let mut me = OnDemand::new(
+			cache,
+			request_ttl,
+			request_backoff_start,
+			request_backoff_max,
+			request_backoff_rounds_max,
+			request_number_of_consecutive_errors,
+		);
+		me.no_immediate_dispatch = true;
+
+		me
+	}
+
+
 	// maybe dispatch pending requests.
 	// sometimes
 	fn attempt_dispatch(&self, ctx: &BasicContext) {

diff --git a/ethcore/light/src/on_demand/tests.rs b/ethcore/light/src/on_demand/tests.rs
@@ -29,7 +29,7 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 use std::thread;
 
-use super::{request, OnDemand, Peer, HeaderRef};
+use super::{request, OnDemand, OnDemandRequester, Peer, HeaderRef};
 
 // useful contexts to give the service.
 enum Context {

diff --git a/parity/light_helpers/epoch_fetch.rs b/parity/light_helpers/epoch_fetch.rs
@@ -27,7 +27,7 @@ use futures::{future, Future};
 use futures::future::Either;
 
 use light::client::fetch::ChainDataFetcher;
-use light::on_demand::{request, OnDemand};
+use light::on_demand::{request, OnDemand, OnDemandRequester};
 
 use parking_lot::RwLock;
 use ethereum_types::H256;

diff --git a/parity/light_helpers/queue_cull.rs b/parity/light_helpers/queue_cull.rs
@@ -24,7 +24,7 @@ use sync::{LightSync, LightNetworkDispatcher};
 use io::{IoContext, IoHandler, TimerToken};
 
 use light::client::LightChainClient;
-use light::on_demand::{request, OnDemand};
+use light::on_demand::{request, OnDemand, OnDemandRequester};
 use light::TransactionQueue;
 
 use futures::{future, Future};

diff --git a/rpc/src/v1/helpers/dispatch/light.rs b/rpc/src/v1/helpers/dispatch/light.rs
@@ -20,7 +20,7 @@ use ethereum_types::{H256, Address, U256};
 use light::TransactionQueue as LightTransactionQueue;
 use light::cache::Cache as LightDataCache;
 use light::client::LightChainClient;
-use light::on_demand::{request, OnDemand};
+use light::on_demand::{request, OnDemandRequester};
 use parking_lot::{Mutex, RwLock};
 use stats::Corpus;
 use sync::{LightSyncProvider, LightNetworkDispatcher, ManageNetwork};
@@ -37,13 +37,17 @@ use v1::types::{RichRawTransaction as RpcRichRawTransaction,};
 use super::{Dispatcher, Accounts, SignWith, PostSign};
 
 /// Dispatcher for light clients -- fetches default gas price, next nonce, etc. from network.
-pub struct LightDispatcher<S: LightSyncProvider + LightNetworkDispatcher + ManageNetwork + 'static> {
+pub struct LightDispatcher<S, OD>
+where
+	S: LightSyncProvider + LightNetworkDispatcher + ManageNetwork + 'static,
+	OD: OnDemandRequester + 'static
+{
 	/// Sync service.
 	pub sync: Arc<S>,
 	/// Header chain client.
 	pub client: Arc<LightChainClient>,
 	/// On-demand request service.
-	pub on_demand: Arc<OnDemand>,
+	pub on_demand: Arc<OD>,
 	/// Data cache.
 	pub cache: Arc<Mutex<LightDataCache>>,
 	/// Transaction queue.
@@ -54,17 +58,18 @@ pub struct LightDispatcher<S: LightSyncProvider + LightNetworkDispatcher + Manag
 	pub gas_price_percentile: usize,
 }
 
-impl<S> LightDispatcher<S>
+impl<S, OD> LightDispatcher<S, OD>
 where
-	S: LightSyncProvider + LightNetworkDispatcher + ManageNetwork + 'static
+	S: LightSyncProvider + LightNetworkDispatcher + ManageNetwork + 'static,
+	OD: OnDemandRequester + 'static
 {
 	/// Create a new `LightDispatcher` from its requisite parts.
 	///
 	/// For correct operation, the OnDemand service is assumed to be registered as a network handler,
 	pub fn new(
 		sync: Arc<S>,
 		client: Arc<LightChainClient>,
-		on_demand: Arc<OnDemand>,
+		on_demand: Arc<OD>,
 		cache: Arc<Mutex<LightDataCache>>,
 		transaction_queue: Arc<RwLock<LightTransactionQueue>>,
 		nonces: Arc<Mutex<nonce::Reservations>>,
@@ -117,9 +122,10 @@ where
 	}
 }
 
-impl<S> Clone for LightDispatcher<S>
+impl<S, OD> Clone for LightDispatcher<S, OD>
 where
-	S: LightSyncProvider + LightNetworkDispatcher + ManageNetwork + 'static
+	S: LightSyncProvider + LightNetworkDispatcher + ManageNetwork + 'static,
+	OD: OnDemandRequester + 'static
 {
 	fn clone(&self) -> Self {
 		Self {
@@ -134,9 +140,10 @@ where
 	}
 }
 
-impl<S> Dispatcher for LightDispatcher<S>
+impl<S, OD> Dispatcher for LightDispatcher<S, OD>
 where
-	S: LightSyncProvider + LightNetworkDispatcher + ManageNetwork + 'static
+	S: LightSyncProvider + LightNetworkDispatcher + ManageNetwork + 'static,
+	OD: OnDemandRequester + 'static
 {
 	// Ignore the `force_nonce` flag in order to always query the network when fetching the nonce and
 	// the account state. If the nonce is specified in the transaction use that nonce instead but do the
@@ -239,12 +246,16 @@ where
 
 /// Get a recent gas price corpus.
 // TODO: this could be `impl Trait`.
-pub fn fetch_gas_price_corpus<S: LightSyncProvider + LightNetworkDispatcher + ManageNetwork + 'static>(
+pub fn fetch_gas_price_corpus<S, OD>(
 	sync: Arc<S>,
 	client: Arc<LightChainClient>,
-	on_demand: Arc<OnDemand>,
+	on_demand: Arc<OD>,
 	cache: Arc<Mutex<LightDataCache>>,
-) -> BoxFuture<Corpus<U256>> {
+) -> BoxFuture<Corpus<U256>>
+where
+	S: LightSyncProvider + LightNetworkDispatcher + ManageNetwork + 'static,
+	OD: OnDemandRequester + 'static
+{
 	const GAS_PRICE_SAMPLE_SIZE: usize = 100;
 
 	if let Some(cached) = { cache.lock().gas_price_corpus() } {