Make unbounded channels size warning exact (part 2) (#13504)

substrate/Cargo.lock

@@ -9231,6 +9231,7 @@ name = "sc-service-test"
 version = "2.0.0"
 dependencies = [
  "array-bytes",
+ "async-channel",
  "fdlimit",
  "futures",
  "log",
@@ -9435,13 +9436,14 @@ dependencies = [
 name = "sc-utils"
 version = "4.0.0-dev"
 dependencies = [
- "backtrace",
+ "async-channel",
  "futures",
  "futures-timer",
  "lazy_static",
  "log",
  "parking_lot 0.12.1",
  "prometheus",
+ "sp-arithmetic",
  "tokio-test",
 ]
 

substrate/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
 		};

substrate/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()
 	}
 }

substrate/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);

substrate/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"

substrate/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]

substrate/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"]

substrate/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                                   |
 | ------------ | --------------------------------------------- |

substrate/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                                   |

substrate/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(())

substrate/client/utils/src/mpsc.rs

@@ -16,287 +16,164 @@
 // You should have received a copy of the GNU General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-//! Features to meter unbounded channels
+//! Code to meter unbounded channels.
 
-#[cfg(not(feature = "metered"))]
-mod inner {
-	// just aliased, non performance implications
-	use futures::channel::mpsc::{self, UnboundedReceiver, UnboundedSender};
-	pub type TracingUnboundedSender<T> = UnboundedSender<T>;
-	pub type TracingUnboundedReceiver<T> = UnboundedReceiver<T>;
+use crate::metrics::UNBOUNDED_CHANNELS_COUNTER;
+use async_channel::{Receiver, Sender, TryRecvError, TrySendError};
+use futures::{
+	stream::{FusedStream, Stream},
+	task::{Context, Poll},
+};
+use log::error;
+use sp_arithmetic::traits::SaturatedConversion;
+use std::{
+	backtrace::Backtrace,
+	pin::Pin,
+	sync::{
+		atomic::{AtomicBool, Ordering},
+		Arc,
+	},
+};
 
-	/// Alias `mpsc::unbounded`
-	pub fn tracing_unbounded<T>(
-		_key: &'static str,
-	) -> (TracingUnboundedSender<T>, TracingUnboundedReceiver<T>) {
-		mpsc::unbounded()
-	}
+/// Wrapper Type around [`async_channel::Sender`] that increases the global
+/// measure when a message is added.
+#[derive(Debug)]
+pub struct TracingUnboundedSender<T> {
+	inner: Sender<T>,
+	name: &'static str,
+	queue_size_warning: usize,
+	warning_fired: Arc<AtomicBool>,
+	creation_backtrace: Arc<Backtrace>,
 }
 
-#[cfg(feature = "metered")]
-mod inner {
-	// tracing implementation
-	use crate::metrics::UNBOUNDED_CHANNELS_COUNTER;
-	use backtrace::Backtrace;
-	use futures::{
-		channel::mpsc::{
-			self, SendError, TryRecvError, TrySendError, UnboundedReceiver, UnboundedSender,
-		},
-		sink::Sink,
-		stream::{FusedStream, Stream},
-		task::{Context, Poll},
-	};
-	use log::error;
-	use std::{
-		pin::Pin,
-		sync::{
-			atomic::{AtomicBool, AtomicI64, Ordering},
-			Arc,
-		},
-	};
-
-	/// Wrapper Type around `UnboundedSender` that increases the global
-	/// measure when a message is added
-	#[derive(Debug)]
-	pub struct TracingUnboundedSender<T> {
-		inner: UnboundedSender<T>,
-		name: &'static str,
-		// To not bother with ordering and possible underflow errors of the unsigned counter
-		// we just use `i64` and `Ordering::Relaxed`, and perceive `queue_size` as approximate.
-		// It can turn < 0 though.
-		queue_size: Arc<AtomicI64>,
-		queue_size_warning: i64,
-		warning_fired: Arc<AtomicBool>,
-		creation_backtrace: Arc<Backtrace>,
-	}
-
-	// Strangely, deriving `Clone` requires that `T` is also `Clone`.
-	impl<T> Clone for TracingUnboundedSender<T> {
-		fn clone(&self) -> Self {
-			Self {
-				inner: self.inner.clone(),
-				name: self.name,
-				queue_size: self.queue_size.clone(),
-				queue_size_warning: self.queue_size_warning,
-				warning_fired: self.warning_fired.clone(),
-				creation_backtrace: self.creation_backtrace.clone(),
-			}
-		}
-	}
-
-	/// Wrapper Type around `UnboundedReceiver` that decreases the global
-	/// measure when a message is polled
-	#[derive(Debug)]
-	pub struct TracingUnboundedReceiver<T> {
-		inner: UnboundedReceiver<T>,
-		name: &'static str,
-		queue_size: Arc<AtomicI64>,
-	}
-
-	/// Wrapper around `mpsc::unbounded` that tracks the in- and outflow via
-	/// `UNBOUNDED_CHANNELS_COUNTER` and warns if the message queue grows
-	/// above the warning threshold.
-	pub fn tracing_unbounded<T>(
-		name: &'static str,
-		queue_size_warning: i64,
-	) -> (TracingUnboundedSender<T>, TracingUnboundedReceiver<T>) {
-		let (s, r) = mpsc::unbounded();
-		let queue_size = Arc::new(AtomicI64::new(0));
-		let sender = TracingUnboundedSender {
-			inner: s,
-			name,
-			queue_size: queue_size.clone(),
-			queue_size_warning,
-			warning_fired: Arc::new(AtomicBool::new(false)),
-			creation_backtrace: Arc::new(Backtrace::new_unresolved()),
-		};
-		let receiver = TracingUnboundedReceiver { inner: r, name, queue_size };
-		(sender, receiver)
-	}
-
-	impl<T> TracingUnboundedSender<T> {
-		/// Proxy function to mpsc::UnboundedSender
-		pub fn poll_ready(&self, ctx: &mut Context) -> Poll<Result<(), SendError>> {
-			self.inner.poll_ready(ctx)
-		}
-
-		/// Proxy function to mpsc::UnboundedSender
-		pub fn is_closed(&self) -> bool {
-			self.inner.is_closed()
-		}
-
-		/// Proxy function to mpsc::UnboundedSender
-		pub fn close_channel(&self) {
-			self.inner.close_channel()
-		}
-
-		/// Proxy function to mpsc::UnboundedSender
-		pub fn disconnect(&mut self) {
-			self.inner.disconnect()
-		}
-
-		pub fn start_send(&mut self, msg: T) -> Result<(), SendError> {
-			// The underlying implementation of [`UnboundedSender::start_send`] is the same as
-			// [`UnboundedSender::unbounded_send`], so we just reuse the message counting and
-			// error reporting code from `unbounded_send`.
-			self.unbounded_send(msg).map_err(TrySendError::into_send_error)
-		}
-
-		/// Proxy function to mpsc::UnboundedSender
-		pub fn unbounded_send(&self, msg: T) -> Result<(), TrySendError<T>> {
-			self.inner.unbounded_send(msg).map(|s| {
-				UNBOUNDED_CHANNELS_COUNTER.with_label_values(&[self.name, "send"]).inc();
-
-				let queue_size = self.queue_size.fetch_add(1, Ordering::Relaxed);
-				if queue_size == self.queue_size_warning &&
-					self.warning_fired
-						.compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed)
-						.is_ok()
-				{
-					// `warning_fired` and `queue_size` are not synchronized, so it's possible
-					// that the warning is fired few times before the `warning_fired` is seen
-					// by all threads. This seems better than introducing a mutex guarding them.
-					let mut backtrace = (*self.creation_backtrace).clone();
-					backtrace.resolve();
-					error!(
-						"The number of unprocessed messages in channel `{}` reached {}.\n\
-						 The channel was created at:\n{:?}",
-						self.name, self.queue_size_warning, backtrace,
-					);
-				}
-
-				s
-			})
-		}
-
-		/// Proxy function to mpsc::UnboundedSender
-		pub fn same_receiver(&self, other: &UnboundedSender<T>) -> bool {
-			self.inner.same_receiver(other)
-		}
-	}
-
-	impl<T> TracingUnboundedReceiver<T> {
-		fn consume(&mut self) {
-			// consume all items, make sure to reflect the updated count
-			let mut count = 0;
-			loop {
-				if self.inner.is_terminated() {
-					break
-				}
-
-				match self.try_next() {
-					Ok(Some(..)) => count += 1,
-					_ => break,
-				}
-			}
-			// and discount the messages
-			if count > 0 {
-				UNBOUNDED_CHANNELS_COUNTER
-					.with_label_values(&[self.name, "dropped"])
-					.inc_by(count);
-			}
-		}
-
-		/// Proxy function to mpsc::UnboundedReceiver
-		/// that consumes all messages first and updates the counter
-		pub fn close(&mut self) {
-			self.consume();
-			self.inner.close()
-		}
-
-		/// Proxy function to mpsc::UnboundedReceiver
-		/// that discounts the messages taken out
-		pub fn try_next(&mut self) -> Result<Option<T>, TryRecvError> {
-			self.inner.try_next().map(|s| {
-				if s.is_some() {
-					let _ = self.queue_size.fetch_sub(1, Ordering::Relaxed);
-					UNBOUNDED_CHANNELS_COUNTER.with_label_values(&[self.name, "received"]).inc();
-				}
-				s
-			})
-		}
-	}
-
-	impl<T> Drop for TracingUnboundedReceiver<T> {
-		fn drop(&mut self) {
-			self.consume();
-		}
-	}
-
-	impl<T> Unpin for TracingUnboundedReceiver<T> {}
-
-	impl<T> Stream for TracingUnboundedReceiver<T> {
-		type Item = T;
-
-		fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
-			let s = self.get_mut();
-			match Pin::new(&mut s.inner).poll_next(cx) {
-				Poll::Ready(msg) => {
-					if msg.is_some() {
-						let _ = s.queue_size.fetch_sub(1, Ordering::Relaxed);
-						UNBOUNDED_CHANNELS_COUNTER.with_label_values(&[s.name, "received"]).inc();
-					}
-					Poll::Ready(msg)
-				},
-				Poll::Pending => Poll::Pending,
-			}
-		}
-	}
-
-	impl<T> FusedStream for TracingUnboundedReceiver<T> {
-		fn is_terminated(&self) -> bool {
-			self.inner.is_terminated()
-		}
-	}
-
-	impl<T> Sink<T> for TracingUnboundedSender<T> {
-		type Error = SendError;
-
-		fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-			TracingUnboundedSender::poll_ready(&*self, cx)
-		}
-
-		fn start_send(mut self: Pin<&mut Self>, msg: T) -> Result<(), Self::Error> {
-			TracingUnboundedSender::start_send(&mut *self, msg)
-		}
-
-		fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-			Poll::Ready(Ok(()))
-		}
-
-		fn poll_close(
-			mut self: Pin<&mut Self>,
-			_: &mut Context<'_>,
-		) -> Poll<Result<(), Self::Error>> {
-			self.disconnect();
-			Poll::Ready(Ok(()))
-		}
-	}
-
-	impl<T> Sink<T> for &TracingUnboundedSender<T> {
-		type Error = SendError;
-
-		fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-			TracingUnboundedSender::poll_ready(*self, cx)
-		}
-
-		fn start_send(self: Pin<&mut Self>, msg: T) -> Result<(), Self::Error> {
-			self.unbounded_send(msg).map_err(TrySendError::into_send_error)
-		}
-
-		fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-			Poll::Ready(Ok(()))
-		}
-
-		fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-			// The difference with `TracingUnboundedSender` is intentional. The underlying
-			// implementation differs for `UnboundedSender<T>` and `&UnboundedSender<T>`:
-			// the latter closes the channel completely with `close_channel()`, while the former
-			// only closes this specific sender with `disconnect()`.
-			self.close_channel();
-			Poll::Ready(Ok(()))
+// Strangely, deriving `Clone` requires that `T` is also `Clone`.
+impl<T> Clone for TracingUnboundedSender<T> {
+	fn clone(&self) -> Self {
+		Self {
+			inner: self.inner.clone(),
+			name: self.name,
+			queue_size_warning: self.queue_size_warning,
+			warning_fired: self.warning_fired.clone(),
+			creation_backtrace: self.creation_backtrace.clone(),
 		}
 	}
 }
 
