RPC: Implements eth_subscribe("syncing") (#10311)

* implement eth_subscibe syncing

* fix imports

* add is_major_syncing to SyncProvider

* add eth_subscribe(syncing) support for light clients

* tests

* fix TestSyncClient

* correct LightFetch impl

* added currentBlock, startingBlock, highestBlock to PubSubSyncStatus

* fixed tests

* fix PR grumbles

* improve code style

* is_syncing -> syncing

* code style

* use ethereum types

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,7 +41,7 @@ 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;
@@ -47,6 +49,8 @@ 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;

ethcore/sync/src/chain/mod.rs

@@ -99,6 +99,8 @@ use std::cmp;
 use std::time::{Duration, Instant};
 use hash::keccak;
 use heapsize::HeapSizeOf;
+use futures::sync::mpsc as futures_mpsc;
+use api::Notification;
 use ethereum_types::{H256, U256};
 use fastmap::{H256FastMap, H256FastSet};
 use parking_lot::{Mutex, RwLock, RwLockWriteGuard};
@@ -618,6 +620,8 @@ pub struct ChainSync {
 	private_tx_handler: Option<Arc<PrivateTxHandler>>,
 	/// Enable warp sync.
 	warp_sync: WarpSync,
+
+	status_sinks: Vec<futures_mpsc::UnboundedSender<SyncState>>
 }
 
 impl ChainSync {
@@ -649,6 +653,7 @@ impl ChainSync {
 			transactions_stats: TransactionsStats::default(),
 			private_tx_handler,
 			warp_sync: config.warp_sync,
+			status_sinks: Vec::new()
 		};
 		sync.update_targets(chain);
 		sync
@@ -707,6 +712,29 @@ impl ChainSync {
 		self.peers.clear();
 	}
 
+	/// returns the receiving end of a future::mpsc channel that can
+	/// be polled for changes to node's SyncState.
+	pub fn sync_notifications(&mut self) -> Notification<SyncState> {
+		let (sender, receiver) = futures_mpsc::unbounded();
+		self.status_sinks.push(sender);
+		receiver
+	}
+
+	/// notify all subscibers of a new SyncState
+	fn notify_sync_state(&mut self, state: SyncState) {
+		// remove any sender whose receiving end has been dropped
+		self.status_sinks.retain(|sender| {
+			sender.unbounded_send(state).is_ok()
+		});
+	}
+
+	/// sets a new SyncState
+	fn set_state(&mut self, state: SyncState) {
+		self.notify_sync_state(state);
+
+		self.state = state;
+	}
+
 	/// Reset sync. Clear all downloaded data but keep the queue.
 	/// Set sync state to the given state or to the initial state if `None` is provided.
 	fn reset(&mut self, io: &mut SyncIo, state: Option<SyncState>) {
@@ -721,7 +749,10 @@ impl ChainSync {
 				}
 			}
 		}
-		self.state = state.unwrap_or_else(|| Self::get_init_state(self.warp_sync, io.chain()));
+
+		let warp_sync = self.warp_sync;
+
+		self.set_state(state.unwrap_or_else(|| Self::get_init_state(warp_sync, io.chain())));
 		// Reactivate peers only if some progress has been made
 		// since the last sync round of if starting fresh.
 		self.active_peers = self.peers.keys().cloned().collect();
@@ -808,7 +839,7 @@ impl ChainSync {
 			}
 		} else if timeout && !self.warp_sync.is_warp_only() {
 			trace!(target: "sync", "No snapshots found, starting full sync");
-			self.state = SyncState::Idle;
+			self.set_state(SyncState::Idle);
 			self.continue_sync(io);
 		}
 	}
@@ -820,10 +851,10 @@ impl ChainSync {
 					SyncRequester::request_snapshot_manifest(self, io, *p);
 				}
 			}
