diff --git a/Cargo.lock b/Cargo.lock index 3a85a3f..08cd52a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -844,6 +844,7 @@ dependencies = [ name = "channel" version = "0.1.0" dependencies = [ + "metrics", "tokio", ] diff --git a/common/channel/Cargo.toml b/common/channel/Cargo.toml index cbdc369..d49d24f 100644 --- a/common/channel/Cargo.toml +++ b/common/channel/Cargo.toml @@ -5,3 +5,4 @@ edition = "2021" [dependencies] tokio = { version = "1.19.2", features = ["sync", "time"] } +metrics = { workspace = true } diff --git a/common/channel/src/channel.rs b/common/channel/src/channel.rs index 2f5206a..d70aa4d 100644 --- a/common/channel/src/channel.rs +++ b/common/channel/src/channel.rs @@ -1,6 +1,8 @@ use crate::error::Error; -use std::time::Duration; -use tokio::sync::mpsc::error::TryRecvError; +use metrics::{register_meter_with_group, Counter, CounterUsize, Histogram, Meter, Sample}; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use tokio::sync::mpsc::error::{SendError, TryRecvError}; use tokio::sync::{mpsc, oneshot}; use tokio::time::timeout; @@ -19,56 +21,139 @@ pub struct Channel { } impl Channel { - pub fn unbounded() -> (Sender, Receiver) { + pub fn unbounded(name: &str) -> (Sender, Receiver) { let (sender, receiver) = mpsc::unbounded_channel(); - (Sender { chan: sender }, Receiver { chan: receiver }) + + let metrics_group = format!("common_channel_{}", name); + let metrics_queued = CounterUsize::register_with_group(metrics_group.as_str(), "size"); + + ( + Sender { + chan: sender, + metrics_send_qps: register_meter_with_group(metrics_group.as_str(), "send"), + metrics_queued: metrics_queued.clone(), + metrics_timeout: CounterUsize::register_with_group( + metrics_group.as_str(), + "timeout", + ), + }, + Receiver { + chan: receiver, + metrics_recv_qps: register_meter_with_group(metrics_group.as_str(), "recv"), + metrics_queued, + metrics_queue_latency: Sample::ExpDecay(0.015).register_with_group( + metrics_group.as_str(), + "latency", + 1024, + ), + }, + ) + } +} + +enum TimedMessage { + Notification(Instant, N), + Request(Instant, Req, ResponseSender), +} + +impl From> for TimedMessage { + fn from(value: Message) -> Self { + match value { + Message::Notification(n) => TimedMessage::Notification(Instant::now(), n), + Message::Request(req, res) => TimedMessage::Request(Instant::now(), req, res), + } + } +} + +impl TimedMessage { + fn into_message(self) -> (Instant, Message) { + match self { + TimedMessage::Notification(since, n) => (since, Message::Notification(n)), + TimedMessage::Request(since, req, res) => (since, Message::Request(req, res)), + } } } pub struct Sender { - chan: mpsc::UnboundedSender>, + chan: mpsc::UnboundedSender>, + + metrics_send_qps: Arc, + metrics_queued: Arc>, + metrics_timeout: Arc>, } impl Clone for Sender { fn clone(&self) -> Self { Sender { chan: self.chan.clone(), + metrics_send_qps: self.metrics_send_qps.clone(), + metrics_queued: self.metrics_queued.clone(), + metrics_timeout: self.metrics_timeout.clone(), } } } impl Sender { pub fn notify(&self, msg: N) -> Result<(), Error> { - self.chan - .send(Message::Notification(msg)) - .map_err(|e| Error::SendError(e)) + self.send(Message::Notification(msg)) } pub async fn request(&self, request: Req) -> Result> { let (sender, receiver) = oneshot::channel(); - self.chan - .send(Message::Request(request, sender)) - .map_err(|e| Error::SendError(e))?; + self.send(Message::Request(request, sender))?; timeout(DEFAULT_REQUEST_TIMEOUT, receiver) .await - .map_err(|_| Error::TimeoutError)? + .map_err(|_| { + self.metrics_timeout.inc(1); + Error::TimeoutError + })? .map_err(|e| Error::RecvError(e)) } + + fn send(&self, message: Message) -> Result<(), Error> { + match self.chan.send(message.into()) { + Ok(()) => { + self.metrics_send_qps.mark(1); + self.metrics_queued.inc(1); + Ok(()) + } + Err(e) => { + let (_, msg) = e.0.into_message(); + Err(Error::SendError(SendError(msg))) + } + } + } } pub struct Receiver { - chan: mpsc::UnboundedReceiver>, + chan: mpsc::UnboundedReceiver>, + + metrics_recv_qps: Arc, + metrics_queued: Arc>, + metrics_queue_latency: Arc, } impl Receiver { pub async fn recv(&mut self) -> Option> { - self.chan.recv().await + let data = self.chan.recv().await?; + Some(self.on_recv_data(data)) } pub fn try_recv(&mut self) -> Result, TryRecvError> { - self.chan.try_recv() + let data = self.chan.try_recv()?; + Ok(self.on_recv_data(data)) + } + + fn on_recv_data(&self, data: TimedMessage) -> Message { + self.metrics_recv_qps.mark(1); + self.metrics_queued.dec(1); + + let (since, msg) = data.into_message(); + self.metrics_queue_latency.update_since(since); + + msg } } @@ -91,7 +176,7 @@ mod tests { #[tokio::test] async fn request_response() { - let (tx, mut rx) = Channel::::unbounded(); + let (tx, mut rx) = Channel::::unbounded("test"); let task1 = async move { match rx.recv().await.expect("not dropped") { diff --git a/node/router/src/libp2p_event_handler.rs b/node/router/src/libp2p_event_handler.rs index 19ca59a..b3bb839 100644 --- a/node/router/src/libp2p_event_handler.rs +++ b/node/router/src/libp2p_event_handler.rs @@ -815,7 +815,7 @@ mod tests { let runtime = TestRuntime::default(); let (network_globals, keypair) = Context::new_network_globals(); let (network_send, network_recv) = mpsc::unbounded_channel(); - let (sync_send, sync_recv) = channel::Channel::unbounded(); + let (sync_send, sync_recv) = channel::Channel::unbounded("test"); let (chunk_pool_send, _chunk_pool_recv) = mpsc::unbounded_channel(); let store = LogManager::memorydb(LogConfig::default()).unwrap(); Self { diff --git a/node/sync/src/service.rs b/node/sync/src/service.rs index a01d2d8..9282d99 100644 --- a/node/sync/src/service.rs +++ b/node/sync/src/service.rs @@ -157,7 +157,7 @@ impl SyncService { event_recv: broadcast::Receiver, catch_up_end_recv: oneshot::Receiver<()>, ) -> Result { - let (sync_send, sync_recv) = channel::Channel::unbounded(); + let (sync_send, sync_recv) = channel::Channel::unbounded("sync"); let store = Store::new(store, executor.clone()); // init auto sync @@ -912,7 +912,7 @@ mod tests { create_file_location_cache(init_peer_id, vec![txs[0].id()]); let (network_send, mut network_recv) = mpsc::unbounded_channel::(); - let (_, sync_recv) = channel::Channel::unbounded(); + let (_, sync_recv) = channel::Channel::unbounded("test"); let mut sync = SyncService { config: Config::default(), @@ -941,7 +941,7 @@ mod tests { create_file_location_cache(init_peer_id, vec![txs[0].id()]); let (network_send, mut network_recv) = mpsc::unbounded_channel::(); - let (_, sync_recv) = channel::Channel::unbounded(); + let (_, sync_recv) = channel::Channel::unbounded("test"); let mut sync = SyncService { config: Config::default(),