use swap strategy in SharedWal

Author: MarinPostma

libsql-wal/src/registry.rs

@@ -3,7 +3,7 @@ use std::num::NonZeroU64;
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
-use std::time::Instant;
+use std::time::{Duration, Instant};
 
 use dashmap::DashMap;
 use libsql_sys::ffi::Sqlite3DbHeader;
@@ -27,6 +27,9 @@ use crate::segment::Segment;
 use crate::segment::{current::CurrentSegment, sealed::SealedSegment};
 use crate::shared_wal::{SharedWal, SwapLog};
 use crate::storage::{OnStoreCallback, Storage};
+use crate::swap_strategy::duration::DurationSwapStrategy;
+use crate::swap_strategy::frame_count::FrameCountSwapStrategy;
+use crate::swap_strategy::SwapStrategy;
 use crate::transaction::TxGuard;
 use crate::{LibsqlFooter, LIBSQL_PAGE_SIZE};
 use libsql_sys::name::NamespaceName;
@@ -332,6 +335,17 @@ where
 
         let (new_frame_notifier, _) = tokio::sync::watch::channel(next_frame_no.get() - 1);
 
+        // FIXME: make swap strategy configurable
+        // This strategy will perform a swap if either the wal is bigger than 20k frames, or older
+        // than 10 minutes, or if the frame count is greater than a 1000 and the wal was last
+        // swapped more than 30 secs ago
+        let swap_strategy = Box::new(
+            DurationSwapStrategy::new(Duration::from_secs(5 * 60))
+                .or(FrameCountSwapStrategy::new(20_000))
+                .or(FrameCountSwapStrategy::new(1000)
+                    .and(DurationSwapStrategy::new(Duration::from_secs(30)))),
+        );
+
         let shared = Arc::new(SharedWal {
             current,
             wal_lock: Default::default(),
@@ -347,8 +361,8 @@ where
             )),
             shutdown: false.into(),
             checkpoint_notifier: self.checkpoint_notifier.clone(),
-            max_segment_size: 1000.into(),
             io: self.io.clone(),
+            swap_strategy,
         });
 
         self.opened

libsql-wal/src/shared_wal.rs

@@ -1,5 +1,5 @@
 use std::collections::BTreeMap;
-use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::sync::Arc;
 use std::time::Instant;
 
@@ -16,6 +16,7 @@ use crate::io::file::FileExt;
 use crate::io::Io;
 use crate::replication::storage::ReplicateFromStorage;
 use crate::segment::current::CurrentSegment;
+use crate::swap_strategy::SwapStrategy;
 use crate::transaction::{ReadTransaction, Savepoint, Transaction, TxGuard, WriteTransaction};
 use libsql_sys::name::NamespaceName;
 
@@ -46,15 +47,14 @@ pub struct SharedWal<IO: Io> {
     pub(crate) registry: Arc<dyn SwapLog<IO>>,
     #[allow(dead_code)] // used by replication
     pub(crate) checkpointed_frame_no: AtomicU64,
-    /// max frame_no acknoledged by the durable storage
+    /// max frame_no acknowledged by the durable storage
     pub(crate) durable_frame_no: Arc<Mutex<u64>>,
     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>,
-    /// maximum size the segment is allowed to grow
-    pub(crate) max_segment_size: AtomicUsize,
     pub(crate) io: Arc<IO>,
+    pub(crate) swap_strategy: Box<dyn SwapStrategy>,
 }
 
 impl<IO: Io> SharedWal<IO> {
@@ -275,9 +275,10 @@ impl<IO: Io> SharedWal<IO> {
         }
 
         if tx.is_commited()
-            && current.count_committed() > self.max_segment_size.load(Ordering::Relaxed)
+            && self.swap_strategy.should_swap(current.count_committed())
         {
             self.swap_current(&tx)?;
+            self.swap_strategy.swapped();
         }
 
         Ok(())