RPC: Implements eth_subscribe("syncing")

ethcore/light/src/client/mod.rs

@@ -34,6 +34,7 @@ use common_types::blockchain_info::BlockChainInfo;
 use common_types::encoded;
 use common_types::header::Header;
 use common_types::ids::BlockId;
+use common_types::verification_queue_info::VerificationQueueInfo as BlockQueueInfo;
 
 use kvdb::KeyValueDB;
 
@@ -91,6 +92,9 @@ pub trait LightChainClient: Send + Sync {
 	/// Attempt to get a block hash by block id.
 	fn block_hash(&self, id: BlockId) -> Option<H256>;
 
+	/// Get block queue information.
+	fn queue_info(&self) -> BlockQueueInfo;
+
 	/// Attempt to get block header by block id.
 	fn block_header(&self, id: BlockId) -> Option<encoded::Header>;
 
@@ -125,9 +129,6 @@ pub trait LightChainClient: Send + Sync {
 	/// Flush the queue.
 	fn flush_queue(&self);
 
-	/// Get queue info.
-	fn queue_info(&self) -> queue::QueueInfo;
-
 	/// Get the `i`th CHT root.
 	fn cht_root(&self, i: usize) -> Option<H256>;
 
@@ -534,13 +535,18 @@ impl<T: ChainDataFetcher> Client<T> {
 	}
 }
 
+
 impl<T: ChainDataFetcher> LightChainClient for Client<T> {
 	fn add_listener(&self, listener: Weak<LightChainNotify>) {
 		Client::add_listener(self, listener)
 	}
 
 	fn chain_info(&self) -> BlockChainInfo { Client::chain_info(self) }
 
+	fn queue_info(&self) -> queue::QueueInfo {
+		self.queue.queue_info()
+	}
+
 	fn queue_header(&self, header: Header) -> EthcoreResult<H256> {
 		self.import_header(header)
 	}
@@ -600,10 +606,6 @@ impl<T: ChainDataFetcher> LightChainClient for Client<T> {
 		Client::flush_queue(self);
 	}
 
-	fn queue_info(&self) -> queue::QueueInfo {
-		self.queue.queue_info()
-	}
-
 	fn cht_root(&self, i: usize) -> Option<H256> {
 		Client::cht_root(self, i)
 	}

ethcore/src/client/client.rs

@@ -1672,6 +1672,10 @@ impl BlockChainClient for Client {
 		r
 	}
 
+	fn queue_info(&self) -> BlockQueueInfo {
+		self.importer.block_queue.queue_info()
+	}
+
 	fn disable(&self) {
 		self.set_mode(Mode::Off);
 		self.enabled.store(false, AtomicOrdering::Relaxed);
@@ -1934,10 +1938,6 @@ impl BlockChainClient for Client {
 		self.chain.read().block_receipts(hash)
 	}
 
-	fn queue_info(&self) -> BlockQueueInfo {
-		self.importer.block_queue.queue_info()
-	}
-
 	fn is_queue_empty(&self) -> bool {
 		self.importer.block_queue.is_empty()
 	}

ethcore/src/client/test_client.rs

@@ -56,7 +56,7 @@ use client::{
 	TransactionId, UncleId, TraceId, TraceFilter, LastHashes, CallAnalytics,
 	ProvingBlockChainClient, ScheduleInfo, ImportSealedBlock, BroadcastProposalBlock, ImportBlock, StateOrBlock,
 	Call, StateClient, EngineInfo, AccountData, BlockChain, BlockProducer, SealedBlockImporter, IoClient,
-	BadBlocks,
+	BadBlocks
 };
 use engines::EthEngine;
 use error::{Error, EthcoreResult};
@@ -68,7 +68,7 @@ use spec::Spec;
 use state::StateInfo;
 use state_db::StateDB;
 use trace::LocalizedTrace;
-use verification::queue::QueueInfo;
+use verification::queue::QueueInfo as BlockQueueInfo;
 use verification::queue::kind::blocks::Unverified;
 
 /// Test client.
@@ -649,6 +649,17 @@ impl BlockChainClient for TestBlockChainClient {
 		self.execution_result.read().clone().unwrap()
 	}
 
+	fn queue_info(&self) -> BlockQueueInfo {
+		BlockQueueInfo {
+			verified_queue_size: self.queue_size.load(AtomicOrder::Relaxed),
+			unverified_queue_size: 0,
+			verifying_queue_size: 0,
+			max_queue_size: 0,
+			max_mem_use: 0,
+			mem_used: 0,
+		}
+	}
+
 	fn replay_block_transactions(&self, _block: BlockId, _analytics: CallAnalytics) -> Result<Box<Iterator<Item = (H256, Executed)>>, CallError> {
 		Ok(Box::new(self.traces.read().clone().unwrap().into_iter().map(|t| t.transaction_hash.unwrap_or(H256::new())).zip(self.execution_result.read().clone().unwrap().into_iter())))
 	}
@@ -817,17 +828,6 @@ impl BlockChainClient for TestBlockChainClient {
 		None
 	}
 
-	fn queue_info(&self) -> QueueInfo {
-		QueueInfo {
-			verified_queue_size: self.queue_size.load(AtomicOrder::Relaxed),
-			unverified_queue_size: 0,
-			verifying_queue_size: 0,
-			max_queue_size: 0,
-			max_mem_use: 0,
-			mem_used: 0,
-		}
-	}
-
 	fn clear_queue(&self) {
 	}
 

ethcore/src/client/traits.rs

@@ -237,6 +237,9 @@ pub trait BlockChainClient : Sync + Send + AccountData + BlockChain + CallContra
 			.expect("code will return Some if given BlockId::Latest; qed")
 	}
 
+	/// Get block queue information.
+	fn queue_info(&self) -> BlockQueueInfo;
+
 	/// Get address code hash at given block's state.
 
 	/// Get value of the storage at given position at the given block's state.
@@ -285,9 +288,6 @@ pub trait BlockChainClient : Sync + Send + AccountData + BlockChain + CallContra
 	/// Get block receipts data by block header hash.
 	fn block_receipts(&self, hash: &H256) -> Option<BlockReceipts>;
 
-	/// Get block queue information.
-	fn queue_info(&self) -> BlockQueueInfo;
-
 	/// Returns true if block queue is empty.
 	fn is_queue_empty(&self) -> bool {
 		self.queue_info().is_empty()

ethcore/sync/Cargo.toml

@@ -32,6 +32,8 @@ rand = "0.4"
 rlp = { version = "0.3.0", features = ["ethereum"] }
 trace-time = "0.1"
 triehash-ethereum = {version = "0.2", path = "../../util/triehash-ethereum" }
+futures = "0.1"
+parity-runtime = { path = "../../util/runtime" }
 
 [dev-dependencies]
 env_logger = "0.5"

ethcore/sync/src/api.rs

@@ -28,6 +28,8 @@ use network::client_version::ClientVersion;
 
 use types::pruning_info::PruningInfo;
 use ethereum_types::{H256, H512, U256};
+use futures::sync::mpsc as futures_mpsc;
+use futures::Stream;
 use io::{TimerToken};
 use ethkey::Secret;
 use ethcore::client::{BlockChainClient, ChainNotify, NewBlocks, ChainMessageType};
@@ -39,14 +41,16 @@ use std::net::{SocketAddr, AddrParseError};
 use std::str::FromStr;
 use parking_lot::{RwLock, Mutex};
 use chain::{ETH_PROTOCOL_VERSION_63, ETH_PROTOCOL_VERSION_62,
-	PAR_PROTOCOL_VERSION_1, PAR_PROTOCOL_VERSION_2, PAR_PROTOCOL_VERSION_3};
+	PAR_PROTOCOL_VERSION_1, PAR_PROTOCOL_VERSION_2, PAR_PROTOCOL_VERSION_3, SyncState};
 use chain::sync_packet::SyncPacket::{PrivateTransactionPacket, SignedPrivateTransactionPacket};
 use light::client::AsLightClient;
 use light::Provider;
 use light::net::{
 	self as light_net, LightProtocol, Params as LightParams,
 	Capabilities, Handler as LightHandler, EventContext, SampleStore,
 };
+use parity_runtime::Executor;
+use std::sync::atomic::{AtomicBool, Ordering};
 use network::IpFilter;
 use private_tx::PrivateTxHandler;
 use types::transaction::UnverifiedTransaction;
@@ -131,6 +135,9 @@ impl Default for SyncConfig {
 	}
 }
 
+/// receiving end of a futures::mpsc channel
+pub type Notification<T> = futures_mpsc::UnboundedReceiver<T>;
+
 /// Current sync status
 pub trait SyncProvider: Send + Sync {
 	/// Get sync status
@@ -142,8 +149,14 @@ pub trait SyncProvider: Send + Sync {
 	/// Get the enode if available.
 	fn enode(&self) -> Option<String>;
 
+	/// gets sync status notifications
+	fn sync_notification(&self) -> Notification<SyncState>;
+
 	/// Returns propagation count for pending transactions.
 	fn transactions_stats(&self) -> BTreeMap<H256, TransactionStats>;
+
+	/// are we in the middle of a major sync?
+	fn is_major_syncing(&self) -> bool;
 }
 
 /// Transaction stats
@@ -266,6 +279,8 @@ impl PriorityTask {
 pub struct Params {
 	/// Configuration.
 	pub config: SyncConfig,
+	/// Runtime executor
+	pub executor: Executor,
 	/// Blockchain client.
 	pub chain: Arc<BlockChainClient>,
 	/// Snapshot service.
@@ -296,6 +311,8 @@ pub struct EthSync {
 	light_subprotocol_name: [u8; 3],
 	/// Priority tasks notification channel
 	priority_tasks: Mutex<mpsc::Sender<PriorityTask>>,
+	/// for state tracking
+	is_major_syncing: Arc<AtomicBool>
 }
 
 fn light_params(
@@ -355,6 +372,30 @@ impl EthSync {
 			params.private_tx_handler.as_ref().cloned(),
 			priority_tasks_rx,
 		);
+
+		let is_major_syncing = Arc::new(AtomicBool::new(false));
+
+		{
+			// spawn task that constantly updates EthSync.is_major_sync
+			let notifications = sync.write().sync_notifications();
+			let moved_client = Arc::downgrade(&params.chain);
+			let moved_is_major_syncing = is_major_syncing.clone();
+
+			params.executor.spawn(notifications.for_each(move |sync_status| {
+				if let Some(queue_info) = moved_client.upgrade().map(|client| client.queue_info()) {
+					let is_syncing_state = match sync_status {
+						SyncState::Idle | SyncState::NewBlocks => false,
+						_ => true
+					};
+					let is_verifying = queue_info.unverified_queue_size + queue_info.verified_queue_size > 3;
+					moved_is_major_syncing.store(is_verifying || is_syncing_state, Ordering::SeqCst);
+					return Ok(())
+				}
+
+				// client has been dropped
+				return Err(())
+			}));
+		}
 		let service = NetworkService::new(params.network_config.clone().into_basic()?, connection_filter)?;
 
 		let sync = Arc::new(EthSync {
@@ -370,6 +411,7 @@ impl EthSync {
 			light_subprotocol_name: params.config.light_subprotocol_name,
 			attached_protos: params.attached_protos,
 			priority_tasks: Mutex::new(priority_tasks_tx),
+			is_major_syncing
 		});
 
 		Ok(sync)
@@ -420,6 +462,14 @@ impl SyncProvider for EthSync {
 	fn transactions_stats(&self) -> BTreeMap<H256, TransactionStats> {
 		self.eth_handler.sync.transactions_stats()
 	}
+
+	fn sync_notification(&self) -> Notification<SyncState> {
+		self.eth_handler.sync.write().sync_notifications()
+	}
+
+	fn is_major_syncing(&self) -> bool {
+		self.is_major_syncing.load(Ordering::SeqCst)
+	}
 }
 
 const PEERS_TIMER: TimerToken = 0;