{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}
{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE ScopedTypeVariables #-}

-- | Maintenance of LeiosDB's volatile and immutable partitions.
--
--   * promotion: volatile EBs are copied to the immutable partition once they are old enough;
--   * garbage collection of the volatile partition;
--   * the in-memory 'LeiosDbStats'.
module Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Maintenance
  ( -- * Promotion to the immutable partition
    sqlPromoteToImmutable
  , startCopier

    -- * GC mark
  , sqlGarbageCollect
  , gcMark

    -- * GC sweep
  , SweeperConn (..)
  , SweepState (..)
  , sweepEbBatch
  , sweepOrphanBatch
  , gcReinit

    -- * GC batch sizes
  , defaultGcBatchSize
  , gcOrphanTxBatchSize

    -- * Stats
  , initialStats
  , startVolatileStatsSampler
  , bumpVolatileStats
  , bumpVolatileStatsVar
  , bumpImmutableStats
  ) where

import Cardano.Slotting.Slot (SlotNo (..))
import Control.Concurrent (threadDelay)
import Control.Concurrent.Class.MonadSTM.Strict
  ( StrictTVar
  , check
  , modifyTVar
  , readTVar
  , readTVarIO
  , writeTVar
  )
import Control.Monad (filterM, forever, unless, void, when)
import Control.Monad.Class.MonadThrow
  ( bracket
  , catch
  , displayException
  )
import Control.ResourceRegistry
  ( ResourceRegistry
  , Thread
  , forkLinkedThread
  )
import Control.Tracer (Tracer, traceWith)
import qualified Data.ByteString as BS
import Data.Int (Int64)
import Data.String (fromString)
import qualified Database.SQLite3.Direct as DB
import qualified GHC.Conc as IO (atomically)
import GHC.Stack (HasCallStack)
import Ouroboros.Consensus.Leios.Types
  ( EbHash (..)
  , LeiosPoint (..)
  )
import Ouroboros.Consensus.Storage.LeiosDB.API (Promise (..))
import Ouroboros.Consensus.Storage.LeiosDB.Exception
  ( LeiosDbException (..)
  , throwLeiosDbException
  )
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Connection
  ( Conn (..)
  , closeChecked
  , openRawConnection
  , withReadOnlyConn
  )
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Primitives
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Queries
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.Statements
import Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.WriteQueue
  ( WriteJob (..)
  , WriteQueue
  , submitJob
  )
import Ouroboros.Consensus.Storage.LeiosDB.Trace (LeiosDbStats (..), TraceLeiosDb (..))
import Ouroboros.Consensus.Util.IOLike (atomically)
import System.Directory (doesFileExist, getFileSize)

-- * Copying EBs to the immutable partition

-- | Implements 'leiosDbPromoteToImmutable':
--   - synchronously pin the EB rows in the volatile partition for promotion (@status@ 0 -> 1),
--     through the writer ('PinEb');
--   - ring the copier's doorbell.
sqlPromoteToImmutable :: WriteQueue -> StrictTVar IO Bool -> [LeiosPoint] -> IO ()
sqlPromoteToImmutable :: WriteQueue -> StrictTVar IO Bool -> [LeiosPoint] -> IO ()
sqlPromoteToImmutable WriteQueue
writeQueue StrictTVar IO Bool
copierDoorbell [LeiosPoint]
points = Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless ([LeiosPoint] -> Bool
forall a. [a] -> Bool
forall (t :: * -> *) a. Foldable t => t a -> Bool
null [LeiosPoint]
points) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
  Promise IO () -> IO ()
forall (m :: * -> *) a. Promise m a -> m a
await (Promise IO () -> IO ()) -> IO (Promise IO ()) -> IO ()
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< WriteQueue -> (WriteResult () -> WriteJob) -> IO (Promise IO ())
forall a.
HasCallStack =>
WriteQueue -> (WriteResult a -> WriteJob) -> IO (Promise IO a)
submitJob WriteQueue
writeQueue ([EbHash] -> WriteResult () -> WriteJob
PinEb [LeiosPoint -> EbHash
pointEbHash LeiosPoint
p | LeiosPoint
p <- [LeiosPoint]
points])
  -- notify the copier thread that there's work to be done
  STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ StrictTVar IO Bool -> Bool -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> a -> STM m ()
writeTVar StrictTVar IO Bool
copierDoorbell Bool
True

-- | The copier's connection and its prepared statements.
data CopierConn = CopierConn
  { CopierConn -> Database
ccDb :: !DB.Database
  -- ^ main = immutable partition, @vol@ = attached volatile partition
  , CopierConn -> CopierStmts
ccStmts :: !CopierStmts
  -- ^ Precompiled statements used by the copier.
  }

-- | Copy one pinned EB's closure into the immutable partition. 'True' when
-- it landed, so the caller may have its volatile rows marked as copied.
copyEbToImmutable ::
  Tracer IO TraceLeiosDb ->
  StrictTVar IO LeiosDbStats ->
  CopierConn ->
  EbHash ->
  IO Bool
copyEbToImmutable :: Tracer IO TraceLeiosDb
-> StrictTVar IO LeiosDbStats -> CopierConn -> EbHash -> IO Bool
copyEbToImmutable Tracer IO TraceLeiosDb
tracer StrictTVar IO LeiosDbStats
statsVar CopierConn
conn EbHash
ebHash =
  IO (Maybe Int)
appendToImmutable IO (Maybe Int) -> (Maybe Int -> IO Bool) -> IO Bool
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
    Maybe Int
Nothing -> do
      -- EB is not ready to be copied
      Tracer IO TraceLeiosDb -> TraceLeiosDb -> IO ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer IO TraceLeiosDb
tracer (TraceLeiosDb -> IO ()) -> TraceLeiosDb -> IO ()
forall a b. (a -> b) -> a -> b
$
        [Char] -> [Char] -> TraceLeiosDb
TraceLeiosDbCopyError
          (EbHash -> [Char]
forall a. Show a => a -> [Char]
show EbHash
ebHash)
          [Char]
"pinned EB has no complete closure in the volatile partition"
      Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
False
    Just Int
_copiedTxs -> do
      -- successfully copied, bump the stats
      StrictTVar IO LeiosDbStats -> Int -> IO ()
bumpImmutableStats StrictTVar IO LeiosDbStats
statsVar Int
1
      Tracer IO TraceLeiosDb -> TraceLeiosDb -> IO ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer IO TraceLeiosDb
tracer (TraceLeiosDb -> IO ()) -> TraceLeiosDb -> IO ()
forall a b. (a -> b) -> a -> b
$ Int -> TraceLeiosDb
TraceLeiosDbCopiedToImmutable Int
1
      Bool -> IO Bool
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
True
 where
  CopierConn{Database
ccDb :: CopierConn -> Database
ccDb :: Database
ccDb, CopierStmts
ccStmts :: CopierConn -> CopierStmts
ccStmts :: CopierStmts
ccStmts} = CopierConn
conn
  CopierStmts{Statement
ccCompleteness :: Statement
ccCompleteness :: CopierStmts -> Statement
ccCompleteness, Statement
ccInsertEb :: Statement
ccInsertEb :: CopierStmts -> Statement
ccInsertEb, Statement
ccInsertEbTxs :: Statement
ccInsertEbTxs :: CopierStmts -> Statement
ccInsertEbTxs, Statement
ccInsertTxs :: Statement
ccInsertTxs :: CopierStmts -> Statement
ccInsertTxs} = CopierStmts
ccStmts

  -- Attempt to do the actual copying.
  --
  -- Returns the number of copied body rows, or Nothing if
  -- the volatile partition does not hold the full closure.
  appendToImmutable :: IO (Maybe Int)
  appendToImmutable :: IO (Maybe Int)
appendToImmutable =
    Database -> IO (Maybe Int) -> IO (Maybe Int)
forall a. HasCallStack => Database -> IO a -> IO a
dbWithTransaction Database
ccDb (IO (Maybe Int) -> IO (Maybe Int))
-> IO (Maybe Int) -> IO (Maybe Int)
forall a b. (a -> b) -> a -> b
$ do
      -- the cert-RB is on our chain, hence the EB must be complete by this point.
      -- Still check if it is, as a defensive programming measure, as it's very cheap.
      (bodyCount, closureCount) <-
        Statement -> IO (Int64, Int64) -> IO (Int64, Int64)
forall a. Statement -> IO a -> IO a
useStmt Statement
ccCompleteness (IO (Int64, Int64) -> IO (Int64, Int64))
-> IO (Int64, Int64) -> IO (Int64, Int64)
forall a b. (a -> b) -> a -> b
$ do
          HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
ccCompleteness ParamIndex
1 (EbHash -> ByteString
ebHashBytes EbHash
ebHash)
          -- step the first time, expecting a single result row
          HasCallStack => Statement -> IO StepResult
Statement -> IO StepResult
dbStepSafe Statement
ccCompleteness IO StepResult
-> (StepResult -> IO (Int64, Int64)) -> IO (Int64, Int64)
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
            StepResult
DB.Done ->
              -- no row: critical error, fail fast and loud.
              -- this should not happen.
              [Char] -> IO (Int64, Int64)
forall (m :: * -> *) a.
(HasCallStack, MonadThrow m) =>
[Char] -> m a
throwLeiosDbException [Char]
"sql_copy_completeness: expected a row"
            StepResult
DB.Row -> do
              n <- Statement -> ColumnIndex -> IO Int64
DB.columnInt64 Statement
ccCompleteness ColumnIndex
0
              m <- DB.columnInt64 ccCompleteness 1
              -- step again, expecting no more row
              dbStepSafe ccCompleteness >>= \case
                StepResult
DB.Done ->
                  -- we have our result
                  (Int64, Int64) -> IO (Int64, Int64)
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Int64
n, Int64
m)
                StepResult
