fix deadlock

not really related to that, but it started triggerring consistently
after I rebased

- there could be a race condition between a segment being sealed, stored
  and registered for checkpoint (the checkpointer could be notified
  before the segment was added to the tail)
  - the checkpointer would not be notifier when the last reader left the
    segment, so we would hang
- we could enter a busy loop when we got a notification for a namespace
  while there was still an in-flight job for that namespace, blocking
      the runtime.

Author: MarinPostma

libsql-wal/src/checkpointer.rs

@@ -89,6 +89,8 @@ pub struct Checkpointer<P> {
     join_set: JoinSet<(NamespaceName, crate::error::Result<()>)>,
     processing: Vec<NamespaceName>,
     errors: usize,
+    /// previous iteration of the loop resulted in no work being enqueued
+    no_work: bool,
 }
 
 #[allow(private_bounds)]
@@ -111,6 +113,7 @@ where
             join_set: JoinSet::new(),
             processing: Vec::new(),
             errors: 0,
+            no_work: false,
         }
     }
 
@@ -164,17 +167,19 @@ where
             }
             // don't wait if there is stuff to enqueue
             _ = std::future::ready(()), if !self.scheduled.is_empty()
-                && self.join_set.len() < self.max_checkpointing_conccurency => (),
+                && self.join_set.len() < self.max_checkpointing_conccurency && !self.no_work => (),
         }
 
         let n_available = self.max_checkpointing_conccurency - self.join_set.len();
         if n_available > 0 {
+            self.no_work = true;
             for namespace in self
                 .scheduled
                 .difference(&self.checkpointing)
                 .take(n_available)
                 .cloned()
             {
+                self.no_work = false;
                 self.processing.push(namespace.clone());
                 let perform_checkpoint = self.perform_checkpoint.clone();
                 self.join_set.spawn(async move {

libsql-wal/src/registry.rs

@@ -6,9 +6,9 @@ use std::sync::Arc;
 use dashmap::DashMap;
 use libsql_sys::ffi::Sqlite3DbHeader;
 use parking_lot::{Condvar, Mutex};
+use rand::Rng;
 use tokio::sync::{mpsc, Notify, Semaphore};
 use tokio::task::JoinSet;
-use rand::Rng;
 use zerocopy::{AsBytes, FromZeroes};
 
 use crate::checkpointer::CheckpointMessage;
@@ -128,9 +128,8 @@ where
                     update_durable(fno, notifier, durable_frame_no, namespace).await;
                 })
             });
-            self.storage
-                .store(&shared.namespace, sealed.clone(), None, cb);
-            new.tail().push(sealed);
+            new.tail().push(sealed.clone());
+            self.storage.store(&shared.namespace, sealed, None, cb);
         }
 
         shared.current.swap(Arc::new(new));
@@ -313,6 +312,7 @@ where
                 namespace.clone(),
             )),
             shutdown: false.into(),
+            checkpoint_notifier: self.checkpoint_notifier.clone(),
         });
 
         self.opened

libsql-wal/src/replication/replicator.rs

