Runtime diagnostics for leaked messages in unbounded channels (#12971)

substrate/client/utils/src/mpsc.rs

@@ -45,71 +45,136 @@ mod inner {
 		stream::{FusedStream, Stream},
 		task::{Context, Poll},
 	};
-	use std::pin::Pin;
+	use log::error;
+	use std::{
+		backtrace::{Backtrace, BacktraceStatus},
+		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>(&'static str, UnboundedSender<T>);
+	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(self.0, self.1.clone())
+			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>(&'static str, UnboundedReceiver<T>);
+	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`
+	/// `UNBOUNDED_CHANNELS_COUNTER` and warns if the message queue grows
+	/// above the warning threshold.
 	pub fn tracing_unbounded<T>(
-		key: &'static str,
+		name: &'static str,
+		queue_size_warning: i64,
 	) -> (TracingUnboundedSender<T>, TracingUnboundedReceiver<T>) {
 		let (s, r) = mpsc::unbounded();
-		(TracingUnboundedSender(key, s), TracingUnboundedReceiver(key, r))
+		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::capture()),
+		};
+		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.1.poll_ready(ctx)
+			self.inner.poll_ready(ctx)
 		}
 
 		/// Proxy function to mpsc::UnboundedSender
 		pub fn is_closed(&self) -> bool {
-			self.1.is_closed()
+			self.inner.is_closed()
 		}
 
 		/// Proxy function to mpsc::UnboundedSender
 		pub fn close_channel(&self) {
-			self.1.close_channel()
+			self.inner.close_channel()
 		}
 
 		/// Proxy function to mpsc::UnboundedSender
 		pub fn disconnect(&mut self) {
-			self.1.disconnect()
+			self.inner.disconnect()
 		}
 
-		/// Proxy function to mpsc::UnboundedSender
 		pub fn start_send(&mut self, msg: T) -> Result<(), SendError> {
-			self.1.start_send(msg)
+			// 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.1.unbounded_send(msg).map(|s| {
-				UNBOUNDED_CHANNELS_COUNTER.with_label_values(&[self.0, "send"]).inc();
+			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.load(Ordering::Relaxed)
+				{
+					// `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.
+					self.warning_fired.store(true, Ordering::Relaxed);
+					match self.creation_backtrace.status() {
+						BacktraceStatus::Captured => error!(
+							"The number of unprocessed messages in channel `{}` reached {}.\n\
+							 The channel was created at:\n{}",
+							self.name, self.queue_size_warning, self.creation_backtrace,
+						),
+						_ => error!(
+							"The number of unprocessed messages in channel `{}` reached {}.",
+							self.name, self.queue_size_warning,
+						),
+					}
+				}
+
 				s
 			})
 		}
 
 		/// Proxy function to mpsc::UnboundedSender
 		pub fn same_receiver(&self, other: &UnboundedSender<T>) -> bool {
-			self.1.same_receiver(other)
+			self.inner.same_receiver(other)
 		}
 	}
 