-pub use inner::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};
+/// Wrapper Type around [`async_channel::Receiver`] that decreases the global
+/// measure when a message is polled.
+#[derive(Debug)]
+pub struct TracingUnboundedReceiver<T> {
+	inner: Receiver<T>,
+	name: &'static str,
+}
+
+/// Wrapper around [`async_channel::unbounded`] that tracks the in- and outflow via
+/// `UNBOUNDED_CHANNELS_COUNTER` and warns if the message queue grows
+/// above the warning threshold.
+pub fn tracing_unbounded<T>(
+	name: &'static str,
+	queue_size_warning: usize,
+) -> (TracingUnboundedSender<T>, TracingUnboundedReceiver<T>) {
+	let (s, r) = async_channel::unbounded();
+	let sender = TracingUnboundedSender {
+		inner: s,
+		name,
+		queue_size_warning,
+		warning_fired: Arc::new(AtomicBool::new(false)),
+		creation_backtrace: Arc::new(Backtrace::force_capture()),
+	};
+	let receiver = TracingUnboundedReceiver { inner: r, name };
+	(sender, receiver)
+}
+
+impl<T> TracingUnboundedSender<T> {
+	/// Proxy function to [`async_channel::Sender`].
+	pub fn is_closed(&self) -> bool {
+		self.inner.is_closed()
+	}
+
+	/// Proxy function to [`async_channel::Sender`].
+	pub fn close(&self) -> bool {
+		self.inner.close()
+	}
+
+	/// Proxy function to `async_channel::Sender::try_send`.
+	pub fn unbounded_send(&self, msg: T) -> Result<(), TrySendError<T>> {
+		self.inner.try_send(msg).map(|s| {
+			UNBOUNDED_CHANNELS_COUNTER.with_label_values(&[self.name, "send"]).inc();
+
+			if self.inner.len() >= self.queue_size_warning &&
+				self.warning_fired
+					.compare_exchange(false, true, Ordering::Relaxed, Ordering::Relaxed)
+					.is_ok()
+			{
+				error!(
+					"The number of unprocessed messages in channel `{}` exceeded {}.\n\
+					 The channel was created at:\n{}\n
+					 Last message was sent from:\n{}",
+					self.name,
+					self.queue_size_warning,
+					self.creation_backtrace,
+					Backtrace::force_capture(),
+				);
+			}
+
+			s
+		})
+	}
+}
+
+impl<T> TracingUnboundedReceiver<T> {
+	/// Proxy function to [`async_channel::Receiver`].
+	pub fn close(&mut self) -> bool {
+		self.inner.close()
+	}
+
+	/// Proxy function to [`async_channel::Receiver`]
+	/// that discounts the messages taken out.
+	pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
+		self.inner.try_recv().map(|s| {
+			UNBOUNDED_CHANNELS_COUNTER.with_label_values(&[self.name, "received"]).inc();
+			s
+		})
+	}
+}
+
+impl<T> Drop for TracingUnboundedReceiver<T> {
+	fn drop(&mut self) {
+		self.close();
+		// the number of messages about to be dropped
+		let count = self.inner.len();
+		// discount the messages
+		if count > 0 {
+			UNBOUNDED_CHANNELS_COUNTER
+				.with_label_values(&[self.name, "dropped"])
+				.inc_by(count.saturated_into());
+		}
+	}
+}
+
+impl<T> Unpin for TracingUnboundedReceiver<T> {}
+
+impl<T> Stream for TracingUnboundedReceiver<T> {
+	type Item = T;
+
+	fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
+		let s = self.get_mut();
+		match Pin::new(&mut s.inner).poll_next(cx) {
+			Poll::Ready(msg) => {
+				if msg.is_some() {
+					UNBOUNDED_CHANNELS_COUNTER.with_label_values(&[s.name, "received"]).inc();
+				}
+				Poll::Ready(msg)
+			},
+			Poll::Pending => Poll::Pending,
+		}
+	}
+}
+
+impl<T> FusedStream for TracingUnboundedReceiver<T> {
+	fn is_terminated(&self) -> bool {
+		self.inner.is_terminated()
+	}
+}

