Switch GrandPa to new futures

core/finality-grandpa/Cargo.toml

@@ -6,12 +6,12 @@ edition = "2018"
 
 [dependencies]
 fork-tree = { path = "../../core/utils/fork-tree" }
-futures = "0.1.29"
-futures03 = { package = "futures-preview", version = "0.3.0-alpha.19", features = ["compat"] }
+futures01 = { package = "futures", version = "0.1.29" }
+futures-preview = { version = "0.3.0-alpha.19", features = ["compat"] }
+futures-timer = "0.3.0"
 log = "0.4.8"
 parking_lot = "0.9.0"
 tokio-executor = "0.1.8"
-tokio-timer = "0.2.11"
 rand = "0.7.2"
 codec = { package = "parity-scale-codec", version = "1.0.0", features = ["derive"] }
 sr-primitives = {  path = "../sr-primitives" }
@@ -26,14 +26,17 @@ inherents = { package = "substrate-inherents", path = "../../core/inherents" }
 network = { package = "substrate-network", path = "../network" }
 srml-finality-tracker = { path = "../../srml/finality-tracker" }
 fg_primitives = { package = "substrate-finality-grandpa-primitives", path = "primitives" }
-grandpa = { package = "finality-grandpa", version = "0.9.0", features = ["derive-codec"] }
+# TODO: switch to crates.io version
+grandpa = { package = "finality-grandpa", git = "https://github.com/paritytech/finality-grandpa", features = ["derive-codec"] }
+#grandpa = { package = "finality-grandpa", version = "0.9.0", features = ["derive-codec"] }
 
 [dev-dependencies]
-grandpa = { package = "finality-grandpa", version = "0.9.0", features = ["derive-codec", "test-helpers"] }
+# TODO: switch to crates.io version
+grandpa = { package = "finality-grandpa", git = "https://github.com/paritytech/finality-grandpa", features = ["derive-codec", "test-helpers"] }
+#grandpa = { package = "finality-grandpa", version = "0.9.0", features = ["derive-codec", "test-helpers"] }
 network = { package = "substrate-network", path = "../network", features = ["test-helpers"] }
 keyring = { package = "substrate-keyring", path = "../keyring" }
 test-client = { package = "substrate-test-runtime-client", path = "../test-runtime/client"}
 babe_primitives = { package = "substrate-consensus-babe-primitives", path = "../consensus/babe/primitives" }
 env_logger = "0.7.0"
-tokio = "0.1.22"
 tempfile = "3.1.0"

core/finality-grandpa/src/communication/gossip.rs

@@ -89,14 +89,16 @@ use codec::{Encode, Decode};
 use fg_primitives::AuthorityId;
 
 use substrate_telemetry::{telemetry, CONSENSUS_DEBUG};
-use log::{trace, debug, warn};
+use log::{trace, debug};
 use futures::prelude::*;
-use futures::sync::mpsc;
+use futures::channel::mpsc;
 
 use crate::{environment, CatchUp, CompactCommit, SignedMessage};
 use super::{cost, benefit, Round, SetId};
 
 use std::collections::{HashMap, VecDeque};
+use std::pin::Pin;
+use std::task::{Context, Poll};
 use std::time::{Duration, Instant};
 
 const REBROADCAST_AFTER: Duration = Duration::from_secs(60 * 5);
@@ -1218,7 +1220,7 @@ impl ReportStream {
 	/// Consume the report stream, converting it into a future that
 	/// handles all reports.
 	pub(super) fn consume<B, N>(self, net: N)
-		-> impl Future<Item=(),Error=()> + Send + 'static
+		-> impl Future<Output=()> + Send + 'static + Unpin
 	where
 		B: BlockT,
 		N: super::Network<B> + Send + 'static,
@@ -1240,25 +1242,23 @@ struct ReportingTask<B, N> {
 }
 
 impl<B: BlockT, N: super::Network<B>> Future for ReportingTask<B, N> {
-	type Item = ();
-	type Error = ();
+	type Output = ();
 
-	fn poll(&mut self) -> Poll<(), ()> {
+	fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<()> {
 		loop {
-			match self.reports.poll() {
-				Err(_) => {
-					warn!(target: "afg", "Report stream terminated unexpectedly");
-					return Ok(Async::Ready(()))
-				}
-				Ok(Async::Ready(None)) => return Ok(Async::Ready(())),
-				Ok(Async::Ready(Some(PeerReport { who, cost_benefit }))) =>
+			match Stream::poll_next(Pin::new(&mut self.reports), cx) {
+				Poll::Ready(None) => return Poll::Ready(()),
+				Poll::Ready(Some(PeerReport { who, cost_benefit })) =>
 					self.net.report(who, cost_benefit),
-				Ok(Async::NotReady) => return Ok(Async::NotReady),
+				Poll::Pending => return Poll::Pending,
 			}
 		}
 	}
 }
 
+impl<B, N> Unpin for ReportingTask<B, N> {
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;