add document lock stats

6 files changed, 272 insertions, 3 deletions

diff --git a/src/api/stats.rs b/src/api/stats.rs
index 45d91da..07521ab 100644
--- a/src/api/stats.rs
+++ b/src/api/stats.rs
@@ -96,6 +96,9 @@ struct DocumentStats {
     saved_bytes_secs_ago: Option<u64>,
     /// How long the last measurement took (copy and save), in milliseconds.
     saved_bytes_took_ms: Option<u64>,
+    /// How long takers of the document's lock waited for it, since start: the node's
+    /// work on the one document queues there.
+    lock_wait: crate::document::lock::LockWaits,
 }
 
 #[derive(Debug, Serialize)]
@@ -245,6 +248,7 @@ async fn collect(state: &AppState) -> StatsResponse {
             saved_bytes: saved.map(|s| s.bytes),
             saved_bytes_secs_ago: saved.map(|s| s.at.elapsed().as_secs()),
             saved_bytes_took_ms: saved.map(|s| s.took.as_millis() as u64),
+            lock_wait: state.doc.waits(),
         },
         catch_up: CatchUpStats {
             sources: node.catch_up_sources,
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 8e6fde7..9cbeff4 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -420,6 +420,94 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
         );
     }
 
+    out.family(
+        "document_lock_wait_seconds",
+        "histogram",
+        "Time waited for the document's lock, by mode (read, write).",
+    );
+    let lock_wait = &document.lock_wait;
+    for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] {
+        let mut cumulative = 0;
+        for (le_us, count) in waits.buckets() {
+            cumulative += count;
+            let le = match le_us {
+                Some(us) => (us as f64 / 1e6).to_string(),
+                None => "+Inf".to_owned(),
+            };
+            out.sample(
+                "document_lock_wait_seconds_bucket",
+                &[("mode", mode), ("le", &le)],
+                cumulative as f64,
+            );
+        }
+        out.sample(
+            "document_lock_wait_seconds_sum",
+            &[("mode", mode)],
+            waits.sum_us as f64 / 1e6,
+        );
+        out.sample(
+            "document_lock_wait_seconds_count",
+            &[("mode", mode)],
+            waits.count as f64,
+        );
+    }
+    out.family(
+        "document_lock_wait_max_seconds",
+        "gauge",
+        "Longest wait for the document's lock since start, by mode.",
+    );
+    for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] {
+        out.sample(
+            "document_lock_wait_max_seconds",
+            &[("mode", mode)],
+            waits.max_us as f64 / 1e6,
+        );
+    }
+
+    out.family(
+        "document_lock_wait_seconds",
+        "histogram",
+        "Time waited for the document's lock, by mode (read, write).",
+    );
+    let lock_wait = &document.lock_wait;
+    for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] {
+        let mut cumulative = 0;
+        for (le_us, count) in waits.buckets() {
+            cumulative += count;
+            let le = match le_us {
+                Some(us) => (us as f64 / 1e6).to_string(),
+                None => "+Inf".to_owned(),
+            };
+            out.sample(
+                "document_lock_wait_seconds_bucket",
+                &[("mode", mode), ("le", &le)],
+                cumulative as f64,
+            );
+        }
+        out.sample(
+            "document_lock_wait_seconds_sum",
+            &[("mode", mode)],
+            waits.sum_us as f64 / 1e6,
+        );
+        out.sample(
+            "document_lock_wait_seconds_count",
+            &[("mode", mode)],
+            waits.count as f64,
+        );
+    }
+    out.family(
+        "document_lock_wait_max_seconds",
+        "gauge",
+        "Longest wait for the document's lock since start, by mode.",
+    );
+    for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] {
+        out.sample(
+            "document_lock_wait_max_seconds",
+            &[("mode", mode)],
+            waits.max_us as f64 / 1e6,
+        );
+    }
+
     let catch_up = &stats.catch_up;
     out.single(
         "catch_up_sources_total",
diff --git a/src/document/lock.rs b/src/document/lock.rs
new file mode 100644
index 0000000..a394b52
--- /dev/null
+++ b/src/document/lock.rs
@@ -0,0 +1,175 @@
+//! The lock around the replica (`AppState::doc`), counting how long its takers wait
+//! (/admin/stats `document.lockWait`): whether the one document is what the node's
+//! work queues behind.
+
+use std::sync::Mutex;
+use std::time::{Duration, Instant};
+
+use serde::Serialize;
+use tokio::sync::{RwLock, RwLockReadGuard, RwLockWriteGuard};
+
+use super::replica::Replica;
+use crate::locks::MutexExt;
+
+/// Upper bounds (inclusive, microseconds) of the wait buckets; a last bucket takes the
+/// rest.
+const WAIT_BUCKETS_US: [u64; 7] = [10, 100, 1_000, 10_000, 100_000, 1_000_000, 10_000_000];
+
+/// `tokio::sync::RwLock<Replica>`, with its waits counted.
+pub(crate) struct DocumentLock {
+    replica: RwLock<Replica>,
+    waits: Mutex<LockWaits>,
+}
+
+/// How long takers of the lock waited for it, by mode, since start.
+#[derive(Debug, Clone, Copy, Default, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct LockWaits {
+    /// Readers: wait for a writer only.
+    pub(crate) read: Waits,
+    /// Writers (applying sync messages, generating them, writes, saves): wait for
+    /// everyone.
+    pub(crate) write: Waits,
+}
+
+#[derive(Debug, Clone, Copy, Default)]
+pub(crate) struct Waits {
+    /// Per bucket of `WAIT_BUCKETS_US`, then the rest; not cumulative.
+    buckets: [u64; WAIT_BUCKETS_US.len() + 1],
+    pub(crate) count: u64,
+    pub(crate) sum_us: u64,
+    pub(crate) max_us: u64,
+}
+
+impl Waits {
+    fn record(&mut self, waited: Duration) {
+        let us = u64::try_from(waited.as_micros()).unwrap_or(u64::MAX);
+        let bucket = WAIT_BUCKETS_US
+            .iter()
+            .position(|&bound| us <= bound)
+            .unwrap_or(WAIT_BUCKETS_US.len());
+        self.buckets[bucket] += 1;
+        self.count += 1;
+        self.sum_us = self.sum_us.saturating_add(us);
+        self.max_us = self.max_us.max(us);
+    }
+
+    /// Each bucket's upper bound in microseconds (None for the last) and count.
+    pub(crate) fn buckets(&self) -> impl Iterator<Item = (Option<u64>, u64)> + '_ {
+        WAIT_BUCKETS_US
+            .iter()
+            .copied()
+            .map(Some)
+            .chain([None])
+            .zip(self.buckets.iter().copied())
+    }
+}
+
+impl Serialize for Waits {
+    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        #[derive(Serialize)]
+        #[serde(rename_all = "camelCase")]
+        struct Bucket {
+            le_us: Option<u64>,
+            count: u64,
+        }
+        #[derive(Serialize)]
+        #[serde(rename_all = "camelCase")]
+        struct Out {
+            count: u64,
+            sum_us: u64,
+            max_us: u64,
+            /// Not cumulative; the last bucket has no bound.
+            buckets: Vec<Bucket>,
+        }
+        Out {
+            count: self.count,
+            sum_us: self.sum_us,
+            max_us: self.max_us,
+            buckets: self
+                .buckets()
+                .map(|(le_us, count)| Bucket { le_us, count })
+                .collect(),
+        }
+        .serialize(serializer)
+    }
+}
+
+#[derive(Clone, Copy)]
+enum Mode {
+    Read,
+    Write,
+}
+
+impl DocumentLock {
+    pub(crate) fn new(replica: Replica) -> Self {
+        Self {
+            replica: RwLock::new(replica),
+            waits: Mutex::default(),
+        }
+    }
+
+    fn waited(&self, mode: Mode, since: Instant) {
+        let waited = since.elapsed();
+        let mut waits = self.waits.lock_or_recover();
+        match mode {
+            Mode::Read => waits.read.record(waited),
+            Mode::Write => waits.write.record(waited),
+        }
+    }
+
+    pub(crate) async fn read(&self) -> RwLockReadGuard<'_, Replica> {
+        let since = Instant::now();
+        let guard = self.replica.read().await;
+        self.waited(Mode::Read, since);
+        guard
+    }
+
+    pub(crate) async fn write(&self) -> RwLockWriteGuard<'_, Replica> {
+        let since = Instant::now();
+        let guard = self.replica.write().await;
+        self.waited(Mode::Write, since);
+        guard
+    }
+
+    /// Outside the runtime only (`spawn_blocking`), like `RwLock::blocking_read`.
+    pub(crate) fn blocking_read(&self) -> RwLockReadGuard<'_, Replica> {
+        let since = Instant::now();
+        let guard = self.replica.blocking_read();
+        self.waited(Mode::Read, since);
+        guard
+    }
+
+    /// Outside the runtime only (`spawn_blocking`), like `RwLock::blocking_write`.
+    pub(crate) fn blocking_write(&self) -> RwLockWriteGuard<'_, Replica> {
+        let since = Instant::now();
+        let guard = self.replica.blocking_write();
+        self.waited(Mode::Write, since);
+        guard
+    }
+
+    pub(crate) fn waits(&self) -> LockWaits {
+        *self.waits.lock_or_recover()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn waits_fall_in_their_buckets() {
+        let mut waits = Waits::default();
+        waits.record(Duration::from_micros(5));
+        waits.record(Duration::from_micros(10));
+        waits.record(Duration::from_millis(2));
+        waits.record(Duration::from_secs(60));
+        let buckets: Vec<_> = waits.buckets().collect();
+        assert_eq!(buckets[0], (Some(10), 2));
+        assert_eq!(buckets[3], (Some(10_000), 1));
+        assert_eq!(buckets[7], (None, 1));
+        assert_eq!(waits.count, 4);
+        assert_eq!(waits.sum_us, 5 + 10 + 2_000 + 60_000_000);
+        assert_eq!(waits.max_us, 60_000_000);
+    }
+}
diff --git a/src/document/mod.rs b/src/document/mod.rs
index 3f489aa..b86d1cc 100644
--- a/src/document/mod.rs
+++ b/src/document/mod.rs
@@ -2,6 +2,7 @@
 //! (merged-view paths over JSON) the HTTP API and remakes work with.
 
 pub(crate) mod file;