-			self.state = SyncState::SnapshotManifest;
+			self.set_state(SyncState::SnapshotManifest);
 			trace!(target: "sync", "New snapshot sync with {:?}", peers);
 		} else {
-			self.state = SyncState::SnapshotData;
+			self.set_state(SyncState::SnapshotData);
 			trace!(target: "sync", "Resumed snapshot sync with {:?}", peers);
 		}
 	}
@@ -904,7 +935,7 @@ impl ChainSync {
 	/// Enter waiting state
 	fn pause_sync(&mut self) {
 		trace!(target: "sync", "Block queue full, pausing sync");
-		self.state = SyncState::Waiting;
+		self.set_state(SyncState::Waiting);
 	}
 
 	/// Find something to do for a peer. Called for a new peer or when a peer is done with its task.
@@ -955,7 +986,7 @@ impl ChainSync {
 						if let Some(request) = self.new_blocks.request_blocks(peer_id, io, num_active_peers) {
 							SyncRequester::request_blocks(self, io, peer_id, request, BlockSet::NewBlocks);
 							if self.state == SyncState::Idle {
-								self.state = SyncState::Blocks;
+								self.set_state(SyncState::Blocks);
 							}
 							return;
 						}
@@ -987,7 +1018,7 @@ impl ChainSync {
 							self.snapshot.initialize(io.snapshot_service());
 							if self.snapshot.done_chunks() - (state_chunks_done + block_chunks_done) as usize > MAX_SNAPSHOT_CHUNKS_DOWNLOAD_AHEAD {
 								trace!(target: "sync", "Snapshot queue full, pausing sync");
-								self.state = SyncState::SnapshotWaiting;
+								self.set_state(SyncState::SnapshotWaiting);
 								return;
 							}
 						},
@@ -1171,12 +1202,12 @@ impl ChainSync {
 	fn check_resume(&mut self, io: &mut SyncIo) {
 		match self.state {
 			SyncState::Waiting if !io.chain().queue_info().is_full() => {
-				self.state = SyncState::Blocks;
+				self.set_state(SyncState::Blocks);
 				self.continue_sync(io);
 			},
 			SyncState::SnapshotData => match io.snapshot_service().status() {
 				RestorationStatus::Inactive | RestorationStatus::Failed => {
-					self.state = SyncState::SnapshotWaiting;
+					self.set_state(SyncState::SnapshotWaiting);
 				},
 				RestorationStatus::Initializing { .. } | RestorationStatus::Ongoing { .. } => (),
 			},
@@ -1192,13 +1223,13 @@ impl ChainSync {
 					RestorationStatus::Ongoing { state_chunks_done, block_chunks_done, .. } => {
 						if !self.snapshot.is_complete() && self.snapshot.done_chunks() - (state_chunks_done + block_chunks_done) as usize <= MAX_SNAPSHOT_CHUNKS_DOWNLOAD_AHEAD {
 							trace!(target:"sync", "Resuming snapshot sync");
-							self.state = SyncState::SnapshotData;
+							self.set_state(SyncState::SnapshotData);
 							self.continue_sync(io);
 						}
 					},
 					RestorationStatus::Failed => {
 						trace!(target: "sync", "Snapshot restoration aborted");
-						self.state = SyncState::WaitingPeers;
+						self.set_state(SyncState::WaitingPeers);
 						self.snapshot.clear();
 						self.continue_sync(io);
 					},
@@ -1421,7 +1452,8 @@ pub mod tests {
 	}
 
 	pub fn dummy_sync_with_peer(peer_latest_hash: H256, client: &BlockChainClient) -> ChainSync {
-		let mut sync = ChainSync::new(SyncConfig::default(), client, None);
+
+		let mut sync = ChainSync::new(SyncConfig::default(), client, None,);
 		insert_dummy_peer(&mut sync, 0, peer_latest_hash);
 		sync
 	}

ethcore/sync/src/lib.rs

@@ -32,10 +32,12 @@ extern crate ethstore;
 extern crate fastmap;
 extern crate keccak_hash as hash;
 extern crate parity_bytes as bytes;
+extern crate parity_runtime;
 extern crate parking_lot;
 extern crate rand;
 extern crate rlp;
 extern crate triehash_ethereum;
+extern crate futures;
 
 extern crate ethcore_light as light;
 

ethcore/sync/src/light_sync/mod.rs

@@ -45,13 +45,16 @@ use light::net::{
 	EventContext, Capabilities, ReqId, Status,
 	Error as NetError,
 };
+use chain::SyncState as ChainSyncState;
 use light::request::{self, CompleteHeadersRequest as HeadersRequest};
 use network::PeerId;
 use ethereum_types::{H256, U256};
 use parking_lot::{Mutex, RwLock};
 use rand::{Rng, OsRng};
+use futures::sync::mpsc;
 
 use self::sync_round::{AbortReason, SyncRound, ResponseContext};
+use api::Notification;
 
 mod response;
 mod sync_round;
@@ -275,6 +278,7 @@ pub struct LightSync<L: AsLightClient> {
 	client: Arc<L>,
 	rng: Mutex<OsRng>,
 	state: Mutex<SyncStateWrapper>,
+	senders: RwLock<Vec<mpsc::UnboundedSender<ChainSyncState>>>,
 	// We duplicate this state tracking to avoid deadlocks in `is_major_importing`.
 	is_idle: Mutex<bool>,
 }
@@ -454,9 +458,21 @@ impl<L: AsLightClient> LightSync<L> {
 	/// Sets the LightSync's state, and update
 	/// `is_idle`
 	fn set_state(&self, state: &mut SyncStateWrapper, next_state: SyncState) {
+
+		match next_state {
+			SyncState::Idle => self.notify_senders(ChainSyncState::Idle),
+			_ => self.notify_senders(ChainSyncState::Blocks)
+		};
+
 		state.set(next_state, &mut self.is_idle.lock());
 	}
 
+	fn notify_senders(&self, state: ChainSyncState) {
+		self.senders.write().retain(|sender| {
+			sender.unbounded_send(state).is_ok()
+		})
+	}
+
 	// Begins a search for the common ancestor and our best block.
 	// does not lock state, instead has a mutable reference to it passed.
 	fn begin_search(&self, state: &mut SyncStateWrapper) {
@@ -667,6 +683,14 @@ impl<L: AsLightClient> LightSync<L> {
 			self.set_state(&mut state, next_state);
 		}
 	}
+
+	// returns receiving end of futures::mpsc::unbounded channel
+	// poll the channel for changes to sync state.
+	fn sync_notification(&self) -> Notification<ChainSyncState> {
+		let (sender, receiver) = futures::sync::mpsc::unbounded();
+		self.senders.write().push(sender);
+		receiver
+	}
 }
 
 // public API
@@ -683,6 +707,7 @@ impl<L: AsLightClient> LightSync<L> {
 			pending_reqs: Mutex::new(HashMap::new()),
 			client: client,
 			rng: Mutex::new(OsRng::new()?),
+			senders: RwLock::new(Vec::new()),
 			state: Mutex::new(SyncStateWrapper::idle()),
 			is_idle: Mutex::new(true),
 		})
@@ -699,6 +724,10 @@ pub trait SyncInfo {
 
 	/// Whether major sync is underway.
 	fn is_major_importing(&self) -> bool;
+
+	/// returns the receieving end of a futures::mpsc unbounded channel
+	/// poll the channel for changes to sync state
+	fn sync_notification(&self) -> Notification<ChainSyncState>;
 }
 
 impl<L: AsLightClient> SyncInfo for LightSync<L> {
@@ -720,4 +749,7 @@ impl<L: AsLightClient> SyncInfo for LightSync<L> {
 		is_verifying || is_syncing
 	}
 
+	fn sync_notification(&self) -> Notification<ChainSyncState> {
+		self.sync_notification()
+	}
 }