Add performance counters gated by metrics feature

cloneable · cloneable · commit 3443a71187aa · 2023-02-12T15:57:01.000+01:00
diff --git a/Cargo.toml b/Cargo.toml
@@ -28,7 +28,7 @@ pem = "1.0.1"
 percent-encoding = "2.1.0"
 pin-project = "1.0.2"
 priority-queue = "1"
-serde = "1"
+serde = { version = "1", features = ["derive", "rc"] }
 serde_json = "1"
 socket2 = "0.4.2"
 thiserror = "1.0.4"
@@ -92,6 +92,7 @@ default-rustls = [
     "mysql_common/frunk",
     "rustls-tls",
 ]
+metrics = []
 minimal = ["flate2/zlib"]
 native-tls-tls = ["native-tls", "tokio-native-tls"]
 rustls-tls = [
diff --git a/src/buffer_pool.rs b/src/buffer_pool.rs
@@ -6,6 +6,7 @@
 // option. All files in the project carrying such notice may not be copied,
 // modified, or distributed except according to those terms.
 
+use crate::metrics::BufferPoolMetrics;
 use crossbeam::queue::ArrayQueue;
 use std::{mem::replace, ops::Deref, sync::Arc};
 
@@ -14,6 +15,7 @@ pub struct BufferPool {
     buffer_size_cap: usize,
     buffer_init_cap: usize,
     pool: ArrayQueue<Vec<u8>>,
+    metrics: BufferPoolMetrics,
 }
 
 impl BufferPool {
@@ -37,14 +39,21 @@ impl BufferPool {
             pool: ArrayQueue::new(pool_cap),
             buffer_size_cap,
             buffer_init_cap,
+            metrics: Default::default(),
         }
     }
 
     pub fn get(self: &Arc<Self>) -> PooledBuf {
-        let buf = self
-            .pool
-            .pop()
-            .unwrap_or_else(|| Vec::with_capacity(self.buffer_init_cap));
+        let buf = match self.pool.pop() {
+            Some(buf) => {
+                self.metrics.reuses.inc();
+                buf
+            }
+            None => {
+                self.metrics.creations.inc();
+                Vec::with_capacity(self.buffer_init_cap)
+            }
+        };
         debug_assert_eq!(buf.len(), 0);
         PooledBuf(buf, self.clone())
     }
@@ -64,7 +73,15 @@ impl BufferPool {
         buf.shrink_to(self.buffer_size_cap);
 
         // ArrayQueue will make sure to drop the buffer if capacity is exceeded
-        let _ = self.pool.push(buf);
+        match self.pool.push(buf) {
+            Ok(()) => self.metrics.returns.inc(),
+            Err(_buf) => self.metrics.discards.inc(),
+        };
+    }
+
+    #[cfg(feature = "metrics")]
+    pub(crate) fn snapshot_metrics(&self) -> BufferPoolMetrics {
+        self.metrics.clone()
     }
 }
 
