armdb 0.11.2

//! Group commit: one `fdatasync` serves every record appended before it
//! started, so `durable()` does not cost one fsync per transaction.
//!
//! The gate is deliberately unaware of *what* it syncs — the caller passes the
//! closure — so it can be tested without touching a file.

use crate::error::DbResult;
use crate::sync::{Condvar, Mutex};

struct SyncState {
    /// Sequence number of the last appended record.
    requested: u64,
    /// Sequence number covered by a completed fsync.
    completed: u64,
    /// A leader is inside `do_fsync` right now.
    in_progress: bool,
}

pub(crate) struct SyncGroup {
    state: Mutex<SyncState>,
    cv: Condvar,
}

impl SyncGroup {
    pub(crate) fn new() -> Self {
        Self {
            state: Mutex::new(SyncState {
                requested: 0,
                completed: 0,
                in_progress: false,
            }),
            cv: Condvar::new(),
        }
    }

    /// Register an appended record. Must be called while the writer lock is
    /// held, so sequence numbers match append order.
    pub(crate) fn record_write(&self) -> u64 {
        let mut st = crate::sync::lock(&self.state);
        st.requested += 1;
        st.requested
    }

    /// Return once a completed fsync covers `seq`.
    pub(crate) fn sync_upto(&self, seq: u64, do_fsync: &dyn Fn() -> DbResult<()>) -> DbResult<()> {
        loop {
            let target = {
                let mut st = crate::sync::lock(&self.state);
                if st.completed >= seq {
                    return Ok(());
                }
                if st.in_progress {
                    // Portable across the three Condvar flavours (`sync.rs`).
                    drop(crate::sync::cv_wait(&self.cv, st));
                    continue;
                }
                st.in_progress = true;
                st.requested
            };

            let result = do_fsync();

            {
                let mut st = crate::sync::lock(&self.state);
                st.in_progress = false;
                if result.is_ok() {
                    st.completed = st.completed.max(target);
                }
            }
            self.cv.notify_all();
            result?;
        }
    }
}

#[cfg(test)]
mod tests {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};

    use super::*;

    #[test]
    fn txlog_group_one_fsync_serves_a_batch() {
        let group = Arc::new(SyncGroup::new());
        let fsyncs = Arc::new(AtomicUsize::new(0));
        let gate = Arc::new(std::sync::Barrier::new(8));

        let mut handles = Vec::new();
        for _ in 0..8 {
            let group = group.clone();
            let fsyncs = fsyncs.clone();
            let gate = gate.clone();
            handles.push(std::thread::spawn(move || {
                let seq = group.record_write();
                gate.wait(); // make the batch overlap
                group
                    .sync_upto(seq, &|| {
                        fsyncs.fetch_add(1, Ordering::SeqCst);
                        std::thread::sleep(std::time::Duration::from_millis(20));
                        Ok(())
                    })
                    .unwrap();
            }));
        }
        for h in handles {
            h.join().unwrap();
        }

        let n = fsyncs.load(Ordering::SeqCst);
        assert!(n >= 1, "at least one fsync must happen");
        assert!(
            n < 8,
            "a batch of 8 committers must not cost 8 fsyncs, got {n}"
        );
    }

    #[test]
    fn txlog_group_already_synced_returns_without_fsync() {
        let group = SyncGroup::new();
        let seq = group.record_write();
        let calls = AtomicUsize::new(0);
        let f = || {
            calls.fetch_add(1, Ordering::SeqCst);
            Ok(())
        };
        group.sync_upto(seq, &f).unwrap();
        group.sync_upto(seq, &f).unwrap();
        assert_eq!(calls.load(Ordering::SeqCst), 1, "the second wait is free");
    }

    #[test]
    fn txlog_group_propagates_an_fsync_error() {
        let group = SyncGroup::new();
        let seq = group.record_write();
        let err = group
            .sync_upto(seq, &|| Err(crate::DbError::Internal("disk on fire")))
            .expect_err("the fsync error reaches the committer");
        assert!(format!("{err}").contains("disk on fire"));
    }
}