Switch GrandPa to new futures

bin/node-template/src/service.rs

@@ -154,6 +154,8 @@ pub fn new_full<C: Send + Default + 'static>(config: Configuration<C, GenesisCon
 		is_authority,
 	};
 
+	use futures::prelude::*;
+
 	match (is_authority, disable_grandpa) {
 		(false, false) => {
 			// start the lightweight GRANDPA observer
@@ -163,7 +165,7 @@ pub fn new_full<C: Send + Default + 'static>(config: Configuration<C, GenesisCon
 				service.network(),
 				service.on_exit(),
 				service.spawn_task_handle(),
-			)?);
+			)?.unit_error().compat());
 		},
 		(true, false) => {
 			// start the full GRANDPA voter
@@ -180,7 +182,8 @@ pub fn new_full<C: Send + Default + 'static>(config: Configuration<C, GenesisCon
 
 			// the GRANDPA voter task is considered infallible, i.e.
 			// if it fails we take down the service with it.
-			service.spawn_essential_task(grandpa::run_grandpa_voter(voter_config)?);
+			service.spawn_essential_task(grandpa::run_grandpa_voter(voter_config)?
+				.unit_error().compat());
 		},
 		(_, true) => {
 			grandpa::setup_disabled_grandpa(

bin/node/cli/src/service.rs

@@ -223,7 +223,7 @@ macro_rules! new_full {
 					service.network(),
 					service.on_exit(),
 					service.spawn_task_handle(),
-				)?);
+				)?.unit_error().compat());
 			},
 			(true, false) => {
 				// start the full GRANDPA voter
@@ -239,7 +239,8 @@ macro_rules! new_full {
 				};
 				// the GRANDPA voter task is considered infallible, i.e.
 				// if it fails we take down the service with it.
-				service.spawn_essential_task(grandpa::run_grandpa_voter(grandpa_config)?);
+				service.spawn_essential_task(grandpa::run_grandpa_voter(grandpa_config)?
+					.map(|()| Ok::<(), ()>(())).compat());
 			},
 			(_, true) => {
 				grandpa::setup_disabled_grandpa(

client/finality-grandpa/Cargo.toml

@@ -6,8 +6,8 @@ edition = "2018"
 
 [dependencies]
 fork-tree = { version = "2.0.0", path = "../../utils/fork-tree" }
-futures = "0.1.29"
-futures03 = { package = "futures", version = "0.3.1", features = ["compat"] }
+futures = "0.3.1"
+futures01 = { package = "futures", version = "0.1.29" }
 futures-timer = "2.0.2"
 log = "0.4.8"
 parking_lot = "0.9.0"
@@ -27,10 +27,11 @@ sc-network = { version = "0.8", path = "../network" }
 sc-network-gossip = { version = "2.0.0", path = "../network-gossip" }
 sp-finality-tracker = { version = "2.0.0", path = "../../primitives/finality-tracker" }
 sp-finality-grandpa = { version = "2.0.0", path = "../../primitives/finality-grandpa" }
-finality-grandpa = { version = "0.10.1", features = ["derive-codec"] }
+# See https://github.com/paritytech/finality-grandpa/pull/100
+finality-grandpa = { git = "https://github.com/expenses/finality-grandpa", branch = "future", features = ["derive-codec"] }
 
 [dev-dependencies]
-finality-grandpa = { version = "0.10.1", features = ["derive-codec", "test-helpers"] }
+finality-grandpa = { git = "https://github.com/expenses/finality-grandpa", branch = "future", features = ["derive-codec", "test-helpers"] }
 sc-network = { version = "0.8", path = "../network" }
 sc-network-test = { version = "2.0.0", path = "../network/test" }
 sp-keyring = { version = "2.0.0", path = "../../primitives/keyring" }

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

@@ -89,16 +89,18 @@ use parity_scale_codec::{Encode, Decode};
 use sp_finality_grandpa::AuthorityId;
 
 use sc_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 rand::seq::SliceRandom;
 
 use crate::{environment, CatchUp, CompactCommit, SignedMessage};
 use super::{cost, benefit, Round, SetId};
 
 use std::collections::{HashMap, VecDeque, HashSet};
 use std::time::{Duration, Instant};
+use std::pin::Pin;
+use std::task::{Poll, Context};
 
 const REBROADCAST_AFTER: Duration = Duration::from_secs(60 * 5);
 const CATCH_UP_REQUEST_TIMEOUT: Duration = Duration::from_secs(45);
@@ -1460,7 +1462,7 @@ impl ReportStream {
 	/// Consume the report stream, converting it into a future that
 	/// handles all reports.
 	pub(super) fn consume<B>(self, net: GossipEngine<B>)
-		-> impl Future<Item=(),Error=()> + Send + 'static
+		-> impl Future<Output=()> + Send + 'static + Unpin
 	where
 		B: BlockT,
 	{
@@ -1479,20 +1481,15 @@ struct ReportingTask<B: BlockT> {
 }
 
 impl<B: BlockT> Future for ReportingTask<B> {
-	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,
 			}
 		}
 	}