diff --git a/src/conn/mod.rs b/src/conn/mod.rs
@@ -38,6 +38,7 @@ use crate::{
     consts::{CapabilityFlags, Command, StatusFlags},
     error::*,
     io::Stream,
+    metrics::ConnMetrics,
     opts::Opts,
     queryable::{
         query_result::{QueryResult, ResultSetMeta},
@@ -56,6 +57,8 @@ pub mod stmt_cache;
 
 /// Helper that asynchronously disconnects the givent connection on the default tokio executor.
 fn disconnect(mut conn: Conn) {
+    conn.metrics().disconnects.inc();
+
     let disconnected = conn.inner.disconnected;
 
     // Mark conn as disconnected.
@@ -114,6 +117,7 @@ struct ConnInner {
     /// One-time connection-level infile handler.
     infile_handler:
         Option<Pin<Box<dyn Future<Output = crate::Result<InfileData>> + Send + Sync + 'static>>>,
+    conn_metrics: Arc<ConnMetrics>,
 }
 
 impl fmt::Debug for ConnInner {
@@ -133,6 +137,7 @@ impl fmt::Debug for ConnInner {
 impl ConnInner {
     /// Constructs an empty connection.
     fn empty(opts: Opts) -> ConnInner {
+        let conn_metrics: Arc<ConnMetrics> = Default::default();
         ConnInner {
             capabilities: opts.get_capabilities(),
             status: StatusFlags::empty(),
@@ -147,14 +152,15 @@ impl ConnInner {
             tx_status: TxStatus::None,
             last_io: Instant::now(),
             wait_timeout: Duration::from_secs(0),
-            stmt_cache: StmtCache::new(opts.stmt_cache_size()),
+            stmt_cache: StmtCache::new(opts.stmt_cache_size(), conn_metrics.clone()),
             socket: opts.socket().map(Into::into),
             opts,
             nonce: Vec::default(),
             auth_plugin: AuthPlugin::MysqlNativePassword,
             auth_switched: false,
             disconnected: false,
             infile_handler: None,
+            conn_metrics,
         }
     }
 
@@ -166,6 +172,18 @@ impl ConnInner {
             .as_mut()
             .ok_or_else(|| DriverError::ConnectionClosed.into())
     }
+
+    fn set_pool(&mut self, pool: Option<Pool>) {
+        let conn_metrics = if let Some(ref pool) = pool {
+            Arc::clone(&pool.inner.metrics.conn)
+        } else {
+            Default::default()
+        };
+        self.conn_metrics = Arc::clone(&conn_metrics);
+        self.stmt_cache.conn_metrics = conn_metrics;
+
+        self.pool = pool;
+    }
 }
 
 /// MySql server connection.
@@ -854,6 +872,8 @@ impl Conn {
             conn.read_wait_timeout().await?;
             conn.run_init_commands().await?;
 
+            conn.metrics().connects.inc();
+
             Ok(conn)
         }
         .boxed()
@@ -957,7 +977,10 @@ impl Conn {
 
         self.inner.stmt_cache.clear();
         self.inner.infile_handler = None;
-        self.inner.pool = pool;
+        self.inner.set_pool(pool);
+
+        // TODO: clear some metrics?
+
         Ok(())
     }
 
@@ -1062,6 +1085,10 @@ impl Conn {
 
         Ok(BinlogStream::new(self))
     }
+
+    pub(crate) fn metrics(&self) -> &ConnMetrics {
+        &self.inner.conn_metrics
+    }
 }
 
 #[cfg(test)]
diff --git a/src/conn/pool/futures/get_conn.rs b/src/conn/pool/futures/get_conn.rs
@@ -140,7 +140,8 @@ impl Future for GetConn {
 
                     return match result {
                         Ok(mut c) => {
-                            c.inner.pool = Some(pool);
+                            c.inner.set_pool(Some(pool));
+                            c.metrics().connects.inc();
                             Poll::Ready(Ok(c))
                         }
                         Err(e) => {
@@ -156,7 +157,7 @@ impl Future for GetConn {
                             self.inner = GetConnInner::Done;
 
                             let pool = self.pool_take();
-                            checked_conn.inner.pool = Some(pool);
+                            checked_conn.inner.set_pool(Some(pool));
                             return Poll::Ready(Ok(checked_conn));
                         }
                         Err(_) => {
diff --git a/src/conn/pool/mod.rs b/src/conn/pool/mod.rs
@@ -25,6 +25,7 @@ use std::{
 use crate::{
     conn::{pool::futures::*, Conn},
     error::*,
+    metrics::PoolMetrics,
     opts::{Opts, PoolOpts},
     queryable::transaction::{Transaction, TxOpts, TxStatus},
 };
@@ -177,6 +178,7 @@ pub struct Inner {
     close: atomic::AtomicBool,
     closed: atomic::AtomicBool,
     exchange: Mutex<Exchange>,
+    pub(crate) metrics: PoolMetrics,
 }
 
 /// Asynchronous pool of MySql connections.
@@ -190,7 +192,7 @@ pub struct Inner {
 #[derive(Debug, Clone)]
 pub struct Pool {
     opts: Opts,
-    inner: Arc<Inner>,
+    pub(super) inner: Arc<Inner>,
     drop: mpsc::UnboundedSender<Option<Conn>>,
 }
 
@@ -219,6 +221,7 @@ impl Pool {
                     exist: 0,
                     recycler: Some((rx, pool_opts)),
                 }),
+                metrics: Default::default(),
             }),
             drop: tx,
         }
@@ -232,6 +235,7 @@ impl Pool {
 
     /// Async function that resolves to `Conn`.
     pub fn get_conn(&self) -> GetConn {
+        self.inner.metrics.gets.inc();
         GetConn::new(self)
     }
 
@@ -249,6 +253,11 @@ impl Pool {
         DisconnectPool::new(self)
     }
 
+    #[cfg(feature = "metrics")]
+    pub fn snapshot_metrics(&self) -> PoolMetrics {
+        self.inner.metrics.clone()
+    }
+
     /// A way to return connection taken from a pool.
     fn return_conn(&mut self, conn: Conn) {
         // NOTE: we're not in async context here, so we can't block or return NotReady
@@ -264,18 +273,23 @@ impl Pool {
         {
             let mut exchange = self.inner.exchange.lock().unwrap();
             if exchange.available.len() < self.opts.pool_opts().active_bound() {
+                self.inner.metrics.returns.inc();
                 exchange.available.push_back(conn.into());
                 if let Some(w) = exchange.waiting.pop() {
                     w.wake();
                 }
                 return;
+            } else {
+                self.inner.metrics.discards.inc();
             }
         }
 
         self.send_to_recycler(conn);
     }
 
     fn send_to_recycler(&self, conn: Conn) {
+        self.inner.metrics.recycler.recycles.inc();
+
         if let Err(conn) = self.drop.send(Some(conn)) {
             let conn = conn.0.unwrap();
 
diff --git a/src/conn/pool/recycler.rs b/src/conn/pool/recycler.rs
@@ -63,18 +63,23 @@ impl Future for Recycler {
         macro_rules! conn_decision {
             ($self:ident, $conn:ident) => {
                 if $conn.inner.stream.is_none() || $conn.inner.disconnected {
+                    self.inner.metrics.recycler.discards.inc();
                     // drop unestablished connection
                     $self.discard.push(futures_util::future::ok(()).boxed());
                 } else if $conn.inner.tx_status != TxStatus::None || $conn.has_pending_result() {
+                    self.inner.metrics.recycler.cleans.inc();
                     $self.cleaning.push($conn.cleanup_for_pool().boxed());
                 } else if $conn.expired() || close {
+                    self.inner.metrics.recycler.discards.inc();
                     $self.discard.push($conn.close_conn().boxed());
                 } else {
                     let mut exchange = $self.inner.exchange.lock().unwrap();
                     if exchange.available.len() >= $self.pool_opts.active_bound() {
                         drop(exchange);
+                        self.inner.metrics.recycler.discards.inc();
                         $self.discard.push($conn.close_conn().boxed());
                     } else {
+                        self.inner.metrics.recycler.returns.inc();
                         exchange.available.push_back($conn.into());
                         if let Some(w) = exchange.waiting.pop() {
                             w.wake();
diff --git a/src/conn/routines/exec.rs b/src/conn/routines/exec.rs
@@ -25,6 +25,8 @@ impl<'a> ExecRoutine<'a> {
 
 impl Routine<()> for ExecRoutine<'_> {
     fn call<'a>(&'a mut self, conn: &'a mut Conn) -> BoxFuture<'a, crate::Result<()>> {
+        conn.metrics().routines.execs.inc();
+
         #[cfg(feature = "tracing")]
         let span = info_span!(
             "mysql_async::exec",
diff --git a/src/conn/routines/next_set.rs b/src/conn/routines/next_set.rs
@@ -24,6 +24,8 @@ where
     P: Protocol,
 {
     fn call<'a>(&'a mut self, conn: &'a mut Conn) -> BoxFuture<'a, crate::Result<()>> {
+        conn.metrics().routines.next_sets.inc();
+
         #[cfg(feature = "tracing")]
         let span = debug_span!(
             "mysql_async::next_set",
diff --git a/src/conn/routines/ping.rs b/src/conn/routines/ping.rs
@@ -14,6 +14,8 @@ pub struct PingRoutine;
 
 impl Routine<()> for PingRoutine {
     fn call<'a>(&'a mut self, conn: &'a mut Conn) -> BoxFuture<'a, crate::Result<()>> {
+        conn.metrics().routines.pings.inc();
+
         #[cfg(feature = "tracing")]
         let span = debug_span!("mysql_async::ping", mysql_async.connection.id = conn.id());
 
diff --git a/src/conn/routines/prepare.rs b/src/conn/routines/prepare.rs
@@ -26,6 +26,8 @@ impl PrepareRoutine {
 
 impl Routine<Arc<StmtInner>> for PrepareRoutine {
     fn call<'a>(&'a mut self, conn: &'a mut Conn) -> BoxFuture<'a, crate::Result<Arc<StmtInner>>> {
+        conn.metrics().routines.prepares.inc();
+
         #[cfg(feature = "tracing")]
         let span = info_span!(
             "mysql_async::prepare",
diff --git a/src/conn/routines/query.rs b/src/conn/routines/query.rs
@@ -22,6 +22,8 @@ impl<'a> QueryRoutine<'a> {
 
 impl Routine<()> for QueryRoutine<'_> {
     fn call<'a>(&'a mut self, conn: &'a mut Conn) -> BoxFuture<'a, crate::Result<()>> {
+        conn.metrics().routines.queries.inc();
+
         #[cfg(feature = "tracing")]
         let span = info_span!(
             "mysql_async::query",
diff --git a/src/conn/routines/reset.rs b/src/conn/routines/reset.rs
@@ -14,6 +14,8 @@ pub struct ResetRoutine;
 
 impl Routine<()> for ResetRoutine {
     fn call<'a>(&'a mut self, conn: &'a mut Conn) -> BoxFuture<'a, crate::Result<()>> {
+        conn.metrics().routines.resets.inc();
+
         #[cfg(feature = "tracing")]
         let span = debug_span!("mysql_async::reset", mysql_async.connection.id = conn.id());
 
diff --git a/src/conn/stmt_cache.rs b/src/conn/stmt_cache.rs
diff --git a/src/lib.rs b/src/lib.rs
diff --git a/src/metrics.rs b/src/metrics.rs

Original file line number	Diff line number	Diff line change
`@@ -24,6 +24,8 @@ where`
`24`	`24`	`P: Protocol,`
`25`	`25`	`{`
`26`	`26`	`fn call<'a>(&'a mut self, conn: &'a mut Conn) -> BoxFuture<'a, crate::Result<()>> {`
	`27`	`+ conn.metrics().routines.next_sets.inc();`
	`28`	`+`
`27`	`29`	`#[cfg(feature = "tracing")]`
`28`	`30`	`let span = debug_span!(`
`29`	`31`	`"mysql_async::next_set",`