DB.Row ->
                  -- another row: critical error, fail fast and loud.
                  -- this should not happen.
                  [Char] -> IO (Int64, Int64)
forall (m :: * -> *) a.
(HasCallStack, MonadThrow m) =>
[Char] -> m a
throwLeiosDbException [Char]
"sql_copy_completeness: expected exactly one row"
      if bodyCount == 0 || bodyCount /= closureCount
        then
          -- the EB is incomplete, don't copy
          pure Nothing
        else do
          -- the EB is complete: body is non-empty and bodyCounty matches closureCount
          --
          -- copy the EB
          useStmt ccInsertEb $ do
            dbBindBlob ccInsertEb 1 (ebHashBytes ebHash)
            dbStep1Safe ccInsertEb
          -- copy the eb-to-transactions mapping
          useStmt ccInsertEbTxs $ do
            dbBindBlob ccInsertEbTxs 1 (ebHashBytes ebHash)
            dbStep1Safe ccInsertEbTxs
          nTxs <- DB.changes ccDb
          -- copy the transactions
          useStmt ccInsertTxs $ do
            dbBindBlob ccInsertTxs 1 (ebHashBytes ebHash)
            dbStep1Safe ccInsertTxs
          pure (Just nTxs)

-- | The copier: the only writer into the immutable partition.
--
-- Returns the copier's thread: cancelling it closes its connection.
startCopier ::
  ResourceRegistry IO ->
  Tracer IO TraceLeiosDb ->
  StrictTVar IO LeiosDbStats ->
  StrictTVar IO Bool ->
  WriteQueue ->
  FilePath ->
  FilePath ->
  IO (Thread IO ())
