perf: reduce clock reads per request from 9 to 3

pgdog-stats/src/client.rs

@@ -4,7 +4,6 @@ use crate::{MemoryStats, state::State};
 
 use std::ops::Add;
 use std::time::Duration;
-use std::time::SystemTime;
 
 /// Client statistics.
 #[derive(Copy, Clone, Debug, Serialize, Deserialize)]
@@ -31,8 +30,6 @@ pub struct Stats {
     pub wait_time: Duration,
     /// Current client state.
     pub state: State,
-    /// Last time this client sent a query.
-    pub last_request: SystemTime,
     /// Number of bytes used by the stream buffer, where all the messages
     /// are stored until they are processed.
     pub memory_stats: MemoryStats,
@@ -62,7 +59,6 @@ impl Stats {
             query_time: Duration::from_secs(0),
             wait_time: Duration::from_secs(0),
             state: State::Idle,
-            last_request: SystemTime::now(),
             memory_stats: MemoryStats::default(),
             prepared_statements: 0,
             locked: false,
@@ -86,7 +82,6 @@ impl Add for Stats {
             query_time: self.query_time.saturating_add(rhs.query_time),
             wait_time: self.wait_time.saturating_add(rhs.wait_time),
             state: rhs.state, // Not summed
-            last_request: self.last_request.max(rhs.last_request),
             memory_stats: self.memory_stats + rhs.memory_stats,
             prepared_statements: self.prepared_statements + rhs.prepared_statements,
             locked: rhs.locked, // Not summed either

pgdog/Cargo.toml

@@ -15,6 +15,7 @@ fips = ["tokio-rustls/fips"]
 bench = []
 
 [dependencies]
+foldhash = { version = "0.2", default-features = false }
 bon.workspace = true
 pin-project = "1"
 tokio.workspace = true

pgdog/src/admin/show_clients.rs

@@ -94,7 +94,7 @@ impl Command for ShowClients {
                 .add("connect_time", format_time(client.connected_at))
                 .add(
                     "last_request",
-                    format_time(DateTime::from(client.stats.last_request)),
+                    format_time(DateTime::from(client.stats.last_request())),
                 )
                 .add("queries", client.stats.queries)
                 .add("transactions", client.stats.transactions)

pgdog/src/auth/token_cache.rs

@@ -21,8 +21,8 @@ use tokio::sync::OnceCell;
 use tracing::error;
 
 use super::Error;
-use crate::tasks;
 use crate::{backend::pool::Connection, config::config};
+use crate::{tasks, util::LruHasher};
 
 /// Client token cache.
 ///
@@ -101,14 +101,16 @@ struct Stats {
 #[derive(Clone, Debug)]
 pub(crate) struct AuthTokenCache {
     // Lookups use peek to preserve FIFO eviction order.
-    inner: Arc<Mutex<LruCache<Key, Arc<Entry>>>>,
+    inner: Arc<Mutex<LruCache<Key, Arc<Entry>, LruHasher>>>,
     stats: Arc<Stats>,
 }
 
 impl Default for AuthTokenCache {
     fn default() -> Self {
         Self {
-            inner: Arc::new(Mutex::new(LruCache::unbounded())),
+            inner: Arc::new(Mutex::new(LruCache::unbounded_with_hasher(
+                Default::default(),
+            ))),
             stats: Arc::new(Stats::default()),
         }
     }

pgdog/src/backend/pool/connection/binding.rs

@@ -12,6 +12,7 @@ use crate::{
 };
 
 use futures::future::join_all;
+use tokio::time::Instant;
 
 use super::*;
 use crate::util::safe_sleep;
@@ -209,6 +210,17 @@ impl Binding {
         }
     }
 
+    pub(crate) fn take_last_ready_for_query(&mut self) -> Option<Instant> {
+        match self {
+            Binding::Direct(direct) => direct.server.take_last_ready_for_query(),
+            Binding::MultiShard(servers) => servers
+                .iter_mut()
+                .filter_map(|server| server.take_last_ready_for_query())
+                .max(),
+            Binding::Transaction(_) | Binding::Admin(_) | Binding::NotConnected => None,
+        }
+    }
+
     /// Protocol is out of sync due to an error in extended protocol.
     pub(crate) fn out_of_sync(&self) -> bool {
         match self {

pgdog/src/backend/pool/connection/mirror/mod.rs

@@ -13,7 +13,7 @@ use crate::backend::Cluster;
 use crate::config::{ConfigAndUsers, config};
 use crate::frontend::client::query_engine::{QueryEngine, QueryEngineContext};
 use crate::frontend::client::timeouts::Timeouts;
-use crate::frontend::client::transaction_type::Transaction;
+use crate::frontend::client::transaction_type::{StatementStart, Transaction};
 use crate::frontend::{ClientComms, PreparedStatements};
 use crate::net::{FrontendPid, Parameter, Parameters, Stream};
 use crate::parameters::ParameterStore;
@@ -47,6 +47,7 @@ pub(crate) struct Mirror {
     pub(crate) stream: Stream,
     /// Transaction state.
     pub(crate) transaction: Option<Transaction>,
+    pub(crate) statement_start: StatementStart,
 }
 
 impl Mirror {
@@ -61,6 +62,7 @@ impl Mirror {
             timeouts: Timeouts::from_config(&config.config.general),
             stream: Stream::dev_null(),
             transaction: None,
+            statement_start: StatementStart::default(),
         }
     }
 

pgdog/src/backend/pool/connection/mod.rs

@@ -2,6 +2,7 @@
 
 use futures::future::try_join_all;
 use tokio::select;
+use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
 
 use crate::{
@@ -459,6 +460,10 @@ impl Connection {
         self.binding.done() && !self.binding.is_locked()
     }
 
+    pub(crate) fn take_last_ready_for_query(&mut self) -> Option<Instant> {
+        self.binding.take_last_ready_for_query()
+    }
+
     /// Lock this connection to the client, preventing it's
     /// release back into the pool.
     pub(crate) fn lock(&mut self, lock: bool) {

pgdog/src/backend/pool/lb/test/mod.rs

@@ -454,7 +454,7 @@ async fn test_lsn_filtering() {
 
         let mut route = Route::read(ShardWithPriority::default());
         route.lsn = Some(caught_up);
-        let request = Request::new(Default::default(), &route, false);
+        let request = Request::new(Default::default(), &route, false, Instant::now());
 
         // A replica exactly at the requested LSN is eligible; the lagging one is not.
         for _ in 0..4 {
@@ -464,7 +464,7 @@ async fn test_lsn_filtering() {
 
         // Fall back to primary when every replica is behind the requested LSN.
         route.lsn = Some(ahead);
-        let request = Request::new(Default::default(), &route, false);
+        let request = Request::new(Default::default(), &route, false, Instant::now());
         let conn = replicas.get(&request).await.expect("primary fallback");
         assert_eq!(conn.pool.id(), primary_id);
     }
@@ -494,7 +494,7 @@ async fn test_include_primary_if_replica_lag() {
         let mut route = Route::read(ShardWithPriority::default());
         for lsn in [None, Some(200), Some(201)] {
             route.lsn = lsn;
-            let request = Request::new(Default::default(), &route, read_only);
+            let request = Request::new(Default::default(), &route, read_only, Instant::now());
 
             // Round robin must never select the primary while an eligible replica exists.
             for _ in 0..4 {
@@ -539,6 +539,7 @@ async fn test_user_read_only_include_primary() {
         Default::default(),
         &Route::read(ShardWithPriority::default()),
         true,
+        Instant::now(),
     );
 
     // Try getting connections multiple times and verify primary is never used
@@ -588,6 +589,7 @@ async fn test_user_read_only_include_primary_no_replicas() {
         Default::default(),
         &Route::read(ShardWithPriority::default()),
         true,
+        Instant::now(),
     );
 
     // Should only use primary since replica is banned

pgdog/src/backend/pool/request.rs

@@ -26,10 +26,15 @@ pub(crate) struct Request {
 
 impl Request {
     /// Create new request for client given a query route.
-    pub(crate) fn new(id: FrontendPid, route: &Route, read_only: bool) -> Self {
+    pub(crate) fn new(
+        id: FrontendPid,
+        route: &Route,
+        read_only: bool,
+        created_at: Instant,
+    ) -> Self {
         Self {
             id,
-            created_at: Instant::now(),
+            created_at,
             read: route.is_read(),
             lsn: route.lsn,
             read_only,

pgdog/src/backend/prepared_statements.rs

@@ -13,6 +13,7 @@ use crate::{
         messages::{ParameterDescription, RowDescription, parse::Parse},
     },
     state::State,
+    util::LruHasher,
 };
 use crate::{net::ErrorResponse, util::time::deadline};
 use parking_lot::RwLock;
@@ -97,7 +98,7 @@ pub(super) enum HandleResult {
 #[derive(Debug)]
 pub(crate) struct PreparedStatements {
     global_cache: Arc<RwLock<GlobalCache>>,
-    local_cache: LruCache<String, LocalStatement>,
+    local_cache: LruCache<String, LocalStatement, LruHasher>,
     state: ProtocolState,
     // Prepared statements being prepared now on the connection.
     parses: VecDeque<String>,
@@ -123,7 +124,7 @@ impl PreparedStatements {
     pub(crate) fn new(oids: Arc<Oids>) -> Self {
         Self {
             global_cache: frontend::PreparedStatements::global(),
-            local_cache: LruCache::unbounded(),
+            local_cache: LruCache::unbounded_with_hasher(Default::default()),
             state: ProtocolState::default(),
             parses: VecDeque::new(),
             parameter_describes: VecDeque::new(),

pgdog/src/backend/server.rs

@@ -145,6 +145,7 @@ pub(crate) struct Server {
     re_synced: bool,
     replication_mode: bool,
     statement_executed: bool,
+    last_ready_for_query: Option<Instant>,
     sending_request: bool,
     pooler_mode: PoolerMode,
     stream_buffer: MessageBuffer,
@@ -445,6 +446,7 @@ impl Server {
             sync_prepared: false,
             in_transaction: false,
             statement_executed: false,
+            last_ready_for_query: None,
             re_synced: false,
             sending_request: false,
             pooler_mode: PoolerMode::Transaction,
@@ -629,6 +631,7 @@ impl Server {
         match message.code() {
             'Z' => {
                 let now = Instant::now();
+                self.last_ready_for_query = Some(now);
                 let rfq = ReadyForQuery::from_bytes(message.payload())?;
 
                 self.stats.query(now, rfq.status == 'T');
@@ -841,6 +844,10 @@ impl Server {
         self.statement_executed
     }
 
+    pub(crate) fn take_last_ready_for_query(&mut self) -> Option<Instant> {
+        self.last_ready_for_query.take()
+    }
+
     /// The server connection permanently failed.
     pub(crate) fn error(&self) -> bool {
         self.stats().get_state() == State::Error
@@ -1428,6 +1435,7 @@ pub(crate) mod test {
                 stream_buffer: MessageBuffer::new(4096, None),
                 disconnect_reason: None,
                 statement_executed: false,
+                last_ready_for_query: None,
                 sending_request: false,
                 password_attempts: 1,
                 max_age: None,

pgdog/src/frontend/client/mod.rs

@@ -8,9 +8,9 @@ use std::pin::pin;
 use std::sync::Arc;
 use std::time::Duration;
 
-use chrono::{DateTime, Utc};
 use pgdog_config::users::PasswordKind;
 use timeouts::Timeouts;
+use tokio::time::Instant;
 use tokio::{select, spawn};
 use tracing::{Level as LogLevel, debug, enabled, error, info, trace, warn};
 
@@ -42,7 +42,7 @@ pub(crate) mod transaction_type;
 
 use query_engine::QueryEngine;
 pub(crate) use sticky::Sticky;
-pub(crate) use transaction_type::{QueryTimestamps, Transaction, TransactionType};
+pub(crate) use transaction_type::{QueryTimestamps, StatementStart, Transaction, TransactionType};
 
 /// PostgreSQL client.
 ///
@@ -96,7 +96,8 @@ pub(crate) struct Client {
     /// Maximum query message size before a warning is logged.
     query_size_limit: Option<usize>,
     /// When we received the first message of the current request.
-    statement_start: DateTime<Utc>,
+    request_start: Instant,
+    statement_start: StatementStart,
 }
 
 /// Inputs to the per-user client certificate check.
@@ -472,7 +473,8 @@ impl Client {
             database,
             query_log_stdout: false,
             query_size_limit: None,
-            statement_start: Utc::now(),
+            request_start: Instant::now(),
+            statement_start: StatementStart::default(),
         }))
     }
 
@@ -512,7 +514,8 @@ impl Client {
             database: "pgdog".to_string(),
             query_log_stdout: false,
             query_size_limit: None,
-            statement_start: Utc::now(),
+            request_start: Instant::now(),
+            statement_start: StatementStart::default(),
         }
     }
 
@@ -658,10 +661,12 @@ impl Client {
             QueryEngineResult::Split { requests, extended } => {
                 let mut requests = requests.into_iter();
                 if extended {
-                    self.transaction.get_or_insert(Transaction::started_at(
-                        TransactionType::Implicit,
-                        self.statement_start,
-                    ));
+                    self.transaction.get_or_insert_with(|| {
+                        Transaction::started_at(
+                            TransactionType::Implicit,
+                            self.statement_start.time(),
+                        )
+                    });
                 }
 
                 while let Some(mut request) = requests.next() {
@@ -678,6 +683,10 @@ impl Client {
                         }
                         _ => panic!("query engine cannot split requests twice"),
                     }
+
+                    if requests.len() > 0 {
+                        self.request_start = Instant::now();
+                    }
                 }
             }
         }
@@ -734,7 +743,8 @@ impl Client {
 
             if !has_set_time {
                 has_set_time = true;
-                self.statement_start = Utc::now();
+                self.request_start = Instant::now();
+                self.statement_start.reset();
             }
 
             // Terminate (B & F).
@@ -747,11 +757,11 @@ impl Client {
         }
 
         if enabled!(LogLevel::DEBUG) {
-            let elapsed_time = Utc::now() - self.statement_start;
+            let elapsed_time = self.request_start.elapsed();
             if !enabled!(LogLevel::TRACE) {
                 debug!(
                     "request buffered [{:.4}ms] {:?}",
-                    elapsed_time.as_seconds_f64() * 1000.0,
+                    elapsed_time.as_secs_f64() * 1000.0,
                     self.client_request
                         .messages
                         .iter()
@@ -761,7 +771,7 @@ impl Client {
             } else {
                 trace!(
                     "request buffered [{:.4}ms]\n{:#?}",
-                    elapsed_time.as_seconds_f64() * 1000.0,
+                    elapsed_time.as_secs_f64() * 1000.0,
                     self.client_request,
                 );
             }