sp-utils => sc-utils (#9677)

* sp-utils => sc-utils

* cargo fmt

* These files are now in the client so should be licensed as GPL3

* Apply suggestions from code review

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

substrate/client/utils/src/lib.rs

@@ -0,0 +1,41 @@
+// This file is part of Substrate.
+
+// Copyright (C) 2020-2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+//! Utilities Primitives for Substrate
+//!
+//! # Features
+//!
+//! ## 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:
+//!
+//! | Label        | Description                                   |
+//! | ------------ | --------------------------------------------- |
+//! | entity       | Name of channel passed to `tracing_unbounded` |
+//! | action       | One of `send`/`received`/`dropped`            |
+
+pub mod metrics;
+pub mod mpsc;
+pub mod status_sinks;

substrate/client/utils/src/metrics.rs

@@ -0,0 +1,58 @@
+// This file is part of Substrate.
+
+// Copyright (C) 2020-2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+
+//! Metering primitives and globals
+
+use lazy_static::lazy_static;
+use prometheus::{
+	core::{AtomicU64, GenericCounter, GenericGauge},
+	Error as PrometheusError, Registry,
+};
+
+#[cfg(feature = "metered")]
+use prometheus::{core::GenericCounterVec, Opts};
+
+lazy_static! {
+	pub static ref TOKIO_THREADS_TOTAL: GenericCounter<AtomicU64> =
+		GenericCounter::new("tokio_threads_total", "Total number of threads created")
+			.expect("Creating of statics doesn't fail. qed");
+	pub static ref TOKIO_THREADS_ALIVE: GenericGauge<AtomicU64> =
+		GenericGauge::new("tokio_threads_alive", "Number of threads alive right now")
+			.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("unbounded_channel_len", "Items in each mpsc::unbounded instance"),
+		&["entity", "action"] // 'name of channel, send|received|dropped
+	).expect("Creating of statics doesn't fail. qed");
+
+}
+
+/// Register the statics to report to registry
+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

@@ -0,0 +1,232 @@
+// This file is part of Substrate.
+
+// Copyright (C) 2020-2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+
+// 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
+
+#[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>;
+
+	/// Alias `mpsc::unbounded`
+	pub fn tracing_unbounded<T>(
+		_key: &'static str,
+	) -> (TracingUnboundedSender<T>, TracingUnboundedReceiver<T>) {
+		mpsc::unbounded()
+	}
+}
+
+#[cfg(feature = "metered")]
+mod inner {
+	// tracing implementation
+	use crate::metrics::UNBOUNDED_CHANNELS_COUNTER;
+	use futures::{
+		channel::mpsc::{
+			self, SendError, TryRecvError, TrySendError, UnboundedReceiver, UnboundedSender,
+		},
+		sink::Sink,
+		stream::{FusedStream, Stream},
+		task::{Context, Poll},
+	};
+	use std::pin::Pin;
+
+	/// Wrapper Type around `UnboundedSender` that increases the global
+	/// measure when a message is added
+	#[derive(Debug)]
+	pub struct TracingUnboundedSender<T>(&'static str, UnboundedSender<T>);
+
+	// 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())
+		}
+	}
+
+	/// Wrapper Type around `UnboundedReceiver` that decreases the global
+	/// measure when a message is polled
+	#[derive(Debug)]
+	pub struct TracingUnboundedReceiver<T>(&'static str, UnboundedReceiver<T>);
+
+	/// Wrapper around `mpsc::unbounded` that tracks the in- and outflow via
+	/// `UNBOUNDED_CHANNELS_COUNTER`
+	pub fn tracing_unbounded<T>(
+		key: &'static str,
+	) -> (TracingUnboundedSender<T>, TracingUnboundedReceiver<T>) {
+		let (s, r) = mpsc::unbounded();
+		(TracingUnboundedSender(key, s), TracingUnboundedReceiver(key, r))
+	}
+
+	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)
+		}
+
+		/// Proxy function to mpsc::UnboundedSender
+		pub fn is_closed(&self) -> bool {
+			self.1.is_closed()
+		}
+
+		/// Proxy function to mpsc::UnboundedSender
+		pub fn close_channel(&self) {
+			self.1.close_channel()
+		}
+
+		/// Proxy function to mpsc::UnboundedSender
+		pub fn disconnect(&mut self) {
+			self.1.disconnect()
+		}
+
+		/// Proxy function to mpsc::UnboundedSender
+		pub fn start_send(&mut self, msg: T) -> Result<(), SendError> {
+			self.1.start_send(msg)
+		}
+
+		/// 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();
+				s
+			})
+		}
+
+		/// Proxy function to mpsc::UnboundedSender
+		pub fn same_receiver(&self, other: &UnboundedSender<T>) -> bool {
+			self.1.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.1.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.0, "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.1.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| {
+				if s.is_some() {
+					UNBOUNDED_CHANNELS_COUNTER.with_label_values(&[self.0, "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.1).poll_next(cx) {
+				Poll::Ready(msg) => {
+					if msg.is_some() {
+						UNBOUNDED_CHANNELS_COUNTER.with_label_values(&[s.0, "received"]).inc();
+					}
+					Poll::Ready(msg)
+				},
+				Poll::Pending => Poll::Pending,
+			}
+		}
+	}
+
+	impl<T> FusedStream for TracingUnboundedReceiver<T> {
+		fn is_terminated(&self) -> bool {
+			self.1.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>> {
+			self.close_channel();
+			Poll::Ready(Ok(()))
+		}
+	}
+}
+
+pub use inner::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};