startCopier :: ResourceRegistry IO
-> Tracer IO TraceLeiosDb
-> StrictTVar IO LeiosDbStats
-> StrictTVar IO Bool
-> WriteQueue
-> [Char]
-> [Char]
-> IO (Thread IO ())
startCopier ResourceRegistry IO
registry Tracer IO TraceLeiosDb
tracer StrictTVar IO LeiosDbStats
statsVar StrictTVar IO Bool
copierDoorbell WriteQueue
writeQueue [Char]
volPath [Char]
immPath =
  ResourceRegistry IO -> [Char] -> IO () -> IO (Thread IO ())
forall (m :: * -> *) a.
(MonadAsync m, MonadFork m, MonadMask m, MonadEvaluate m,
 HasCallStack) =>
ResourceRegistry m -> [Char] -> m a -> m (Thread m a)
forkLinkedThread ResourceRegistry IO
registry [Char]
"leiosdb-copier" (IO () -> IO (Thread IO ())) -> IO () -> IO (Thread IO ())
forall a b. (a -> b) -> a -> b
$
    IO Database -> (Database -> IO ()) -> (Database -> IO ()) -> IO ()
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (HasCallStack => [Char] -> IO Database
[Char] -> IO Database
openRawConnection [Char]
immPath) HasCallStack => Database -> IO ()
Database -> IO ()
closeChecked ((Database -> IO ()) -> IO ()) -> (Database -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Database
db -> do
      Database -> [Char] -> (Statement -> IO ()) -> IO ()
forall a.
HasCallStack =>
Database -> [Char] -> (Statement -> IO a) -> IO a
withStmt Database
db [Char]
"ATTACH ? AS vol" ((Statement -> IO ()) -> IO ()) -> (Statement -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Statement
stmt -> do
        HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindUtf8 Statement
stmt ParamIndex
1 ([Char] -> ByteString
forall a. IsString a => [Char] -> a
fromString [Char]
volPath)
        HasCallStack => Statement -> IO ()
Statement -> IO ()
dbStep1Safe Statement
stmt
      IO CopierStmts
-> (CopierStmts -> IO ()) -> (CopierStmts -> IO ()) -> IO ()
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (HasCallStack => Database -> IO CopierStmts
Database -> IO CopierStmts
prepareCopierStmts Database
db) CopierStmts -> IO ()
finalizeCopierStmts ((CopierStmts -> IO ()) -> IO ())
-> (CopierStmts -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \CopierStmts
stmts ->
        IO Statement
-> (Statement -> IO ()) -> (Statement -> IO ()) -> IO ()
forall a b c. IO a -> (a -> IO b) -> (a -> IO c) -> IO c
forall (m :: * -> *) a b c.
MonadThrow m =>
m a -> (a -> m b) -> (a -> m c) -> m c
bracket (HasCallStack => Database -> Utf8 -> IO Statement
Database -> Utf8 -> IO Statement
dbPrepare Database
db ([Char] -> Utf8
forall a. IsString a => [Char] -> a
fromString [Char]
sql_next_pinned_eb)) Statement -> IO ()
dbFinalize ((Statement -> IO ()) -> IO ()) -> (Statement -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$
          CopierConn -> Statement -> IO ()
runCopier (Database -> CopierStmts -> CopierConn
CopierConn Database
db CopierStmts
stmts)
 where
  runCopier ::
    CopierConn ->
    -- 'sql_next_pinned_eb'
    DB.Statement ->
    IO ()
  runCopier :: CopierConn -> Statement -> IO ()
runCopier CopierConn
copierConn Statement
nextPinnedStmt = IO ()
forall {b}. IO b
loop
   where
    CopierConn{Database
ccDb :: CopierConn -> Database
ccDb :: Database
ccDb} = CopierConn
copierConn

    nextPinnedBatch :: IO [EbHash]
nextPinnedBatch = do
      HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
nextPinnedStmt ParamIndex
1 (Int -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral Int
copyBatchSize)
      Statement -> IO [EbHash] -> IO [EbHash]
forall a. Statement -> IO a -> IO a
useStmt Statement
nextPinnedStmt (IO [EbHash] -> IO [EbHash]) -> IO [EbHash] -> IO [EbHash]
forall a b. (a -> b) -> a -> b
$
        let rows :: [EbHash] -> IO [EbHash]
rows [EbHash]
acc =
              HasCallStack => Statement -> IO StepResult
Statement -> IO StepResult
dbStepSafe Statement
nextPinnedStmt IO StepResult -> (StepResult -> IO [EbHash]) -> IO [EbHash]
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
                StepResult
DB.Done -> [EbHash] -> IO [EbHash]
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ([EbHash] -> [EbHash]
forall a. [a] -> [a]
reverse [EbHash]
acc)
                StepResult
DB.Row -> do
                  h <- ByteString -> EbHash
MkEbHash (ByteString -> EbHash) -> IO ByteString -> IO EbHash
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Statement -> ColumnIndex -> IO ByteString
DB.columnBlob Statement
nextPinnedStmt ColumnIndex
0
                  rows (h : acc)
         in [EbHash] -> IO [EbHash]
rows []

    -- 'True' when the EB is in the immutable partition and its pin can be
    -- retired. The EB stays pinned either way, so it is still next to copy.
    -- Backing off is what keeps a closure that never completes -- or a
    -- partition that keeps refusing the write -- from spinning here.
    copyOne :: EbHash -> IO Bool
copyOne EbHash
ebHash =
      Tracer IO TraceLeiosDb
-> StrictTVar IO LeiosDbStats -> CopierConn -> EbHash -> IO Bool
copyEbToImmutable Tracer IO TraceLeiosDb
tracer StrictTVar IO LeiosDbStats
statsVar CopierConn
copierConn EbHash
ebHash
        IO Bool -> (LeiosDbException -> IO Bool) -> IO Bool
forall e a. Exception e => IO a -> (e -> IO a) -> IO a
forall (m :: * -> *) e a.
(MonadCatch m, Exception e) =>
m a -> (e -> m a) -> m a
`catch` \(LeiosDbException
e :: LeiosDbException) -> do
          _ <- Database -> Utf8 -> IO (Either (Error, Utf8) ())
DB.exec Database
ccDb Utf8
"ROLLBACK"
          traceWith tracer $ TraceLeiosDbCopyError (show ebHash) (displayException e)
          pure False

    -- One 'MarkCopied' for the batch: the mark is what makes a copied EB
    -- evictable, and under sync a queue round-trip per EB queues behind
    -- saturated inserts.
    copyBatch :: [EbHash] -> IO ()
copyBatch [EbHash]
ebHashes = do
      copied <- (EbHash -> IO Bool) -> [EbHash] -> IO [EbHash]
forall (m :: * -> *) a.
Applicative m =>
(a -> m Bool) -> [a] -> m [a]
filterM EbHash -> IO Bool
copyOne [EbHash]
ebHashes
      if null copied
        then threadDelay copyRetryMicros
        else void . await =<< submitJob writeQueue (MarkCopied copied)

    loop :: IO b
loop = do
      -- Clear the copier doorbell before looking if there's any copying work to do,
      -- so a pin that lands while we look rings again instead of being lost.
      STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ StrictTVar IO Bool -> Bool -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> a -> STM m ()
writeTVar StrictTVar IO Bool
copierDoorbell Bool
False
      IO [EbHash]
nextPinnedBatch IO [EbHash] -> ([EbHash] -> IO b) -> IO b
forall a b. IO a -> (a -> IO b) -> IO b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
        batch :: [EbHash]
batch@(EbHash
_ : [EbHash]
_) -> [EbHash] -> IO ()
copyBatch [EbHash]
batch IO () -> IO b -> IO b
forall a b. IO a -> IO b -> IO b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> IO b
loop
        [] -> do
          STM () -> IO ()
forall a. STM a -> IO a
IO.atomically (STM () -> IO ()) -> STM () -> IO ()
forall a b. (a -> b) -> a -> b
$ StrictTVar IO Bool -> STM IO Bool
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar IO Bool
copierDoorbell STM Bool -> (Bool -> STM ()) -> STM ()
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Bool -> STM ()
Bool -> STM IO ()
forall (m :: * -> *). MonadSTM m => Bool -> STM m ()
check
          IO b
loop

-- | How long the copier waits before trying a pinned EB again.
copyRetryMicros :: Int
copyRetryMicros :: Int
copyRetryMicros = Int
1000000

-- | How many pinned EBs the copier takes per pass, and so per 'MarkCopied'.
copyBatchSize :: Int
copyBatchSize :: Int
copyBatchSize = Int
32

-- * Garbage collection of the volatile partition

-- | How many EBs one sweep transaction takes, by default.
defaultGcBatchSize :: Int64
defaultGcBatchSize :: Int64
defaultGcBatchSize = Int64
4

-- | How many orphaned txs to GC in one sweep.
gcOrphanTxBatchSize :: Int64
gcOrphanTxBatchSize :: Int64
gcOrphanTxBatchSize = Int64
1024

-- | Page size of the 'gcReinit' scan.
gcCandidatesPageSize :: Int64
gcCandidatesPageSize :: Int64
gcCandidatesPageSize = Int64
4096

-- | Implements 'leiosDbGarbageCollect': the MARK phase of GC mark-and-sweep,
-- as a 'GcMark' job on the writer (see 'gcMark').
sqlGarbageCollect :: WriteQueue -> SlotNo -> IO ()
sqlGarbageCollect :: WriteQueue -> SlotNo -> IO ()
sqlGarbageCollect WriteQueue
writeQueue SlotNo
gcSlot =
  Promise IO () -> IO ()
forall (m :: * -> *) a. Promise m a -> m a
await (Promise IO () -> IO ()) -> IO (Promise IO ()) -> IO ()
forall (m :: * -> *) a b. Monad m => (a -> m b) -> m a -> m b
=<< WriteQueue -> (WriteResult () -> WriteJob) -> IO (Promise IO ())
forall a.
HasCallStack =>
WriteQueue -> (WriteResult a -> WriteJob) -> IO (Promise IO a)
submitJob WriteQueue
writeQueue (SlotNo -> WriteResult () -> WriteJob
GcMark SlotNo
gcSlot)

-- | The MARK phase of GC mark-and-sweep:
--   - mark for GC (@status = 3@) every EB hash all of whose announcements are older
--     than the given slot and not pinned (@status = 1@);
--   - stage its txs as GC candidates;
--   - ask for a sweep.
--
-- Runs on the writer. It does not do much work, but rather primes the state
-- for the writer's own sweep steps (see 'startWriter').
gcMark ::
  HasCallStack =>
  StrictTVar IO Bool ->
  DB.Database ->
  GcStmts ->
  SlotNo ->
  IO ()
gcMark :: HasCallStack =>
StrictTVar IO Bool -> Database -> GcStmts -> SlotNo -> IO ()
gcMark StrictTVar IO Bool
sweepDoorbell Database
db GcStmts
gcStmts SlotNo
gcSlot = do
  let GcStmts{Statement
gsHasWork :: Statement
gsHasWork :: GcStmts -> Statement
gsHasWork, Statement
gsAddGcCandidatesTxs :: Statement
gsAddGcCandidatesTxs :: GcStmts -> Statement
gsAddGcCandidatesTxs, Statement
gsMarkEbForGC :: Statement
gsMarkEbForGC :: GcStmts -> Statement
gsMarkEbForGC} = GcStmts
gcStmts
  -- check if GC has any work to do
  hasWork <-
    Statement -> IO Bool -> IO Bool
forall a. Statement -> IO a -> IO a
useStmt Statement
gsHasWork (IO Bool -> IO Bool) -> IO Bool -> IO Bool
forall a b. (a -> b) -> a -> b
$ do
      HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
gsHasWork ParamIndex
1 Int64
slot
      (Int64 -> Int64 -> Bool
forall a. Eq a => a -> a -> Bool
/= Int64
0) (Int64 -> Bool) -> IO Int64 -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> HasCallStack => Statement -> IO Int64
Statement -> IO Int64
readSingleInt64 Statement
gsHasWork
  when hasWork $ do
    (nTxsStagedAsGCCandidates, nEbsMarked) <-
      dbWithWriteTransactionRaw db $ do
        -- transactions must be marked for GC before their EBs,
        -- due to the way the sql statements are written.
        -- mark transactions b for GC
        useStmt gsAddGcCandidatesTxs $ do
          dbBindInt64 gsAddGcCandidatesTxs 1 slot
          dbStep1Safe gsAddGcCandidatesTxs
        nTxsStagedAsGCCandidates <- DB.changes db
        -- now mark the EB
        useStmt gsMarkEbForGC $ do
          dbBindInt64 gsMarkEbForGC 1 slot
          dbStep1Safe gsMarkEbForGC
        nEbsMarked <- DB.changes db
        pure (nTxsStagedAsGCCandidates, nEbsMarked)
    when (nTxsStagedAsGCCandidates > 0 || nEbsMarked > 0) $
      atomically $
        writeTVar sweepDoorbell True
 where
  slot :: Int64
slot = Word64 -> Int64
forall a b. (Integral a, Num b) => a -> b
fromIntegral (SlotNo -> Word64
unSlotNo SlotNo
gcSlot)

-- * Sweeping GC-marked rows out of the volatile partition

-- | The sweeper's connection and its prepared statements.
data SweeperConn = SweeperConn
  { SweeperConn -> Database
swDb :: !DB.Database
  -- ^ The writer's connection to the volatile partition.
  , SweeperConn -> SweeperStmts
swStmts :: !SweeperStmts
  -- ^ Precompiled statements used by the sweeper.
  }

-- | One 'SweepEbBatch' transaction: evict up to the given number of GC-marked
-- EBs. Runs on the writer.
sweepEbBatch :: SweeperConn -> Int64 -> IO Int
sweepEbBatch :: SweeperConn -> Int64 -> IO Int
sweepEbBatch SweeperConn
conn Int64
batchSize = do
  let SweeperConn{Database
swDb :: SweeperConn -> Database
swDb :: Database
swDb, SweeperStmts
swStmts :: SweeperConn -> SweeperStmts
swStmts :: SweeperStmts
swStmts} = SweeperConn
conn
      SweeperStmts{Statement
swPickMarked :: Statement
swPickMarked :: SweeperStmts -> Statement
swPickMarked, Statement
swEvictEbTxs :: Statement
swEvictEbTxs :: SweeperStmts -> Statement
swEvictEbTxs, Statement
swEvictMissingTxs :: Statement
swEvictMissingTxs :: SweeperStmts -> Statement
swEvictMissingTxs, Statement
swEvictEbs :: Statement
swEvictEbs :: SweeperStmts -> Statement
swEvictEbs} = SweeperStmts
swStmts
  Database -> IO Int -> IO Int
forall a. HasCallStack => Database -> IO a -> IO a
dbWithWriteTransactionRaw Database
swDb (IO Int -> IO Int) -> IO Int -> IO Int
forall a b. (a -> b) -> a -> b
$ do
    -- check if any EBs are ready to be evicted
    evictableEbs <- Statement -> IO [ByteString] -> IO [ByteString]
forall a. Statement -> IO a -> IO a
useStmt Statement
swPickMarked (IO [ByteString] -> IO [ByteString])
-> IO [ByteString] -> IO [ByteString]
forall a b. (a -> b) -> a -> b
$ do
      -- a negative LIMIT means no limit in SQLite
      HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
swPickMarked ParamIndex
1 (if Int64
batchSize Int64 -> Int64 -> Bool
forall a. Ord a => a -> a -> Bool
<= Int64
0 then -Int64
1 else Int64
batchSize)
      HasCallStack => Statement -> IO [ByteString]
Statement -> IO [ByteString]
collectBlobs Statement
swPickMarked
    -- evict EBs if any are ready to be GCed
    if null evictableEbs
      then pure 0
      else do
        let evictableEbsJson = [ByteString] -> ByteString
jsonHexArray [ByteString]
evictableEbs
        execJson swEvictEbTxs evictableEbsJson
        execJson swEvictMissingTxs evictableEbsJson
        execJson swEvictEbs evictableEbsJson
        DB.changes swDb

-- | One 'SweepOrphanBatch' transaction: evict up to the given number of
-- orphaned txs, or 'Nothing' if there was nothing to do. Runs on the writer.
sweepOrphanBatch :: SweeperConn -> Int64 -> IO (Maybe Int)
sweepOrphanBatch :: SweeperConn -> Int64 -> IO (Maybe Int)
sweepOrphanBatch SweeperConn
conn Int64
batchSize = do
  let SweeperConn{Database
swDb :: SweeperConn -> Database
swDb :: Database
swDb, SweeperStmts
swStmts :: SweeperConn -> SweeperStmts
swStmts :: SweeperStmts
swStmts} = SweeperConn
conn
      SweeperStmts{Statement
swAnyMarked :: Statement
swAnyMarked :: SweeperStmts -> Statement
swAnyMarked, Statement
swPickOrphans :: Statement
swPickOrphans :: SweeperStmts -> Statement
swPickOrphans, Statement
swOrphanTxs :: Statement
swOrphanTxs :: SweeperStmts -> Statement
swOrphanTxs, Statement
swPopOrphans :: Statement
swPopOrphans :: SweeperStmts -> Statement
swPopOrphans} = SweeperStmts
swStmts
  Database -> IO (Maybe Int) -> IO (Maybe Int)
forall a. HasCallStack => Database -> IO a -> IO a
dbWithWriteTransactionRaw Database
swDb (IO (Maybe Int) -> IO (Maybe Int))
-> IO (Maybe Int) -> IO (Maybe Int)
forall a b. (a -> b) -> a -> b
$ do
    -- don't run the sweep if any GC-marked EBs remain
    blocked <- Statement -> IO Bool -> IO Bool
forall a. Statement -> IO a -> IO a
useStmt Statement
swAnyMarked (IO Bool -> IO Bool) -> IO Bool -> IO Bool
forall a b. (a -> b) -> a -> b
$ (Int64 -> Int64 -> Bool
forall a. Eq a => a -> a -> Bool
/= Int64
0) (Int64 -> Bool) -> IO Int64 -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> HasCallStack => Statement -> IO Int64
Statement -> IO Int64
readSingleInt64 Statement
swAnyMarked
    if blocked
      then pure Nothing
      else do
        -- look for txs to GC
        orphanedTxs <- useStmt swPickOrphans $ do
          dbBindInt64 swPickOrphans 1 batchSize
          collectBlobs swPickOrphans
        if null orphanedTxs
          then pure Nothing
          else do
            let orphanedTxsJson = [ByteString] -> ByteString
jsonHexArray [ByteString]
orphanedTxs
            -- evict transactions
            execJson swOrphanTxs orphanedTxsJson
            nTxs <- DB.changes swDb
            -- and delete them from the GC transaction candidates table
            execJson swPopOrphans orphanedTxsJson
            pure (Just nTxs)

-- | Stage every unstaged GC candidate, one page per transaction.
--
-- The sweeper only ever reads 'gcTxCandidates', which the mark phase fills
-- ('sql_gc_stage_marked'). A tx orphaned by anything else -- a
-- 'truncateLeiosDbAfterSlot' that dropped its EB's rows, a database written
-- before the table existed -- is referenced by nothing and staged nowhere,
-- and would never be collected.
--
-- Only such out-of-band edits can leave that behind, and only before the
-- writer started, so this runs once per process, on the writer.
gcReinit :: SweeperConn -> IO ()
gcReinit :: SweeperConn -> IO ()
gcReinit SweeperConn
conn = do
  let SweeperConn{Database
swDb :: SweeperConn -> Database
swDb :: Database
swDb, SweeperStmts
swStmts :: SweeperConn -> SweeperStmts
swStmts :: SweeperStmts
swStmts} = SweeperConn
conn
      SweeperStmts{Statement
swHasUnstagedGcCandidates :: Statement
swHasUnstagedGcCandidates :: SweeperStmts -> Statement
swHasUnstagedGcCandidates, Statement
swUnstagedGcCandidatesPage :: Statement
swUnstagedGcCandidatesPage :: SweeperStmts -> Statement
swUnstagedGcCandidatesPage, Statement
swInsertGcCandidates :: Statement
swInsertGcCandidates :: SweeperStmts -> Statement
swInsertGcCandidates} = SweeperStmts
swStmts
  anyUnstaged <-
    Statement -> IO Bool -> IO Bool
forall a. Statement -> IO a -> IO a
useStmt Statement
swHasUnstagedGcCandidates (IO Bool -> IO Bool) -> IO Bool -> IO Bool
forall a b. (a -> b) -> a -> b
$
      (Int64 -> Int64 -> Bool
forall a. Eq a => a -> a -> Bool
/= Int64
0) (Int64 -> Bool) -> IO Int64 -> IO Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> HasCallStack => Statement -> IO Int64
Statement -> IO Int64
readSingleInt64 Statement
swHasUnstagedGcCandidates
  let pageLoop ByteString
cursor = do
        page <- Statement -> IO [ByteString] -> IO [ByteString]
forall a. Statement -> IO a -> IO a
useStmt Statement
swUnstagedGcCandidatesPage (IO [ByteString] -> IO [ByteString])
-> IO [ByteString] -> IO [ByteString]
forall a b. (a -> b) -> a -> b
$ do
          HasCallStack => Statement -> ParamIndex -> ByteString -> IO ()
Statement -> ParamIndex -> ByteString -> IO ()
dbBindBlob Statement
swUnstagedGcCandidatesPage ParamIndex
1 ByteString
cursor
          HasCallStack => Statement -> ParamIndex -> Int64 -> IO ()
Statement -> ParamIndex -> Int64 -> IO ()
dbBindInt64 Statement
swUnstagedGcCandidatesPage ParamIndex
2 Int64
gcCandidatesPageSize
          HasCallStack => Statement -> IO [ByteString]
Statement -> IO [ByteString]
collectBlobs Statement
swUnstagedGcCandidatesPage
        unless (null page) $ do
          dbWithWriteTransactionRaw swDb $
            execJson swInsertGcCandidates (jsonHexArray page)
          when (length page == fromIntegral gcCandidatesPageSize) $
            pageLoop (last page)
  when anyUnstaged $ pageLoop BS.empty

-- | How far a sweep pass has got. The writer advances it one batch per turn
-- rather than running a pass to completion, so an insert waits for a batch at
-- most.
data SweepState
  = SweepIdle
  | -- | Evicting GC-marked EBs.
    SweepEbs
      -- | EBs evicted so far
      !Int
  | -- | Evicting the txs they orphaned.
    SweepOrphans
      -- | EBs evicted in the EB phase
      !Int
      -- | txs evicted so far
      !Int
  deriving SweepState -> SweepState -> Bool
(SweepState -> SweepState -> Bool)
-> (SweepState -> SweepState -> Bool) -> Eq SweepState
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: SweepState -> SweepState -> Bool
== :: SweepState -> SweepState -> Bool
$c/= :: SweepState -> SweepState -> Bool
/= :: SweepState -> SweepState -> Bool
Eq

-- * Stats

-- | Initialise 'LeiosDbStats' by counting the EB rows of both partitions.
--   This will only run once per process.
initialStats :: HasCallStack => FilePath -> FilePath -> IO LeiosDbStats
initialStats :: HasCallStack => [Char] -> [Char] -> IO LeiosDbStats
initialStats [Char]
volPath [Char]
immPath = do
  vol <- [Char] -> IO Int
forall {b}. Num b => [Char] -> IO b
countEbs [Char]
volPath
  imm <- countEbs immPath
  pure
    LeiosDbStats
      { volatileEbs = vol
      , immutableEbs = imm
      , walBytes = 0
      }
 where
  countEbs :: [Char] -> IO b
countEbs [Char]
path =
    Int64 -> b
forall a b. (Integral a, Num b) => a -> b
fromIntegral (Int64 -> b) -> IO Int64 -> IO b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [Char] -> (Database -> IO Int64) -> IO Int64
forall a. HasCallStack => [Char] -> (Database -> IO a) -> IO a
withReadOnlyConn [Char]
path (\Database
db -> HasCallStack => Database -> [Char] -> IO Int64
Database -> [Char] -> IO Int64
queryInt64 Database
db [Char]
"SELECT COUNT(*) FROM ebs")

-- * Stats sampling

-- | Fork a thread that traces 'TraceLeiosDbStats' every 10 seconds. Samples
-- the volatile partition's file only.
startVolatileStatsSampler ::
  ResourceRegistry IO ->
  Tracer IO TraceLeiosDb ->
  StrictTVar IO LeiosDbStats ->
  FilePath ->
  IO (Thread IO ())
startVolatileStatsSampler :: ResourceRegistry IO
-> Tracer IO TraceLeiosDb
-> StrictTVar IO LeiosDbStats
-> [Char]
-> IO (Thread IO ())
startVolatileStatsSampler ResourceRegistry IO
registry Tracer IO TraceLeiosDb
tracer StrictTVar IO LeiosDbStats
statsVar [Char]
volPath =
  ResourceRegistry IO -> [Char] -> IO () -> IO (Thread IO ())
forall (m :: * -> *) a.
(MonadAsync m, MonadFork m, MonadMask m, MonadEvaluate m,
 HasCallStack) =>
ResourceRegistry m -> [Char] -> m a -> m (Thread m a)
forkLinkedThread ResourceRegistry IO
registry [Char]
"leiosdb-stats-sampler" (IO () -> IO (Thread IO ())) -> IO () -> IO (Thread IO ())
forall a b. (a -> b) -> a -> b
$ IO () -> IO ()
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
    -- wait one sample window to side-step contention with starting the LeiosDB
    Int -> IO ()
threadDelay Int
tenSeconds
    stats <- StrictTVar IO LeiosDbStats -> IO LeiosDbStats
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> m a
readTVarIO StrictTVar IO LeiosDbStats
statsVar
    -- read the WAL size
    walBytes <- fileSizeOr0 (volPath <> "-wal")
    traceWith tracer $
      TraceLeiosDbStats
        stats
          { walBytes
          }
 where
  tenSeconds :: Int
tenSeconds = Int
10000000

  fileSizeOr0 :: FilePath -> IO Integer
  fileSizeOr0 :: [Char] -> IO Integer
fileSizeOr0 [Char]
path = do
    exists <- [Char] -> IO Bool
doesFileExist [Char]
path
    if exists then getFileSize path else pure 0

-- | Fold a delta into the volatile EB count of the in-memory 'LeiosDbStats'.
bumpVolatileStats :: Conn -> Int -> IO ()
bumpVolatileStats :: Conn -> Int -> IO ()
bumpVolatileStats Conn{StrictTVar IO LeiosDbStats
connStats :: StrictTVar IO LeiosDbStats
connStats :: Conn -> StrictTVar IO LeiosDbStats
connStats} = StrictTVar IO LeiosDbStats -> Int -> IO ()
bumpVolatileStatsVar StrictTVar IO LeiosDbStats
connStats

-- | 'bumpVolatileStats' for the maintenance paths, which have no 'Conn'.
bumpVolatileStatsVar :: StrictTVar IO LeiosDbStats -> Int -> IO ()
bumpVolatileStatsVar :: StrictTVar IO LeiosDbStats -> Int -> IO ()
bumpVolatileStatsVar StrictTVar IO LeiosDbStats
statsVar Int
dEbs =
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (Int
dEbs Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
    STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
      StrictTVar IO LeiosDbStats
-> (LeiosDbStats -> LeiosDbStats) -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar IO LeiosDbStats
statsVar ((LeiosDbStats -> LeiosDbStats) -> STM IO ())
-> (LeiosDbStats -> LeiosDbStats) -> STM IO ()
forall a b. (a -> b) -> a -> b
$
        \LeiosDbStats
s -> LeiosDbStats
s{volatileEbs = volatileEbs s + dEbs}

-- | Fold a delta into the immutable EB count of the in-memory 'LeiosDbStats'.
bumpImmutableStats :: StrictTVar IO LeiosDbStats -> Int -> IO ()
bumpImmutableStats :: StrictTVar IO LeiosDbStats -> Int -> IO ()
bumpImmutableStats StrictTVar IO LeiosDbStats
statsVar Int
dEbs =
  Bool -> IO () -> IO ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (Int
dEbs Int -> Int -> Bool
forall a. Eq a => a -> a -> Bool
== Int
0) (IO () -> IO ()) -> IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
    STM IO () -> IO ()
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM IO () -> IO ()) -> STM IO () -> IO ()
forall a b. (a -> b) -> a -> b
$
      StrictTVar IO LeiosDbStats
-> (LeiosDbStats -> LeiosDbStats) -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar IO LeiosDbStats
statsVar ((LeiosDbStats -> LeiosDbStats) -> STM IO ())
-> (LeiosDbStats -> LeiosDbStats) -> STM IO ()
forall a b. (a -> b) -> a -> b
$
        \LeiosDbStats
s -> LeiosDbStats
s{immutableEbs = immutableEbs s + dEbs}