substrate/client/utils/src/notification.rs

@@ -79,7 +79,7 @@ impl<Payload, TK: TracingKeyStr> NotificationStream<Payload, TK> {
 	}
 
 	/// Subscribe to a channel through which the generic payload can be received.
-	pub fn subscribe(&self, queue_size_warning: i64) -> NotificationReceiver<Payload> {
+	pub fn subscribe(&self, queue_size_warning: usize) -> NotificationReceiver<Payload> {
 		let receiver = self.hub.subscribe((), queue_size_warning);
 		NotificationReceiver { receiver }
 	}

substrate/client/utils/src/pubsub.rs

@@ -164,7 +164,7 @@ impl<M, R> Hub<M, R> {
 	/// Subscribe to this Hub using the `subs_key: K`.
 	///
 	/// A subscription with a key `K` is possible if the Registry implements `Subscribe<K>`.
-	pub fn subscribe<K>(&self, subs_key: K, queue_size_warning: i64) -> Receiver<M, R>
+	pub fn subscribe<K>(&self, subs_key: K, queue_size_warning: usize) -> Receiver<M, R>
 	where
 		R: Subscribe<K> + Unsubscribe,
 	{

substrate/client/utils/src/pubsub/tests/normal_operation.rs

@@ -37,9 +37,8 @@ fn positive_rx_receives_relevant_messages_and_terminates_upon_hub_drop() {
 		// Hub is disposed. The rx_01 should be over after that.
 		std::mem::drop(hub);
 
-		assert!(!rx_01.is_terminated());
-		assert_eq!(None, rx_01.next().await);
 		assert!(rx_01.is_terminated());
+		assert_eq!(None, rx_01.next().await);
 	});
 }