substrate/client/utils/src/status_sinks.rs

@@ -0,0 +1,218 @@
+// This file is part of Substrate.
+
+// Copyright (C) 2019-2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+
+// 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 crate::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};
+use futures::{lock::Mutex, prelude::*};
+use futures_timer::Delay;
+use std::{
+	pin::Pin,
+	task::{Context, Poll},
+	time::Duration,
+};
+
+/// Holds a list of `UnboundedSender`s, each associated with a certain time period. Every time the
+/// period elapses, we push an element on the sender.
+///
+/// Senders are removed only when they are closed.
+pub struct StatusSinks<T> {
+	/// Should only be locked by `next`.
+	inner: Mutex<Inner<T>>,
+	/// Sending side of `Inner::entries_rx`.
+	entries_tx: TracingUnboundedSender<YieldAfter<T>>,
+}
+
+struct Inner<T> {
+	/// The actual entries of the list.
+	entries: stream::FuturesUnordered<YieldAfter<T>>,
+	/// Receives new entries and puts them in `entries`.
+	entries_rx: TracingUnboundedReceiver<YieldAfter<T>>,
+}
+
+struct YieldAfter<T> {
+	delay: Delay,
+	interval: Duration,
+	sender: Option<TracingUnboundedSender<T>>,
+}
+
+impl<T> Default for StatusSinks<T> {
+	fn default() -> Self {
+		Self::new()
+	}
+}
+
+impl<T> StatusSinks<T> {
+	/// Builds a new empty collection.
+	pub fn new() -> StatusSinks<T> {
+		let (entries_tx, entries_rx) = tracing_unbounded("status-sinks-entries");
+
+		StatusSinks {
+			inner: Mutex::new(Inner { entries: stream::FuturesUnordered::new(), entries_rx }),
+			entries_tx,
+		}
+	}
+
+	/// Adds a sender to the collection.
+	///
+	/// The `interval` is the time period between two pushes on the sender.
+	pub fn push(&self, interval: Duration, sender: TracingUnboundedSender<T>) {
+		let _ = self.entries_tx.unbounded_send(YieldAfter {
+			delay: Delay::new(interval),
+			interval,
+			sender: Some(sender),
+		});
+	}
+
+	/// Waits until one of the sinks is ready, then returns an object that can be used to send
+	/// an element on said sink.
+	///
+	/// If the object isn't used to send an element, the slot is skipped.
+	pub async fn next(&self) -> ReadySinkEvent<'_, T> {
+		// This is only ever locked by `next`, which means that one `next` at a time can run.
+		let mut inner = self.inner.lock().await;
+		let inner = &mut *inner;
+
+		loop {
+			// Future that produces the next ready entry in `entries`, or doesn't produce anything
+			// if the list is empty.
+			let next_ready_entry = {
+				let entries = &mut inner.entries;
+				async move {
+					if let Some(v) = entries.next().await {
+						v
+					} else {
+						loop {
+							futures::pending!()
+						}
+					}
+				}
+			};
+
+			futures::select! {
+				new_entry = inner.entries_rx.next() => {
+					if let Some(new_entry) = new_entry {
+						inner.entries.push(new_entry);
+					}
+				},
+				(sender, interval) = next_ready_entry.fuse() => {
+					return ReadySinkEvent {
+						sinks: self,
+						sender: Some(sender),
+						interval,
+					}
+				}
+			}
+		}
+	}
+}
+
+/// One of the sinks is ready.
+#[must_use]
+pub struct ReadySinkEvent<'a, T> {
+	sinks: &'a StatusSinks<T>,
+	sender: Option<TracingUnboundedSender<T>>,
+	interval: Duration,
+}
+
+impl<'a, T> ReadySinkEvent<'a, T> {
+	/// Sends an element on the sender.
+	pub fn send(mut self, element: T) {
+		if let Some(sender) = self.sender.take() {
+			if sender.unbounded_send(element).is_ok() {
+				let _ = self.sinks.entries_tx.unbounded_send(YieldAfter {
+					// Note that since there's a small delay between the moment a task is
+					// woken up and the moment it is polled, the period is actually not
+					// `interval` but `interval + <delay>`. We ignore this problem in
+					// practice.
+					delay: Delay::new(self.interval),
+					interval: self.interval,
+					sender: Some(sender),
+				});
+			}
+		}
+	}
+}
+
+impl<'a, T> Drop for ReadySinkEvent<'a, T> {
+	fn drop(&mut self) {
+		if let Some(sender) = self.sender.take() {
+			if sender.is_closed() {
+				return
+			}
+
+			let _ = self.sinks.entries_tx.unbounded_send(YieldAfter {
+				delay: Delay::new(self.interval),
+				interval: self.interval,
+				sender: Some(sender),
+			});
+		}
+	}
+}
+
+impl<T> futures::Future for YieldAfter<T> {
+	type Output = (TracingUnboundedSender<T>, Duration);
+
+	fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+		let this = Pin::into_inner(self);
+
+		match Pin::new(&mut this.delay).poll(cx) {
+			Poll::Pending => Poll::Pending,
+			Poll::Ready(()) => {
+				let sender = this
+					.sender
+					.take()
+					.expect("sender is always Some unless the future is finished; qed");
+				Poll::Ready((sender, this.interval))
+			},
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::StatusSinks;
+	use crate::mpsc::tracing_unbounded;
+	use futures::prelude::*;
+	use std::time::Duration;
+
+	#[test]
+	fn works() {
+		// We're not testing that the `StatusSink` properly enforces an order in the intervals, as
+		// this easily causes test failures on busy CPUs.
+
+		let status_sinks = StatusSinks::new();
+
+		let (tx, rx) = tracing_unbounded("test");
+		status_sinks.push(Duration::from_millis(100), tx);
+
+		let mut val_order = 5;
+
+		futures::executor::block_on(futures::future::select(
+			Box::pin(async move {
+				loop {
+					let ev = status_sinks.next().await;
+					val_order += 1;
+					ev.send(val_order);
+				}
+			}),
+			Box::pin(async {
+				let items: Vec<i32> = rx.take(3).collect().await;
+				assert_eq!(items, [6, 7, 8]);
+			}),
+		));
+	}
+}