Make unbounded channels size warning exact (part 2)

client/consensus/common/src/import_queue/basic_queue.rs

@@ -118,8 +118,8 @@ impl<B: BlockT> BasicQueueHandle<B> {
 	}
 
 	pub fn close(&mut self) {
-		self.justification_sender.close_channel();
-		self.block_import_sender.close_channel();
+		self.justification_sender.close();
+		self.block_import_sender.close();
 	}
 }
 
@@ -597,11 +597,11 @@ mod tests {
 	fn prioritizes_finality_work_over_block_import() {
 		let (result_sender, mut result_port) = buffered_link::buffered_link(100_000);
 
-		let (worker, mut finality_sender, mut block_import_sender) =
+		let (worker, finality_sender, block_import_sender) =
 			BlockImportWorker::new(result_sender, (), Box::new(()), Some(Box::new(())), None);
 		futures::pin_mut!(worker);
 
-		let mut import_block = |n| {
+		let import_block = |n| {
 			let header = Header {
 				parent_hash: Hash::random(),
 				number: n,
@@ -612,35 +612,37 @@ mod tests {
 
 			let hash = header.hash();
 
-			block_on(block_import_sender.send(worker_messages::ImportBlocks(
-				BlockOrigin::Own,
-				vec![IncomingBlock {
-					hash,
-					header: Some(header),
-					body: None,
-					indexed_body: None,
-					justifications: None,
-					origin: None,
-					allow_missing_state: false,
-					import_existing: false,
-					state: None,
-					skip_execution: false,
-				}],
-			)))
-			.unwrap();
+			block_import_sender
+				.unbounded_send(worker_messages::ImportBlocks(
+					BlockOrigin::Own,
+					vec![IncomingBlock {
+						hash,
+						header: Some(header),
+						body: None,
+						indexed_body: None,
+						justifications: None,
+						origin: None,
+						allow_missing_state: false,
+						import_existing: false,
+						state: None,
+						skip_execution: false,
+					}],
+				))
+				.unwrap();
 
 			hash
 		};
 
-		let mut import_justification = || {
+		let import_justification = || {
 			let hash = Hash::random();
-			block_on(finality_sender.send(worker_messages::ImportJustification(
-				libp2p::PeerId::random(),
-				hash,
-				1,
-				(*b"TEST", Vec::new()),
-			)))
-			.unwrap();
+			finality_sender
+				.unbounded_send(worker_messages::ImportJustification(
+					libp2p::PeerId::random(),
+					hash,
+					1,
+					(*b"TEST", Vec::new()),
+				))
+				.unwrap();
 
 			hash
 		};

client/consensus/common/src/import_queue/buffered_link.rs

@@ -53,7 +53,7 @@ use super::BlockImportResult;
 /// can be used to buffer commands, and the receiver can be used to poll said commands and transfer
 /// them to another link. `queue_size_warning` sets the warning threshold of the channel queue size.
 pub fn buffered_link<B: BlockT>(
-	queue_size_warning: i64,
+	queue_size_warning: usize,
 ) -> (BufferedLinkSender<B>, BufferedLinkReceiver<B>) {
 	let (tx, rx) = tracing_unbounded("mpsc_buffered_link", queue_size_warning);
 	let tx = BufferedLinkSender { tx };
@@ -166,7 +166,7 @@ impl<B: BlockT> BufferedLinkReceiver<B> {
 	}
 
 	/// Close the channel.
-	pub fn close(&mut self) {
+	pub fn close(&mut self) -> bool {
 		self.rx.get_mut().close()
 	}
 }

client/service/src/task_manager/mod.rs

@@ -281,7 +281,7 @@ impl SpawnEssentialTaskHandle {
 		let essential_failed = self.essential_failed_tx.clone();
 		let essential_task = std::panic::AssertUnwindSafe(task).catch_unwind().map(move |_| {
 			log::error!("Essential task `{}` failed. Shutting down service.", name);
-			let _ = essential_failed.close_channel();
+			let _ = essential_failed.close();
 		});
 
 		let _ = self.inner.spawn_inner(name, group, essential_task, task_type);

client/service/test/Cargo.toml

@@ -12,6 +12,7 @@ repository = "https://github.com/paritytech/substrate/"
 targets = ["x86_64-unknown-linux-gnu"]
 
 [dependencies]
+async-channel = "1.8.0"
 array-bytes = "4.1"
 fdlimit = "0.2.1"
 futures = "0.3.21"

client/service/test/src/client/mod.rs

@@ -16,6 +16,7 @@
 // You should have received a copy of the GNU General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
+use async_channel::TryRecvError;
 use futures::executor::block_on;
 use parity_scale_codec::{Decode, Encode, Joiner};
 use sc_block_builder::BlockBuilderProvider;
@@ -175,16 +176,17 @@ fn finality_notification_check(
 	finalized: &[Hash],
 	stale_heads: &[Hash],
 ) {
-	match notifications.try_next() {
-		Ok(Some(notif)) => {
+	match notifications.try_recv() {
+		Ok(notif) => {
 			let stale_heads_expected: HashSet<_> = stale_heads.iter().collect();
 			let stale_heads: HashSet<_> = notif.stale_heads.iter().collect();
 			assert_eq!(notif.tree_route.as_ref(), &finalized[..finalized.len() - 1]);
 			assert_eq!(notif.hash, *finalized.last().unwrap());
 			assert_eq!(stale_heads, stale_heads_expected);
 		},
-		Ok(None) => panic!("unexpected notification result, client send channel was closed"),
-		Err(_) => assert!(finalized.is_empty()),
+		Err(TryRecvError::Closed) =>
+			panic!("unexpected notification result, client send channel was closed"),
+		Err(TryRecvError::Empty) => assert!(finalized.is_empty()),
 	}
 }
 
@@ -983,7 +985,7 @@ fn import_with_justification() {
 
 	finality_notification_check(&mut finality_notifications, &[a1.hash(), a2.hash()], &[]);
 	finality_notification_check(&mut finality_notifications, &[a3.hash()], &[]);
-	assert!(finality_notifications.try_next().is_err());
+	assert!(matches!(finality_notifications.try_recv().unwrap_err(), TryRecvError::Empty));
 }
 
 #[test]
@@ -1038,7 +1040,7 @@ fn importing_diverged_finalized_block_should_trigger_reorg() {
 	assert_eq!(client.chain_info().finalized_hash, b1.hash());
 
 	finality_notification_check(&mut finality_notifications, &[b1.hash()], &[a2.hash()]);
-	assert!(finality_notifications.try_next().is_err());
+	assert!(matches!(finality_notifications.try_recv().unwrap_err(), TryRecvError::Empty));
 }
 
 #[test]
@@ -1124,7 +1126,7 @@ fn finalizing_diverged_block_should_trigger_reorg() {
 
 	finality_notification_check(&mut finality_notifications, &[b1.hash()], &[]);
 	finality_notification_check(&mut finality_notifications, &[b2.hash(), b3.hash()], &[a2.hash()]);
-	assert!(finality_notifications.try_next().is_err());
+	assert!(matches!(finality_notifications.try_recv().unwrap_err(), TryRecvError::Empty));
 }
 
 #[test]
@@ -1227,7 +1229,7 @@ fn finality_notifications_content() {
 
 	finality_notification_check(&mut finality_notifications, &[a1.hash(), a2.hash()], &[c1.hash()]);
 	finality_notification_check(&mut finality_notifications, &[d3.hash(), d4.hash()], &[b2.hash()]);
-	assert!(finality_notifications.try_next().is_err());
+	assert!(matches!(finality_notifications.try_recv().unwrap_err(), TryRecvError::Empty));
 }
 
 #[test]
@@ -1437,7 +1439,7 @@ fn doesnt_import_blocks_that_revert_finality() {
 
 	finality_notification_check(&mut finality_notifications, &[a3.hash()], &[b2.hash()]);
 
-	assert!(finality_notifications.try_next().is_err());
+	assert!(matches!(finality_notifications.try_recv().unwrap_err(), TryRecvError::Empty));
 }
 
 #[test]

client/utils/Cargo.toml

@@ -10,13 +10,14 @@ description = "I/O for Substrate runtimes"
 readme = "README.md"
 
 [dependencies]
-backtrace = "0.3.67"
+async-channel = "1.8.0"
 futures = "0.3.21"
 futures-timer = "3.0.2"
 lazy_static = "1.4.0"
 log = "0.4"
 parking_lot = "0.12.1"
 prometheus = { version = "0.13.0", default-features = false }
+sp-arithmetic = { version = "6.0.0", default-features = false, path = "../../primitives/arithmetic" }
 
 [features]
 default = ["metered"]

client/utils/README.md

@@ -1,16 +1,11 @@
-Utilities Primitives for Substrate
+# Utilities Primitives for Substrate
 
-## Features
+This crate provides `mpsc::tracing_unbounded` function that returns wrapper types to
+`async_channel::Sender<T>` and `async_channel::Receiver<T>`, which register every
+`send`/`received`/`dropped` action happened on the channel.
 
-### metered
-
-This feature changes the behaviour of the function `mpsc::tracing_unbounded`.
-With the disabled feature this function is an alias to `futures::channel::mpsc::unbounded`.
-However, when the feature is enabled it creates wrapper types to `UnboundedSender<T>`
-and `UnboundedReceiver<T>` to register every `send`/`received`/`dropped` action happened on
-the channel.
-
-Also this feature creates and registers a prometheus vector with name `unbounded_channel_len` and labels:
+Also this wrapper creates and registers a prometheus vector with name `unbounded_channel_len`
+and labels:
 
 | Label        | Description                                   |
 | ------------ | --------------------------------------------- |

client/utils/src/lib.rs

@@ -18,17 +18,11 @@
 
 //! Utilities Primitives for Substrate
 //!
-//! # Features
+//! This crate provides `mpsc::tracing_unbounded` function that returns wrapper types to
+//! `async_channel::Sender<T>` and `async_channel::Receiver<T>`, which register every
+//! `send`/`received`/`dropped` action happened on the channel.
 //!
-//! ## metered
-//!
-//! This feature changes the behaviour of the function `mpsc::tracing_unbounded`.
-//! With the disabled feature this function is an alias to `futures::channel::mpsc::unbounded`.
-//! However, when the feature is enabled it creates wrapper types to `UnboundedSender<T>`
-//! and `UnboundedReceiver<T>` to register every `send`/`received`/`dropped` action happened on
-//! the channel.
-//!
-//! Also this feature creates and registers a prometheus vector with name `unbounded_channel_len`
+//! Also this wrapper creates and registers a prometheus vector with name `unbounded_channel_len`
 //! and labels:
 //!
 //! | Label        | Description                                   |

client/utils/src/metrics.rs

@@ -24,7 +24,6 @@ use prometheus::{
 	Error as PrometheusError, Registry,
 };
 
-#[cfg(feature = "metered")]
 use prometheus::{core::GenericCounterVec, Opts};
 
 lazy_static! {
@@ -36,7 +35,6 @@ lazy_static! {
 			.expect("Creating of statics doesn't fail. qed");
 }
 
-#[cfg(feature = "metered")]
 lazy_static! {
 	pub static ref UNBOUNDED_CHANNELS_COUNTER : GenericCounterVec<AtomicU64> = GenericCounterVec::new(
 		Opts::new("substrate_unbounded_channel_len", "Items in each mpsc::unbounded instance"),
@@ -49,8 +47,6 @@ lazy_static! {
 pub fn register_globals(registry: &Registry) -> Result<(), PrometheusError> {
 	registry.register(Box::new(TOKIO_THREADS_ALIVE.clone()))?;
 	registry.register(Box::new(TOKIO_THREADS_TOTAL.clone()))?;
-
-	#[cfg(feature = "metered")]
 	registry.register(Box::new(UNBOUNDED_CHANNELS_COUNTER.clone()))?;
 
 	Ok(())