Separate metrics for messages sent & received (#2721)

* metered channel - sent & received

* Add for readouts

* metrics for both sent & received

* retract on send failure

Cargo.lock

@@ -3879,6 +3879,7 @@ name = "metered-channel"
 version = "0.1.0"
 dependencies = [
  "assert_matches",
+ "derive_more",
  "futures 0.3.13",
  "futures-timer 3.0.2",
 ]

node/metered-channel/Cargo.toml

@@ -8,6 +8,7 @@ description = "Channels with attached Meters"
 [dependencies]
 futures = "0.3.12"
 futures-timer = "3.0.2"
+derive_more = "0.99"
 
 [dev-dependencies]
 assert_matches = "1.4.0"

node/metered-channel/src/bounded.rs

@@ -54,8 +54,7 @@ impl<T> Stream for MeteredReceiver<T> {
 	fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
 		match mpsc::Receiver::poll_next(Pin::new(&mut self.inner), cx) {
 			Poll::Ready(x) => {
-				// always use Ordering::SeqCst to avoid underflows
-				self.meter.fill.fetch_sub(1, Ordering::SeqCst);
+				self.meter.note_received();
 				Poll::Ready(x)
 			}
 			other => other,
@@ -78,7 +77,7 @@ impl<T> MeteredReceiver<T> {
 	pub fn try_next(&mut self) -> Result<Option<T>, mpsc::TryRecvError> {
 		match self.inner.try_next()? {
 			Some(x) => {
-				self.meter.fill.fetch_sub(1, Ordering::SeqCst);
+				self.meter.note_received();
 				Ok(Some(x))
 			}
 			None => Ok(None),
@@ -131,17 +130,22 @@ impl<T> MeteredSender<T> {
 	where
 		Self: Unpin,
 	{
-		self.meter.fill.fetch_add(1, Ordering::SeqCst);
+		self.meter.note_sent();
 		let fut = self.inner.send(item);
 		futures::pin_mut!(fut);
-		fut.await
+		fut.await.map_err(|e| {
+			self.meter.retract_sent();
+			e
+		})
 	}
 
 	/// Attempt to send message or fail immediately.
 	pub fn try_send(&mut self, msg: T) -> result::Result<(), mpsc::TrySendError<T>> {
-		self.inner.try_send(msg)?;
-		self.meter.fill.fetch_add(1, Ordering::SeqCst);
-		Ok(())
+		self.meter.note_sent();
+		self.inner.try_send(msg).map_err(|e| {
+			self.meter.retract_sent();
+			e
+		})
 	}
 }
 
@@ -159,7 +163,6 @@ impl<T> futures::sink::Sink<T> for MeteredSender<T> {
     fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         match Pin::new(&mut self.inner).poll_close(cx) {
             val @ Poll::Ready(_)=> {
-                self.meter.fill.store(0, Ordering::SeqCst);
                 val
             }
             other => other,
@@ -169,7 +172,7 @@ impl<T> futures::sink::Sink<T> for MeteredSender<T> {
     fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         match Pin::new(&mut self.inner).poll_flush(cx) {
             val @ Poll::Ready(_)=> {
-                self.meter.fill.fetch_add(1, Ordering::SeqCst);
+                self.meter.note_sent();
                 val
             }
             other => other,

node/metered-channel/src/lib.rs

@@ -24,6 +24,8 @@ use std::result;
 use std::sync::Arc;
 use std::pin::Pin;
 
+use derive_more::{Add, Display};
+
 mod bounded;
 mod unbounded;
 
@@ -35,22 +37,50 @@ pub use self::unbounded::*;
 pub struct Meter {
 	/// Name of the receiver and sender pair.
 	name: &'static str,
-	// fill state of the channel
-	fill: Arc<AtomicUsize>,
+	// Number of sends on this channel.
+	sent: Arc<AtomicUsize>,
+	// Number of receives on this channel.
+	received: Arc<AtomicUsize>,
+}
+
+/// A readout of sizes from the meter. Note that it is possible, due to asynchrony, for received
+/// to be slightly higher than sent.
+#[derive(Debug, Add, Display, Clone, Default, PartialEq)]
+#[display(fmt = "(sent={} received={})", sent, received)]
+pub struct Readout {
+	/// The amount of messages sent on the channel, in aggregate.
+	pub sent: usize,
+	/// The amount of messages received on the channel, in aggregate.
+	pub received: usize,
 }
 
 impl Meter {
 	/// Count the number of items queued up inside the channel.
-	pub fn queue_count(&self) -> usize {
+	pub fn read(&self) -> Readout {
 		// when obtaining we don't care much about off by one
 		// accuracy
-		self.fill.load(Ordering::Relaxed)
+		Readout {
+			sent: self.sent.load(Ordering::Relaxed),
+			received: self.received.load(Ordering::Relaxed),
+		}
 	}
 
 	/// Obtain the name of the channel `Sender` and `Receiver` pair.
 	pub fn name(&self) -> &'static str {
 		self.name
 	}
+
+	fn note_sent(&self) {
+		self.sent.fetch_add(1, Ordering::Relaxed);
+	}
+
+	fn retract_sent(&self) {
+		self.sent.fetch_sub(1, Ordering::Relaxed);
+	}
+
+	fn note_received(&self) {
+		self.received.fetch_add(1, Ordering::Relaxed);
+	}
 }
 
 #[cfg(test)]
@@ -69,20 +99,20 @@ mod tests {
 		block_on(async move {
 			let (mut tx, mut rx) = channel::<Msg>(5, "goofy");
 			let msg = Msg::default();
-			assert_eq!(rx.meter().queue_count(), 0);
+			assert_eq!(rx.meter().read(), Readout { sent: 0, received: 0 });
 			tx.try_send(msg).unwrap();
-			assert_eq!(tx.meter().queue_count(), 1);
+			assert_eq!(tx.meter().read(), Readout { sent: 1, received: 0 });
 			tx.try_send(msg).unwrap();
 			tx.try_send(msg).unwrap();
 			tx.try_send(msg).unwrap();
-			assert_eq!(tx.meter().queue_count(), 4);
+			assert_eq!(tx.meter().read(), Readout { sent: 4, received: 0 });
 			rx.try_next().unwrap();
-			assert_eq!(rx.meter().queue_count(), 3);
+			assert_eq!(rx.meter().read(), Readout { sent: 4, received: 1 });
 			rx.try_next().unwrap();
 			rx.try_next().unwrap();
-			assert_eq!(tx.meter().queue_count(), 1);
+			assert_eq!(tx.meter().read(), Readout { sent: 4, received: 3 });
 			rx.try_next().unwrap();
-			assert_eq!(rx.meter().queue_count(), 0);
+			assert_eq!(rx.meter().read(), Readout { sent: 4, received: 4 });
 			assert!(rx.try_next().is_err());
 		});
 	}
@@ -96,9 +126,9 @@ mod tests {
 			futures::join!(
 				async move {
 					let msg = Msg::default();
-					assert_eq!(tx.meter().queue_count(), 0);
+					assert_eq!(tx.meter().read(), Readout { sent: 0, received: 0 });
 					tx.try_send(msg).unwrap();
-					assert_eq!(tx.meter().queue_count(), 1);
+					assert_eq!(tx.meter().read(), Readout { sent: 1, received: 0 });
 					tx.try_send(msg).unwrap();
 					tx.try_send(msg).unwrap();
 					tx.try_send(msg).unwrap();
@@ -106,14 +136,14 @@ mod tests {
 				},
 				async move {
 					go.await.expect("Helper oneshot channel must work. qed");
-					assert_eq!(rx.meter().queue_count(), 4);
+					assert_eq!(rx.meter().read(), Readout { sent: 4, received: 0 });
 					rx.try_next().unwrap();
-					assert_eq!(rx.meter().queue_count(), 3);
+					assert_eq!(rx.meter().read(), Readout { sent: 4, received: 1 });
 					rx.try_next().unwrap();
 					rx.try_next().unwrap();
-					assert_eq!(rx.meter().queue_count(), 1);
+					assert_eq!(rx.meter().read(), Readout { sent: 4, received: 3 });
 					rx.try_next().unwrap();
-					assert_eq!(dbg!(rx.meter().queue_count()), 0);
+					assert_eq!(dbg!(rx.meter().read()), Readout { sent: 4, received: 4 });
 				}
 			)
 		});
@@ -130,21 +160,37 @@ mod tests {
 			futures::join!(
 				async move {
 					for i in 0..15 {
-						println!("Sent #{} with a backlog of {} items", i + 1, tx.meter().queue_count());
+						println!("Sent #{} with a backlog of {} items", i + 1, tx.meter().read());
 						let msg = Msg { val: i as u8 + 1u8 };
 						tx.send(msg).await.unwrap();
-						assert!(tx.meter().queue_count() > 0usize);
+						assert!(tx.meter().read().sent > 0usize);
 						Delay::new(Duration::from_millis(20)).await;
 					}
 					()
 				},
 				async move {
 					while let Some(msg) = rx.next().await {
-						println!("rx'd one {} with {} backlogged", msg.val, rx.meter().queue_count());
+						println!("rx'd one {} with {} backlogged", msg.val, rx.meter().read());
 						Delay::new(Duration::from_millis(29)).await;
 					}
 				}
 			)
 		});
 	}
+
+	#[test]
+	fn failed_send_does_not_inc_sent() {
+		let (mut bounded, _) = channel::<Msg>(5, "pluto");
+		let (mut unbounded, _) = unbounded::<Msg>("pluto");
+
+		block_on(async move {
+			assert!(bounded.send(Msg::default()).await.is_err());
+			assert!(bounded.try_send(Msg::default()).is_err());
+			assert_eq!(bounded.meter().read(), Readout { sent: 0, received: 0 });
+
+			assert!(unbounded.send(Msg::default()).await.is_err());
+			assert!(unbounded.unbounded_send(Msg::default()).is_err());
+			assert_eq!(unbounded.meter().read(), Readout { sent: 0, received: 0 });
+		});
+	}
 }

node/metered-channel/src/unbounded.rs

@@ -54,8 +54,7 @@ impl<T> Stream for UnboundedMeteredReceiver<T> {
     fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         match mpsc::UnboundedReceiver::poll_next(Pin::new(&mut self.inner), cx) {
             Poll::Ready(x) => {
-                // always use Ordering::SeqCst to avoid underflows
-                self.meter.fill.fetch_sub(1, Ordering::SeqCst);
+                self.meter.note_received();
                 Poll::Ready(x)
             }
             other => other,
@@ -78,7 +77,7 @@ impl<T> UnboundedMeteredReceiver<T> {
     pub fn try_next(&mut self) -> Result<Option<T>, mpsc::TryRecvError> {
         match self.inner.try_next()? {
             Some(x) => {
-                self.meter.fill.fetch_sub(1, Ordering::SeqCst);
+                self.meter.note_received();
                 Ok(Some(x))
             }
             None => Ok(None),
@@ -131,18 +130,23 @@ impl<T> UnboundedMeteredSender<T> {
     where
         Self: Unpin,
     {
-        self.meter.fill.fetch_add(1, Ordering::SeqCst);
+        self.meter.note_sent();
         let fut = self.inner.send(item);
         futures::pin_mut!(fut);
-        fut.await
+        fut.await.map_err(|e| {
+			self.meter.retract_sent();
+			e
+		})
     }
 
 
     /// Attempt to send message or fail immediately.
     pub fn unbounded_send(&mut self, msg: T) -> result::Result<(), mpsc::TrySendError<T>> {
-        self.inner.unbounded_send(msg)?;
-        self.meter.fill.fetch_add(1, Ordering::SeqCst);
-        Ok(())
+        self.meter.note_sent();
+        self.inner.unbounded_send(msg).map_err(|e| {
+			self.meter.retract_sent();
+			e
+		})
     }
 }
 
@@ -160,7 +164,6 @@ impl<T> futures::sink::Sink<T> for UnboundedMeteredSender<T> {
     fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         match Pin::new(&mut self.inner).poll_ready(cx) {
             val @ Poll::Ready(_)=> {
-                self.meter.fill.store(0, Ordering::SeqCst);
                 val
             }
             other => other,
@@ -170,7 +173,7 @@ impl<T> futures::sink::Sink<T> for UnboundedMeteredSender<T> {
     fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         match Pin::new(&mut self.inner).poll_ready(cx) {
             val @ Poll::Ready(_)=> {
-                self.meter.fill.fetch_add(1, Ordering::SeqCst);
+                self.meter.note_sent();
                 val
             }
             other => other,

node/overseer/src/lib.rs

@@ -1327,8 +1327,10 @@ struct MetricsInner {
 	deactivated_heads_total: prometheus::Counter<prometheus::U64>,
 	messages_relayed_total: prometheus::Counter<prometheus::U64>,
 	message_relay_timings: prometheus::Histogram,
-	to_overseer_channel_queue_size: prometheus::Gauge<prometheus::U64>,
-	from_overseer_channel_queue_size: prometheus::GaugeVec<prometheus::U64>,
+	to_overseer_sent: prometheus::Gauge<prometheus::U64>,
+	to_overseer_received: prometheus::Gauge<prometheus::U64>,
+	from_overseer_sent: prometheus::GaugeVec<prometheus::U64>,
+	from_overseer_received: prometheus::GaugeVec<prometheus::U64>,
 }
 
 #[derive(Default, Clone)]
@@ -1360,15 +1362,21 @@ impl Metrics {
 
 	fn channel_fill_level_snapshot(
 		&self,
-		from_overseer: AllSubsystemsSame<(&'static str, usize)>,
-		to_overseer: usize,
+		from_overseer: AllSubsystemsSame<(&'static str, metered::Readout)>,
+		to_overseer: metered::Readout,
 	) {
 		self.0.as_ref().map(|metrics| {
-			from_overseer.map_subsystems(|(name, queue_size): (_, usize)| {
-				metrics.from_overseer_channel_queue_size.with_label_values(&[name]).set(queue_size as u64);
-			})
+			from_overseer.map_subsystems(|(name, readout): (_, metered::Readout)| {
+				metrics.from_overseer_sent.with_label_values(&[name])
+					.set(readout.sent as u64);
+
+				metrics.from_overseer_received.with_label_values(&[name])
+					.set(readout.received as u64);
+			});
+
+			metrics.to_overseer_sent.set(to_overseer.sent as u64);
+			metrics.to_overseer_received.set(to_overseer.received as u64);
 		});
-		self.0.as_ref().map(|metrics| metrics.to_overseer_channel_queue_size.set(to_overseer as u64));
 	}
 }
 
@@ -1418,11 +1426,11 @@ impl metrics::Metrics for Metrics {
 				)?,
 				registry,
 			)?,
-			from_overseer_channel_queue_size: prometheus::register(
+			from_overseer_sent: prometheus::register(
 				prometheus::GaugeVec::<prometheus::U64>::new(
 					prometheus::Opts::new(
-						"parachain_from_overseer_channel_queue_size",
-						"Number of elements sitting in the channel from the overseer waiting to be processed.",
+						"parachain_from_overseer_sent",
+						"Number of elements sent by the overseer to subsystems",
 					),
 					&[
 						"subsystem_name",
@@ -1430,11 +1438,32 @@ impl metrics::Metrics for Metrics {
 				)?,
 				registry,
 			)?,
-			to_overseer_channel_queue_size: prometheus::register(
+			from_overseer_received: prometheus::register(
+				prometheus::GaugeVec::<prometheus::U64>::new(
+					prometheus::Opts::new(
+						"parachain_from_overseer_received",
+						"Number of elements received by subsystems from overseer",
+					),
+					&[
+						"subsystem_name",
+					],
+				)?,
+				registry,
+			)?,
+			to_overseer_sent: prometheus::register(
+				prometheus::Gauge::<prometheus::U64>::with_opts(
+					prometheus::Opts::new(
+						"parachain_to_overseer_sent",
+						"Number of elements sent by subsystems to overseer",
+					),
+				)?,
+				registry,
+			)?,
+			to_overseer_received: prometheus::register(
 				prometheus::Gauge::<prometheus::U64>::with_opts(
 					prometheus::Opts::new(
-						"parachain_to_overseer_channel_queue_size",
-						"Number of elements sitting in the channel to the overseer waiting to be processed.",
+						"parachain_to_overseer_received",
+						"Number of element received by overseer from subsystems",
 					),
 				)?,
 				registry,
@@ -1820,14 +1849,14 @@ where
 			let metronome = Metronome::new(std::time::Duration::from_millis(950))
 				.for_each(move |_| {
 					let to_subsystem_counts = subsystem_meters.as_ref()
-						.map_subsystems(|&(name, ref meter): &(_, metered::Meter)| (name, meter.queue_count()));
+						.map_subsystems(|&(name, ref meter): &(_, metered::Meter)| (name, meter.read()));
 
 					// We combine the amount of messages from subsystems to the overseer
 					// as well as the amount of messages from external sources to the overseer
 					// into one to_overseer value.
 					metronome_metrics.channel_fill_level_snapshot(
 						to_subsystem_counts,
-						meter_subsystem_to_overseer.queue_count() + meter_external_to_overseer.queue_count(),
+						meter_subsystem_to_overseer.read() + meter_external_to_overseer.read(),
 					);
 
 					async move {
@@ -2467,7 +2496,8 @@ mod tests {
 		assert_eq!(gather[1].get_name(), "parachain_deactivated_heads_total");
 		assert_eq!(gather[2].get_name(), "parachain_messages_relayed_total");
 		assert_eq!(gather[3].get_name(), "parachain_overseer_messages_relay_timings");
-		assert_eq!(gather[4].get_name(), "parachain_to_overseer_channel_queue_size");
+		assert_eq!(gather[4].get_name(), "parachain_to_overseer_received");
+		assert_eq!(gather[5].get_name(), "parachain_to_overseer_sent");
 		let activated = gather[0].get_metric()[0].get_counter().get_value() as u64;
 		let deactivated = gather[1].get_metric()[0].get_counter().get_value() as u64;
 		let relayed = gather[2].get_metric()[0].get_counter().get_value() as u64;