sqlsrv 0.14.0

//! Auto-vacuum runtime.
//!
//! # Notes
//! _This is currently highly experimental and untested_.

use std::{
  cmp,
  sync::Arc,
  thread,
  time::{Duration, Instant}
};

use parking_lot::{Condvar, Mutex, MutexGuard};

use crate::ConnPool;


/// Rquested vacuum action returned by application callback.
pub enum Vacuum {
  /// Do not vacuum this cycle.  Wait for another vacuum event before trying
  /// again.
  Nothing,

  /// Run a vacuum and then wait for a vacuum event before running again.
  Once(usize),

  /// Run vacuum and then run another vacuum
  Repeat(usize)
}

/// Auto-vacuum parameters.
pub struct Params {
  /// If set to `true` an autovacuum will be triggered automatically at start
  /// of the background thread.
  ///
  /// If `false`, a vacuum must be triggered using [`VacCtx::vacuum()`].
  pub autorun: bool,

  /// Closre used to calculate the number of pages to vacuum.
  ///
  /// The input argument is the current database "freelist" count (number of
  /// pages that can be cleaned).  The closure should return the number of
  /// pages it wants to clean.
  ///
  /// Return 0 to omit running vacuum.
  pub page_count: Box<dyn FnMut(usize) -> Vacuum + Send>,

  /// The minimum amount of time to wait between each clean cycle.
  pub cooldown: Duration
}

impl Default for Params {
  /// Construct autovacuum `Params` with:
  /// - `autorun` set to `false`, meaning it must be triggered to run
  /// - A vacuum page count which is half of the current freelist count (but
  ///   capped at 256) and does nothing if the count is below 16.
  /// - I
  fn default() -> Self {
    // Vacuum half of the pages, but cap at 256 (for 4K pages, this is 1MB in
    // total).
    let page_count = |npages| {
      let n = cmp::min(npages / 2, 256);

      if n > 16 {
        Vacuum::Repeat(n)
      } else {
        Vacuum::Nothing
      }
    };

    Self {
      autorun: false,
      page_count: Box::new(page_count),
      cooldown: Duration::from_secs(30)
    }
  }
}

#[derive(Default)]
struct Mutable {
  shutdown: bool,
  do_vacuum: bool
}

impl Mutable {
  fn new() -> Mutex<Self> {
    Mutex::new(Self::default())
  }
}

#[derive(Default)]
struct Shared {
  rw: Mutex<Mutable>,
  signal: Condvar
}

impl Shared {
  fn new() -> Arc<Self> {
    let slf = Self {
      rw: Mutable::new(),
      signal: Condvar::new()
    };
    Arc::new(slf)
  }

  fn lock(&self) -> MutexGuard<'_, Mutable> {
    self.rw.lock()
  }

  #[inline]
  pub(crate) fn with_lock<F, R>(&self, f: F) -> R
  where
    F: FnOnce(&mut Mutable) -> R
  {
    let mut rw = self.lock();
    f(&mut rw)
  }
}

/// Auto-vacuum context.
pub struct VacCtx {
  sh: Arc<Shared>,
  jh: Option<thread::JoinHandle<()>>
}

impl VacCtx {
  pub fn vacuum(&self) {
    self.sh.with_lock(|rw| {
      rw.do_vacuum = true;
      self.sh.signal.notify_one();
    });
  }

  pub const fn take_jh(&mut self) -> Option<thread::JoinHandle<()>> {
    self.jh.take()
  }
}

impl Drop for VacCtx {
  fn drop(&mut self) {
    self.sh.with_lock(|rw| {
      rw.shutdown = true;
      self.sh.signal.notify_one();
    });

    if let Some(jh) = self.jh.take() {
      let _ = jh.join();
    }
  }
}


/// Run the autovacuum background thread.
///
/// The returned [`VacCtx`] can be used to trigger a vacuum cycle by calling
/// [`VacCtx::vacuum()`].
///
/// Dropping the `VacCtx` will terminate the vacuum thread.
#[must_use]
pub fn run(cp: ConnPool, params: Params) -> VacCtx {
  let sh = Shared::new();

  let sh2 = Arc::clone(&sh);
  let jh = thread::spawn(move || {
    autovacuum(&sh2, &cp, params);
  });

  VacCtx { sh, jh: Some(jh) }
}

#[allow(clippy::significant_drop_tightening)]
fn autovacuum(
  sh: &Arc<Shared>,
  cp: &ConnPool,
  Params {
    autorun,
    mut page_count,
    cooldown
  }: Params
) {
  // Keep track of when the last vacuum completed
  let mut last_run: Option<Instant> = None;

  let mut rw = sh.lock();

  rw.do_vacuum = autorun;

  loop {
    if rw.shutdown {
      break;
    }

    // If a run has been completed, then wait until at least `cooldown` time
    // has passed before checking again.
    //
    // The role of this block is simply to make sure the vacuum section below
    // is not reached until the cooldown has passed.
    if let Some(lastrun) = last_run {
      // If a vacuum was recently performed, make sure enough time has passed
      // before running it again.
      let now = Instant::now();
      let elapsed = now - lastrun;
      if elapsed < cooldown {
        //let dur = cooldown - elapsed;
        let dur = cooldown.checked_sub(elapsed).unwrap();
        if sh.signal.wait_for(&mut rw, dur).timed_out() {
          last_run = None;
        }
        continue;
      }
      last_run = None;
    }

    if rw.do_vacuum {
      // A vacuum was requested

      // Clear the vacuum flag (unless requested to run continuously)
      rw.do_vacuum = false;

      // Get the current freelist length
      let Ok(nfree) = cp.freelist_count() else {
        // ToDo: log error
        continue;
      };

      //println!("freelist count: {nfree}");

      // Call closure to ask application how many pages it wants to vacuum
      match page_count(nfree) {
        Vacuum::Nothing => {
          // cooldown applies even when doing nothing
          last_run = Some(Instant::now());
        }
        Vacuum::Once(n) => {
          // Perform vacuum, and then wait for a vacuum event
          let wrconn = cp.writer();
          if wrconn.incremental_vacuum(Some(n)).is_err() {
            // ToDo: log error
          }

          last_run = Some(Instant::now());
        }
        Vacuum::Repeat(n) => {
          // Peform vacuum, then wait for cooldown and then run another vacuum
          // cycle
          let wrconn = cp.writer();
          if wrconn.incremental_vacuum(Some(n)).is_err() {
            // ToDo: log error
          }

          // Automatically vacuum again as soon as cooldown has passed
          rw.do_vacuum = true;
          last_run = Some(Instant::now());
        }
      }
    } else {
      // Nothing to do -- wait for an event
      sh.signal.wait(&mut rw);
    }
  }
}

// vim: set ft=rust et sw=2 ts=2 sts=2 cinoptions=2 tw=79 :