+pub(crate) mod lock;
 pub(crate) mod meta;
 pub(crate) mod replica;
 pub(crate) mod store;
diff --git a/src/node.rs b/src/node.rs
index 97dec79..8aab601 100644
--- a/src/node.rs
+++ b/src/node.rs
@@ -85,7 +85,7 @@ impl Node {
         let replica = Replica::new(config.initial_epoch, config.initial_doc);
         let stamp = replica.stamps();
         let state = AppState {
-            doc: Arc::new(RwLock::new(replica)),
+            doc: Arc::new(crate::document::lock::DocumentLock::new(replica)),
             epoch_tx,
             stamp,
             lifecycle: Arc::new(RwLock::new(NodeLifecycle::Joining {
diff --git a/src/state/mod.rs b/src/state/mod.rs
index d8291a7..fec164d 100644
--- a/src/state/mod.rs
+++ b/src/state/mod.rs
@@ -40,7 +40,8 @@ use tokio::sync::{Mutex, RwLock, broadcast, mpsc, watch};
 use tokio::time::Instant;
 use tracing::{debug, error, warn};
 
-use crate::document::replica::{Epoch, Replica, Stamp};
+use crate::document::lock::DocumentLock;
+use crate::document::replica::{Epoch, Stamp};
 use crate::locks::MutexExt;
 use crate::remake::{Host, RemakeState};
 use crate::swarm::sync::{ChangeSet, SyncEvent};
@@ -58,7 +59,7 @@ pub(crate) enum NodeLifecycle {
 #[derive(Clone)]
 pub(crate) struct AppState {
     /// The document and its epoch. Derefs to `AutoCommit`.
-    pub(crate) doc: Arc<RwLock<Replica>>,
+    pub(crate) doc: Arc<DocumentLock>,
 
     /// Copy of `doc.epoch` for cheap, lock-free reads (handshakes, keepalives, /peers).
     /// Only ever changed while holding `doc`'s write lock, together with `doc.epoch`.