From e7f2d46cd473270de40848e0d331a5ccbcb4771a Mon Sep 17 00:00:00 2001 From: boqiu <82121246@qq.com> Date: Wed, 21 Aug 2024 19:03:50 +0800 Subject: [PATCH] refactor channel metrics --- common/channel/src/channel.rs | 107 ++++++-------------------------- common/channel/src/lib.rs | 1 + common/channel/src/metrics.rs | 112 ++++++++++++++++++++++++++++++++++ 3 files changed, 132 insertions(+), 88 deletions(-) create mode 100644 common/channel/src/metrics.rs diff --git a/common/channel/src/channel.rs b/common/channel/src/channel.rs index d70aa4d..1c70df9 100644 --- a/common/channel/src/channel.rs +++ b/common/channel/src/channel.rs @@ -1,9 +1,10 @@ use crate::error::Error; -use metrics::{register_meter_with_group, Counter, CounterUsize, Histogram, Meter, Sample}; +use crate::metrics::unbounded_channel; +use metrics::{Counter, CounterUsize}; use std::sync::Arc; -use std::time::{Duration, Instant}; -use tokio::sync::mpsc::error::{SendError, TryRecvError}; -use tokio::sync::{mpsc, oneshot}; +use std::time::Duration; +use tokio::sync::mpsc::error::TryRecvError; +use tokio::sync::oneshot; use tokio::time::timeout; const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(3); @@ -22,63 +23,21 @@ pub struct Channel { impl Channel { pub fn unbounded(name: &str) -> (Sender, Receiver) { - let (sender, receiver) = mpsc::unbounded_channel(); - let metrics_group = format!("common_channel_{}", name); - let metrics_queued = CounterUsize::register_with_group(metrics_group.as_str(), "size"); - + let (sender, receiver) = unbounded_channel(metrics_group.as_str()); + let metrics_timeout = CounterUsize::register_with_group(metrics_group.as_str(), "timeout"); ( 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, - ), + metrics_timeout, }, + Receiver { chan: receiver }, ) } } -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>, - - metrics_send_qps: Arc, - metrics_queued: Arc>, + chan: crate::metrics::Sender>, metrics_timeout: Arc>, } @@ -86,8 +45,6 @@ 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(), } } @@ -95,13 +52,17 @@ impl Clone for Sender { impl Sender { pub fn notify(&self, msg: N) -> Result<(), Error> { - self.send(Message::Notification(msg)) + self.chan + .send(Message::Notification(msg)) + .map_err(|e| Error::SendError(e)) } pub async fn request(&self, request: Req) -> Result> { let (sender, receiver) = oneshot::channel(); - self.send(Message::Request(request, sender))?; + self.chan + .send(Message::Request(request, sender)) + .map_err(|e| Error::SendError(e))?; timeout(DEFAULT_REQUEST_TIMEOUT, receiver) .await @@ -111,49 +72,19 @@ impl Sender { })? .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>, - - metrics_recv_qps: Arc, - metrics_queued: Arc>, - metrics_queue_latency: Arc, + chan: crate::metrics::Receiver>, } impl Receiver { pub async fn recv(&mut self) -> Option> { - let data = self.chan.recv().await?; - Some(self.on_recv_data(data)) + self.chan.recv().await } pub fn try_recv(&mut self) -> Result, TryRecvError> { - 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 + self.chan.try_recv() } } diff --git a/common/channel/src/lib.rs b/common/channel/src/lib.rs index f78f284..7978360 100644 --- a/common/channel/src/lib.rs +++ b/common/channel/src/lib.rs @@ -1,5 +1,6 @@ mod channel; pub mod error; +pub mod metrics; pub mod test_util; pub use crate::channel::{Channel, Message, Receiver, ResponseSender, Sender}; diff --git a/common/channel/src/metrics.rs b/common/channel/src/metrics.rs new file mode 100644 index 0000000..5603b54 --- /dev/null +++ b/common/channel/src/metrics.rs @@ -0,0 +1,112 @@ +use std::{fmt::Debug, sync::Arc, time::Instant}; + +use metrics::{register_meter_with_group, Counter, CounterUsize, Histogram, Meter, Sample}; +use tokio::sync::mpsc::{ + error::{SendError, TryRecvError}, + unbounded_channel as new_unbounded_channel, UnboundedReceiver, UnboundedSender, +}; + +pub fn unbounded_channel(metric_name: &str) -> (Sender, Receiver) { + let (sender, receiver) = new_unbounded_channel(); + let metrics_queued = CounterUsize::register_with_group(metric_name, "size"); + ( + Sender::new(sender, metric_name, metrics_queued.clone()), + Receiver::new(receiver, metric_name, metrics_queued), + ) +} + +pub struct Sender { + sender: UnboundedSender<(Instant, T)>, + metrics_send_qps: Arc, + metrics_queued: Arc>, +} + +impl Clone for Sender { + fn clone(&self) -> Self { + Self { + sender: self.sender.clone(), + metrics_send_qps: self.metrics_send_qps.clone(), + metrics_queued: self.metrics_queued.clone(), + } + } +} + +impl Debug for Sender { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{:?}", self.sender) + } +} + +impl Sender { + pub(crate) fn new( + sender: UnboundedSender<(Instant, T)>, + metrics_group: &str, + metrics_queued: Arc>, + ) -> Self { + Self { + sender, + metrics_send_qps: register_meter_with_group(metrics_group, "send"), + metrics_queued, + } + } + + pub fn send(&self, value: T) -> Result<(), SendError> { + match self.sender.send((Instant::now(), value)) { + Ok(()) => { + self.metrics_send_qps.mark(1); + self.metrics_queued.inc(1); + Ok(()) + } + Err(e) => Err(SendError(e.0 .1)), + } + } +} + +pub struct Receiver { + receiver: UnboundedReceiver<(Instant, T)>, + metrics_recv_qps: Arc, + metrics_queued: Arc>, + metrics_queue_latency: Arc, +} + +impl Debug for Receiver { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{:?}", self.receiver) + } +} + +impl Receiver { + pub(crate) fn new( + receiver: UnboundedReceiver<(Instant, T)>, + metrics_group: &str, + metrics_queued: Arc>, + ) -> Self { + Self { + receiver, + metrics_recv_qps: register_meter_with_group(metrics_group, "recv"), + metrics_queued, + metrics_queue_latency: Sample::ExpDecay(0.015).register_with_group( + metrics_group, + "latency", + 1024, + ), + } + } + + fn on_recv(&self, value: (Instant, T)) -> T { + self.metrics_recv_qps.mark(1); + self.metrics_queued.dec(1); + self.metrics_queue_latency.update_since(value.0); + value.1 + } + + pub async fn recv(&mut self) -> Option { + let value = self.receiver.recv().await?; + Some(self.on_recv(value)) + } + + pub fn try_recv(&mut self) -> Result { + let value = self.receiver.try_recv()?; + Ok(self.on_recv(value)) + } +}