@@ -299,7 +299,6 @@ mod test {
         let mut replica_content = vec![0u8; db_content.len()];
         while let Some(f) = stream.next().await {
             let frame = f.unwrap();
-            dbg!(frame.header().page_no());
             let offset = (frame.header().page_no() as usize - 1) * 4096;
             tmp.as_file()
                 .write_all_at(frame.data(), offset as u64)

libsql-wal/src/segment/current.rs

@@ -39,7 +39,7 @@ pub struct CurrentSegment<F> {
     /// lock
     read_locks: Arc<AtomicU64>,
     sealed: AtomicBool,
-    /// current runnign checksum
+    /// current running checksum
     current_checksum: AtomicU32,
     tail: Arc<SegmentList<SealedSegment<F>>>,
 }
@@ -458,8 +458,9 @@ impl<F> CurrentSegment<F> {
         self.read_locks().fetch_add(1, Ordering::SeqCst);
     }
 
-    pub fn dec_reader_count(&self) {
-        self.read_locks().fetch_sub(1, Ordering::SeqCst);
+    /// return true if the reader count is 0
+    pub fn dec_reader_count(&self) -> bool {
+        self.read_locks().fetch_sub(1, Ordering::SeqCst) - 1 == 0
     }
 
     pub fn read_locks(&self) -> &AtomicU64 {
@@ -1017,7 +1018,7 @@ mod test {
 
         let tmp = Arc::new(tempdir().unwrap());
         {
-            let env = TestEnv::new_io_and_tmp(SyncFailBufferIo::default(), tmp.clone());
+            let env = TestEnv::new_io_and_tmp(SyncFailBufferIo::default(), tmp.clone(), false);
             let conn = env.open_conn("test");
             let shared = env.shared("test");
 
@@ -1038,7 +1039,7 @@ mod test {
         }
 
         {
-            let env = TestEnv::new_io_and_tmp(SyncFailBufferIo::default(), tmp.clone());
+            let env = TestEnv::new_io_and_tmp(SyncFailBufferIo::default(), tmp.clone(), false);
             let conn = env.open_conn("test");
             // the db was recovered: we lost some rows, but it still works
             conn.query_row("select count(*) from test", (), |row| {

libsql-wal/src/segment/list.rs

@@ -81,13 +81,23 @@ where
     where
         F: FileExt,
     {
+        struct Guard<'a>(&'a AtomicBool);
+        impl<'a> Drop for Guard<'a> {
+            fn drop(&mut self) {
+                self.0.store(false, Ordering::SeqCst);
+            }
+        }
+
         if self
             .checkpointing
             .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
             .is_err()
         {
             return Ok(None);
         }
+
+        let _g = Guard(&self.checkpointing);
+
         let mut segs = Vec::new();
         let mut current = self.head.load();
         // find the longest chain of segments that can be checkpointed, iow, segments that do not have
@@ -172,8 +182,6 @@ where
 
         db_file.set_len(size_after as u64 * 4096)?;
 
-        self.checkpointing.store(false, Ordering::SeqCst);
-
         Ok(Some(last_replication_index))
     }
 

libsql-wal/src/segment/sealed.rs

@@ -15,11 +15,12 @@ use zerocopy::{AsBytes, FromZeroes};
 use crate::error::Result;
 use crate::io::buf::{IoBufMut, ZeroCopyBuf};
 use crate::io::file::{BufCopy, FileExt};
-use crate::LIBSQL_MAGIC;
 use crate::io::Inspect;
+use crate::segment::{checked_frame_offset, CheckedFrame};
+use crate::LIBSQL_MAGIC;
 
 use super::compacted::{CompactedSegmentDataFooter, CompactedSegmentDataHeader};
-use super::{frame_offset, page_offset, Frame, FrameHeader, Segment, SegmentHeader, SegmentFlags};
+use super::{frame_offset, page_offset, Frame, Segment, SegmentFlags, SegmentHeader};
 
 /// an immutable, wal segment
 #[derive(Debug)]

libsql-wal/src/shared_wal.rs

@@ -7,7 +7,9 @@ use arc_swap::ArcSwap;
 use crossbeam::deque::Injector;
 use crossbeam::sync::Unparker;
 use parking_lot::{Mutex, MutexGuard};
+use tokio::sync::mpsc;
 
+use crate::checkpointer::CheckpointMessage;
 use crate::error::{Error, Result};
 use crate::io::file::FileExt;
 use crate::io::Io;
@@ -48,6 +50,7 @@ pub struct SharedWal<IO: Io> {
     pub(crate) new_frame_notifier: tokio::sync::watch::Sender<u64>,
     pub(crate) stored_segments: Box<dyn ReplicateFromStorage>,
     pub(crate) shutdown: AtomicBool,
+    pub(crate) checkpoint_notifier: mpsc::Sender<CheckpointMessage>,
 }
 
 impl<IO: Io> SharedWal<IO> {
@@ -87,6 +90,8 @@ impl<IO: Io> SharedWal<IO> {
             created_at: Instant::now(),
             conn_id,
             pages_read: 0,
+            namespace: self.namespace.clone(),
+            checkpoint_notifier: self.checkpoint_notifier.clone(),
         }
     }
 

libsql-wal/src/storage/mod.rs

@@ -339,7 +339,7 @@ impl<IO: Io> Storage for TestStorage<IO> {
                 .entry(namespace.clone())
                 .or_default()
                 .insert(key, (out_path, index));
-            tokio::runtime::Handle::current().block_on(on_store(end_frame_no))
+            tokio::runtime::Handle::current().block_on(on_store(end_frame_no));
         }
     }
 

libsql-wal/src/transaction.rs

@@ -3,8 +3,11 @@ use std::ops::{Deref, DerefMut};
 use std::sync::Arc;
 use std::time::Instant;
 
+use libsql_sys::name::NamespaceName;
 use parking_lot::{ArcMutexGuard, RawMutex};
+use tokio::sync::mpsc;
 
+use crate::checkpointer::CheckpointMessage;
 use crate::segment::current::{CurrentSegment, SegmentIndex};
 use crate::shared_wal::WalLock;
 
@@ -77,8 +80,11 @@ pub struct ReadTransaction<F> {
     /// number of pages read by this transaction. This is used to determine whether a write lock
     /// will be re-acquired.
     pub pages_read: usize,
+    pub namespace: NamespaceName,
+    pub checkpoint_notifier: mpsc::Sender<CheckpointMessage>,
 }
 
+// fixme: clone should probably not be implemented for this type, figure a way to do it
 impl<F> Clone for ReadTransaction<F> {
     fn clone(&self) -> Self {
         self.current.inc_reader_count();
@@ -90,14 +96,21 @@ impl<F> Clone for ReadTransaction<F> {
             created_at: self.created_at,
             conn_id: self.conn_id,
             pages_read: self.pages_read,
+            namespace: self.namespace.clone(),
+            checkpoint_notifier: self.checkpoint_notifier.clone(),
         }
     }
 }
 
 impl<F> Drop for ReadTransaction<F> {
     fn drop(&mut self) {
-        // FIXME: if the count drops to 0, register for compaction.
-        self.current.dec_reader_count();
+        // FIXME: it would be more approriate to wait till the segment is stored before notfying,
+        // because we are not waiting for read to be released before that
+        if self.current.dec_reader_count() && self.current.is_sealed() {
+            let _: Result<_, _> = self
+                .checkpoint_notifier
+                .try_send(self.namespace.clone().into());
+        }
     }
 }