@@ -118,7 +183,7 @@ mod inner {
 			// consume all items, make sure to reflect the updated count
 			let mut count = 0;
 			loop {
-				if self.1.is_terminated() {
+				if self.inner.is_terminated() {
 					break
 				}
 
@@ -129,7 +194,9 @@ mod inner {
 			}
 			// and discount the messages
 			if count > 0 {
-				UNBOUNDED_CHANNELS_COUNTER.with_label_values(&[self.0, "dropped"]).inc_by(count);
+				UNBOUNDED_CHANNELS_COUNTER
+					.with_label_values(&[self.name, "dropped"])
+					.inc_by(count);
 			}
 		}
 
@@ -137,15 +204,16 @@ mod inner {
 		/// that consumes all messages first and updates the counter
 		pub fn close(&mut self) {
 			self.consume();
-			self.1.close()
+			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.1.try_next().map(|s| {
+			self.inner.try_next().map(|s| {
 				if s.is_some() {
-					UNBOUNDED_CHANNELS_COUNTER.with_label_values(&[self.0, "received"]).inc();
+					let _ = self.queue_size.fetch_sub(1, Ordering::Relaxed);
+					UNBOUNDED_CHANNELS_COUNTER.with_label_values(&[self.name, "received"]).inc();
 				}
 				s
 			})
@@ -165,10 +233,11 @@ mod inner {
 
 		fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<T>> {
 			let s = self.get_mut();
-			match Pin::new(&mut s.1).poll_next(cx) {
+			match Pin::new(&mut s.inner).poll_next(cx) {
 				Poll::Ready(msg) => {
 					if msg.is_some() {
-						UNBOUNDED_CHANNELS_COUNTER.with_label_values(&[s.0, "received"]).inc();
+						let _ = s.queue_size.fetch_sub(1, Ordering::Relaxed);
+						UNBOUNDED_CHANNELS_COUNTER.with_label_values(&[s.name, "received"]).inc();
 					}
 					Poll::Ready(msg)
 				},
@@ -179,7 +248,7 @@ mod inner {
 
 	impl<T> FusedStream for TracingUnboundedReceiver<T> {
 		fn is_terminated(&self) -> bool {
-			self.1.is_terminated()
+			self.inner.is_terminated()
 		}
 	}
 
@@ -223,6 +292,10 @@ mod inner {
 		}
 
 		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(()))
 		}

substrate/client/utils/src/notification.rs

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

substrate/client/utils/src/notification/tests.rs

@@ -36,7 +36,7 @@ fn notification_channel_simple() {
 
 	// Create a future to receive a single notification
 	// from the stream and verify its payload.
-	let future = stream.subscribe().take(1).for_each(move |payload| {
+	let future = stream.subscribe(100_000).take(1).for_each(move |payload| {
 		let test_payload = closure_payload.clone();
 		async move {
 			assert_eq!(payload, test_payload);

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) -> Receiver<M, R>
+	pub fn subscribe<K>(&self, subs_key: K, queue_size_warning: i64) -> Receiver<M, R>
 	where
 		R: Subscribe<K> + Unsubscribe,
 	{
@@ -178,7 +178,7 @@ impl<M, R> Hub<M, R> {
 		// have the sink disposed.
 		shared_borrowed.registry.subscribe(subs_key, subs_id);
 
-		let (tx, rx) = crate::mpsc::tracing_unbounded(self.tracing_key);
+		let (tx, rx) = crate::mpsc::tracing_unbounded(self.tracing_key, queue_size_warning);
 		assert!(shared_borrowed.sinks.insert(subs_id, tx).is_none(), "Used IDSequence to create another ID. Should be unique until u64 is overflowed. Should be unique.");
 
 		Receiver { shared: Arc::downgrade(&self.shared), subs_id, rx }

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

@@ -27,7 +27,7 @@ fn positive_rx_receives_relevant_messages_and_terminates_upon_hub_drop() {
 		// No subscribers yet. That message is not supposed to get to anyone.
 		hub.send(0);
 
-		let mut rx_01 = hub.subscribe(SubsKey::new());
+		let mut rx_01 = hub.subscribe(SubsKey::new(), 100_000);
 		assert_eq!(hub.subs_count(), 1);
 
 		// That message is sent after subscription. Should be delivered into rx_01.
@@ -49,9 +49,9 @@ fn positive_subs_count_is_correct_upon_drop_of_rxs() {
 		let hub = TestHub::new(TK);
 		assert_eq!(hub.subs_count(), 0);
 
-		let rx_01 = hub.subscribe(SubsKey::new());
+		let rx_01 = hub.subscribe(SubsKey::new(), 100_000);
 		assert_eq!(hub.subs_count(), 1);
-		let rx_02 = hub.subscribe(SubsKey::new());
+		let rx_02 = hub.subscribe(SubsKey::new(), 100_000);
 		assert_eq!(hub.subs_count(), 2);
 
 		std::mem::drop(rx_01);
@@ -69,11 +69,11 @@ fn positive_subs_count_is_correct_upon_drop_of_rxs_on_cloned_hubs() {
 		assert_eq!(hub_01.subs_count(), 0);
 		assert_eq!(hub_02.subs_count(), 0);
 
-		let rx_01 = hub_02.subscribe(SubsKey::new());
+		let rx_01 = hub_02.subscribe(SubsKey::new(), 100_000);
 		assert_eq!(hub_01.subs_count(), 1);
 		assert_eq!(hub_02.subs_count(), 1);
 
-		let rx_02 = hub_02.subscribe(SubsKey::new());
+		let rx_02 = hub_02.subscribe(SubsKey::new(), 100_000);
 		assert_eq!(hub_01.subs_count(), 2);
 		assert_eq!(hub_02.subs_count(), 2);
 

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

@@ -30,7 +30,7 @@ fn t01() {
 	let hub = TestHub::new(TK);
 	assert_hub_props(&hub, 0, 0);
 
-	let rx_01 = hub.subscribe(SubsKey::new());
+	let rx_01 = hub.subscribe(SubsKey::new(), 100_000);
 	assert_hub_props(&hub, 1, 1);
 
 	std::mem::drop(rx_01);
@@ -45,17 +45,17 @@ fn t02() {
 		assert_hub_props(&hub, 0, 0);
 
 		// Subscribe rx-01
-		let rx_01 = hub.subscribe(SubsKey::new());
+		let rx_01 = hub.subscribe(SubsKey::new(), 100_000);
 		assert_hub_props(&hub, 1, 1);
 
 		// Subscribe rx-02 so that its unsubscription will lead to an attempt to drop rx-01 in the
 		// middle of unsubscription of rx-02
-		let rx_02 = hub.subscribe(SubsKey::new().with_receiver(rx_01));
+		let rx_02 = hub.subscribe(SubsKey::new().with_receiver(rx_01), 100_000);
 		assert_hub_props(&hub, 2, 2);
 
 		// Subscribe rx-03 in order to see that it will receive messages after the unclean
 		// unsubscription
-		let mut rx_03 = hub.subscribe(SubsKey::new());
+		let mut rx_03 = hub.subscribe(SubsKey::new(), 100_000);
 		assert_hub_props(&hub, 3, 3);
 
 		// drop rx-02 leads to an attempt to unsubscribe rx-01
@@ -69,7 +69,7 @@ fn t02() {
 
 		// Subscribe rx-04 in order to see that it will receive messages after the unclean
 		// unsubscription
-		let mut rx_04 = hub.subscribe(SubsKey::new());
+		let mut rx_04 = hub.subscribe(SubsKey::new(), 100_000);
 		assert_hub_props(&hub, 3, 3);
 
 		hub.send(2);
@@ -96,8 +96,8 @@ fn t02() {
 }
 
 async fn add_some_subscribers_see_that_messages_are_delivered_and_unsubscribe(hub: &TestHub) {
-	let rx_01 = hub.subscribe(SubsKey::new());
-	let rx_02 = hub.subscribe(SubsKey::new());
+	let rx_01 = hub.subscribe(SubsKey::new(), 100_000);
+	let rx_02 = hub.subscribe(SubsKey::new(), 100_000);
 
 	hub.send(1);
 	hub.send(2);
@@ -121,9 +121,8 @@ fn t03() {
 		add_some_subscribers_see_that_messages_are_delivered_and_unsubscribe(&hub).await;
 		assert_hub_props(&hub, 0, 0);
 
-		assert!(catch_unwind(AssertUnwindSafe(
-			|| hub.subscribe(SubsKey::new().with_panic(SubsKeyPanic::OnSubscribePanicBefore))
-		))
+		assert!(catch_unwind(AssertUnwindSafe(|| hub
+			.subscribe(SubsKey::new().with_panic(SubsKeyPanic::OnSubscribePanicBefore), 100_000)))
 		.is_err());
 
 		assert_hub_props(&hub, 0, 0);
@@ -141,9 +140,8 @@ fn t04() {
 		add_some_subscribers_see_that_messages_are_delivered_and_unsubscribe(&hub).await;
 		assert_hub_props(&hub, 0, 0);
 
-		assert!(catch_unwind(AssertUnwindSafe(
-			|| hub.subscribe(SubsKey::new().with_panic(SubsKeyPanic::OnSubscribePanicAfter))
-		))
+		assert!(catch_unwind(AssertUnwindSafe(|| hub
+			.subscribe(SubsKey::new().with_panic(SubsKeyPanic::OnSubscribePanicAfter), 100_000)))
 		.is_err());
 
 		// the registry has panicked after it has added a subs-id into its internal storage — the
@@ -163,8 +161,8 @@ fn t05() {
 		add_some_subscribers_see_that_messages_are_delivered_and_unsubscribe(&hub).await;
 		assert_hub_props(&hub, 0, 0);
 
-		let rx_01 =
-			hub.subscribe(SubsKey::new().with_panic(SubsKeyPanic::OnUnsubscribePanicBefore));
+		let rx_01 = hub
+			.subscribe(SubsKey::new().with_panic(SubsKeyPanic::OnUnsubscribePanicBefore), 100_000);
 
 		assert_hub_props(&hub, 1, 1);
 		add_some_subscribers_see_that_messages_are_delivered_and_unsubscribe(&hub).await;
@@ -189,7 +187,8 @@ fn t06() {
 		add_some_subscribers_see_that_messages_are_delivered_and_unsubscribe(&hub).await;
 		assert_hub_props(&hub, 0, 0);
 
-		let rx_01 = hub.subscribe(SubsKey::new().with_panic(SubsKeyPanic::OnUnsubscribePanicAfter));
+		let rx_01 = hub
+			.subscribe(SubsKey::new().with_panic(SubsKeyPanic::OnUnsubscribePanicAfter), 100_000);
 
 		assert_hub_props(&hub, 1, 1);
 		add_some_subscribers_see_that_messages_are_delivered_and_unsubscribe(&hub).await;
@@ -214,7 +213,8 @@ fn t07() {
 		add_some_subscribers_see_that_messages_are_delivered_and_unsubscribe(&hub).await;
 		assert_hub_props(&hub, 0, 0);
 
-		let rx_01 = hub.subscribe(SubsKey::new().with_panic(SubsKeyPanic::OnDispatchPanicBefore));
+		let rx_01 =
+			hub.subscribe(SubsKey::new().with_panic(SubsKeyPanic::OnDispatchPanicBefore), 100_000);
 		assert_hub_props(&hub, 1, 1);
 		assert!(catch_unwind(AssertUnwindSafe(|| hub.send(1))).is_err());
 		assert_hub_props(&hub, 1, 1);
@@ -235,7 +235,8 @@ fn t08() {
 		add_some_subscribers_see_that_messages_are_delivered_and_unsubscribe(&hub).await;
 		assert_hub_props(&hub, 0, 0);
 
-		let rx_01 = hub.subscribe(SubsKey::new().with_panic(SubsKeyPanic::OnDispatchPanicAfter));
+		let rx_01 =
+			hub.subscribe(SubsKey::new().with_panic(SubsKeyPanic::OnDispatchPanicAfter), 100_000);
 		assert_hub_props(&hub, 1, 1);
 		assert!(catch_unwind(AssertUnwindSafe(|| hub.send(1))).is_err());
 		assert_hub_props(&hub, 1, 1);

substrate/client/utils/src/status_sinks.rs

@@ -58,7 +58,7 @@ impl<T> Default for StatusSinks<T> {
 impl<T> StatusSinks<T> {
 	/// Builds a new empty collection.
 	pub fn new() -> StatusSinks<T> {
-		let (entries_tx, entries_rx) = tracing_unbounded("status-sinks-entries");
+		let (entries_tx, entries_rx) = tracing_unbounded("status-sinks-entries", 100_000);
 
 		StatusSinks {
 			inner: Mutex::new(Inner { entries: stream::FuturesUnordered::new(), entries_rx }),
@@ -196,7 +196,7 @@ mod tests {
 
 		let status_sinks = StatusSinks::new();
 
-		let (tx, rx) = tracing_unbounded("test");
+		let (tx, rx) = tracing_unbounded("test", 100_000);
 		status_sinks.push(Duration::from_millis(100), tx);
 
 		let mut val_order = 5;