{-# LANGUAGE LambdaCase #-}
{-# LANGUAGE NamedFieldPuns #-}

-- | The submission side of the single writer: the jobs it serves and the
-- bounded queue that carries them. The worker that drains the queue is
-- 'Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.startWriter'.
module Ouroboros.Consensus.Storage.LeiosDB.Impl.SQLite.WriteQueue
  ( WriteJob (..)
  , WriteResult
  , WriteQueue (..)
  , submitJob
  , describeJob
  , failJob
  , maxJobsBetweenMaintenance
  , writerQueueDepth
  ) where

import Cardano.Slotting.Slot (SlotNo (..))
import Control.Concurrent.Class.MonadSTM.Strict
  ( StrictTBQueue
  , StrictTMVar
  , StrictTVar
  , isFullTBQueue
  , newEmptyTMVarIO
  , putTMVar
  , readTMVar
  , readTVar
  , writeTBQueue
  )
import Control.Exception (SomeException, throwIO)
import Control.Monad (join)
import Control.Monad.Class.MonadThrow (onException)
import Control.Tracer (Tracer, traceWith)
import Data.ByteString (ByteString)
import GHC.Stack (HasCallStack)
import qualified GHC.Stack
import Numeric.Natural (Natural)
import Ouroboros.Consensus.Leios.Types
  ( BytesSize
  , EbHash (..)
  , LeiosEb
  , LeiosPoint (..)
  , TxHash (..)
  , leiosEbTxs
  )
import Ouroboros.Consensus.Storage.LeiosDB.API (CompletedEbs, Promise (..))
import Ouroboros.Consensus.Storage.LeiosDB.Exception
  ( LeiosDbException (..)
  , LeiosDbWriteFailure (..)
  )
import Ouroboros.Consensus.Storage.LeiosDB.Trace (TraceLeiosDb (..))
import Ouroboros.Consensus.Util.IOLike (atomically)

-- * The single writer

-- | One queued write, carrying the variable its result lands in.
--
-- The insert jobs come from 'LeiosDbWriter', the rest from the handle's
-- promote\/GC entry points. One worker running them all is what makes the
-- single-writer property total: no two in-process connections ever contend
-- for a write lock on the volatile partition. Sweeping is not a job at all
-- -- the worker does it between jobs; see 'startWriter'.
data WriteJob
  = WriteEbPoint
      -- | the announced EB
      !LeiosPoint
      -- | its announced size
      !BytesSize
      -- | where the worker puts the result
      !(WriteResult ())
  | WriteEbBody
      -- | the EB's point
      !LeiosPoint
      -- | the EB body
      !LeiosEb
      -- | where the worker puts the result
      !(WriteResult CompletedEbs)
  | WriteTxs
      -- | txs with their bytes
      ![(TxHash, ByteString)]
      -- | where the worker puts the result
      !(WriteResult CompletedEbs)
  | -- | Does nothing; awaiting it after the queue's FIFO order means every
    -- write submitted before it has landed.
    Flush !(WriteResult ())
  | -- | Pin EBs for promotion; see 'sqlPromoteToImmutable'.
    PinEb
      -- | the EBs to pin
      ![EbHash]
      -- | where the worker puts the result
      !(WriteResult ())
  | -- | The volatile half of a copy: the EB is in the immutable partition
    -- now, so its volatile rows are evictable. Submitted by 'startCopier'.
    MarkCopied
      -- | the EBs now in the immutable partition
      ![EbHash]
      -- | where the worker puts the result
      !(WriteResult ())
  | -- | The GC MARK phase; see 'gcMark'.
    GcMark
      -- | the GC frontier slot
      !SlotNo
      -- | where the worker puts the result
      !(WriteResult ())
  | -- | Stop the worker: awaiting it after the queue's FIFO order means every
    -- write submitted before it has landed and the connections are closed.
    Shutdown !(WriteResult ())

type WriteResult a = StrictTMVar IO (Either SomeException a)

-- | The writer's submission side: the job queue, and -- once the worker has
-- stopped -- the exception every submission throws instead of queueing.
data WriteQueue = WriteQueue
  { WriteQueue -> StrictTBQueue IO WriteJob
wqJobs :: !(StrictTBQueue IO WriteJob)
  , WriteQueue -> StrictTVar IO (Maybe SomeException)
wqSealed :: !(StrictTVar IO (Maybe SomeException))
  , WriteQueue -> Tracer IO TraceLeiosDb
wqTracer :: !(Tracer IO TraceLeiosDb)
  }

-- | Hand a job to the writer; the 'Promise' waits for its result. Blocks
-- only while the queue is full, which is the backpressure -- and which is
-- traced, since it is the one thing about the writer that no other trace
-- reports: every producer is now waiting on it.
--
-- Checking the seal and queueing are one transaction, so after the worker
-- seals the queue no job can slip in unserved: submission throws the
-- worker's parting exception instead -- also waking any submitter that was
-- blocked on a full queue.
submitJob :: HasCallStack => WriteQueue -> (WriteResult a -> WriteJob) -> IO (Promise IO a)
submitJob :: forall a.
HasCallStack =>
WriteQueue -> (WriteResult a -> WriteJob) -> IO (Promise IO a)
submitJob WriteQueue{StrictTBQueue IO WriteJob
wqJobs :: WriteQueue -> StrictTBQueue IO WriteJob
wqJobs :: StrictTBQueue IO WriteJob
wqJobs, StrictTVar IO (Maybe SomeException)
wqSealed :: WriteQueue -> StrictTVar IO (Maybe SomeException)
wqSealed :: StrictTVar IO (Maybe SomeException)
wqSealed, Tracer IO TraceLeiosDb
wqTracer :: WriteQueue -> Tracer IO TraceLeiosDb
wqTracer :: Tracer IO TraceLeiosDb
wqTracer} WriteResult a -> WriteJob
mkJob = do
  resultVar <- IO (WriteResult a)
forall (m :: * -> *) a. MonadSTM m => m (StrictTMVar m a)
newEmptyTMVarIO
  let job = WriteResult a -> WriteJob
mkJob WriteResult a
resultVar
      -- Failures surface far from their submitter -- on the worker, or on
      -- whichever thread awaits -- so name the write and its submission site.
      wrap SomeException
cause =
        LeiosDbWriteFailure -> LeiosDbException
LeiosDbWriteException
          LeiosDbWriteFailure
            { ldwfWriteJob :: [Char]
ldwfWriteJob = WriteJob -> [Char]
describeJob WriteJob
job
            , ldwfSubmittedFrom :: [Char]
ldwfSubmittedFrom = CallStack -> [Char]
GHC.Stack.prettyCallStack CallStack
HasCallStack => CallStack
GHC.Stack.callStack
            , ldwfWriteFailure :: SomeException
ldwfWriteFailure = SomeException
cause
            }
      -- Take a slot if there is one; 'Left' once the queue is sealed.
      offer =
        StrictTVar IO (Maybe SomeException) -> STM IO (Maybe SomeException)
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar IO (Maybe SomeException)
wqSealed STM (Maybe SomeException)
-> (Maybe SomeException -> STM (Either SomeException Bool))
-> STM (Either SomeException Bool)
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
          Just SomeException
cause -> Either SomeException Bool -> STM (Either SomeException Bool)
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (SomeException -> Either SomeException Bool
forall a b. a -> Either a b
Left SomeException
cause)
          Maybe SomeException
Nothing ->
            StrictTBQueue IO WriteJob -> STM IO Bool
forall (m :: * -> *) a.
MonadSTM m =>
StrictTBQueue m a -> STM m Bool
isFullTBQueue StrictTBQueue IO WriteJob
wqJobs STM Bool
-> (Bool -> STM (Either SomeException Bool))
-> STM (Either SomeException Bool)
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
              Bool
True -> Either SomeException Bool -> STM (Either SomeException Bool)
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Bool -> Either SomeException Bool
forall a b. b -> Either a b
Right Bool
False)
              Bool
False -> Bool -> Either SomeException Bool
forall a b. b -> Either a b
Right Bool
True Either SomeException Bool
-> STM () -> STM (Either SomeException Bool)
forall a b. a -> STM b -> STM a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ StrictTBQueue IO WriteJob -> WriteJob -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTBQueue m a -> a -> STM m ()
writeTBQueue StrictTBQueue IO WriteJob
wqJobs WriteJob
job
      -- And wait for one otherwise.
      park =
        StrictTVar IO (Maybe SomeException) -> STM IO (Maybe SomeException)
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar IO (Maybe SomeException)
wqSealed STM (Maybe SomeException)
-> (Maybe SomeException -> STM (IO ())) -> STM (IO ())
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= \case
          Just SomeException
cause -> IO () -> STM (IO ())
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (LeiosDbException -> IO ()
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO (SomeException -> LeiosDbException
wrap SomeException
cause))
          Maybe SomeException
Nothing -> StrictTBQueue IO WriteJob -> WriteJob -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTBQueue m a -> a -> STM m ()
writeTBQueue StrictTBQueue IO WriteJob
wqJobs WriteJob
job STM () -> STM (IO ()) -> STM (IO ())
forall a b. STM a -> STM b -> STM b
forall (m :: * -> *) a b. Monad m => m a -> m b -> m b
>> IO () -> STM (IO ())
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (() -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ())
  atomically offer >>= \case
    Left SomeException
cause -> LeiosDbException -> IO ()
forall e a. (HasCallStack, Exception e) => e -> IO a
throwIO (SomeException -> LeiosDbException
wrap SomeException
cause)
    Right Bool
True -> () -> IO ()
forall a. a -> IO a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
    Right Bool
False -> do
      Tracer IO TraceLeiosDb -> TraceLeiosDb -> IO ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer IO TraceLeiosDb
wqTracer (TraceLeiosDb -> IO ()) -> TraceLeiosDb -> IO ()
forall a b. (a -> b) -> a -> b
$ [Char] -> TraceLeiosDb
TraceLeiosDbWriterQueueFull (WriteJob -> [Char]
describeJob WriteJob
job)
      -- Otherwise silent: no exception reaches the write path, and there is
      -- no continuation yet to carry one.
      IO (IO ()) -> IO ()
forall (m :: * -> *) a. Monad m => m (m a) -> m a
join (STM IO (IO ()) -> IO (IO ())
forall a. HasCallStack => STM IO a -> IO a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically STM (IO ())
STM IO (IO ())
park)
        IO () -> IO () -> IO ()
forall a b. IO a -> IO b -> IO a
forall (m :: * -> *) a b. MonadCatch m => m a -> m b -> m a
`onException` Tracer IO TraceLeiosDb -> TraceLeiosDb -> IO ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer IO TraceLeiosDb
wqTracer ([Char] -> TraceLeiosDb
TraceLeiosDbWriteAbandoned (WriteJob -> [Char]
describeJob WriteJob
job))
  pure $ Promise (either (throwIO . wrap) pure =<< atomically (readTMVar resultVar))

-- | Name a job for 'LeiosDbWriteException': what it is and what it is about,
-- never its payload.
describeJob :: WriteJob -> String
describeJob :: WriteJob -> [Char]
describeJob = \case
  WriteEbPoint LeiosPoint
point BytesSize
_ WriteResult ()
_ -> [Char]
"WriteEbPoint " [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> LeiosPoint -> [Char]
forall a. Show a => a -> [Char]
show LeiosPoint
point
  WriteEbBody LeiosPoint
point LeiosEb
eb WriteResult CompletedEbs
_ -> [Char]
"WriteEbBody " [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> LeiosPoint -> [Char]
forall a. Show a => a -> [Char]
show LeiosPoint
point [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> [Char]
" (" [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> Int -> [Char]
forall a. Show a => a -> [Char]
show (Vector (TxHash, BytesSize) -> Int
forall a. Vector a -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length (LeiosEb -> Vector (TxHash, BytesSize)
leiosEbTxs LeiosEb
eb)) [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> [Char]
" txs)"
  WriteTxs [(TxHash, ByteString)]
txs WriteResult CompletedEbs
_ -> [Char]
"WriteTxs (" [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> Int -> [Char]
forall a. Show a => a -> [Char]
show ([(TxHash, ByteString)] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [(TxHash, ByteString)]
txs) [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> [Char]
" txs)"
  Flush WriteResult ()
_ -> [Char]
"Flush"
  PinEb [EbHash]
ebHashes WriteResult ()
_ -> [Char]
"PinEb (" [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> Int -> [Char]
forall a. Show a => a -> [Char]
show ([EbHash] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [EbHash]
ebHashes) [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> [Char]
" ebs)"
  MarkCopied [EbHash]
ebHashes WriteResult ()
_ -> [Char]
"MarkCopied (" [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> Int -> [Char]
forall a. Show a => a -> [Char]
show ([EbHash] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [EbHash]
ebHashes) [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> [Char]
" ebs)"
  GcMark SlotNo
slot WriteResult ()
_ -> [Char]
"GcMark " [Char] -> [Char] -> [Char]
forall a. Semigroup a => a -> a -> a
<> SlotNo -> [Char]
forall a. Show a => a -> [Char]
show SlotNo
slot
  Shutdown WriteResult ()
_ -> [Char]
"Shutdown"

-- | Publish the worker's parting exception as a queued job's result.
failJob :: SomeException -> WriteJob -> IO ()
failJob :: SomeException -> WriteJob -> IO ()
failJob SomeException
cause = \case
  WriteEbPoint LeiosPoint
_ BytesSize
_ WriteResult ()
rv -> WriteResult () -> IO ()
forall a. WriteResult a -> IO ()
put WriteResult ()
rv
  WriteEbBody LeiosPoint
_ LeiosEb
_ WriteResult CompletedEbs
rv -> WriteResult CompletedEbs -> IO ()
forall a. WriteResult a -> IO ()
put WriteResult CompletedEbs
rv
  WriteTxs [(TxHash, ByteString)]
_ WriteResult CompletedEbs
rv -> WriteResult CompletedEbs -> IO ()
forall a. WriteResult a -> IO ()
put WriteResult CompletedEbs
rv
  Flush WriteResult ()
rv -> WriteResult () -> IO ()
forall a. WriteResult a -> IO ()
put WriteResult ()
rv
  PinEb [EbHash]
_ WriteResult ()
rv -> WriteResult () -> IO ()
forall a. WriteResult a -> IO ()
put WriteResult ()
rv
  MarkCopied [EbHash]
_ WriteResult ()
rv -> WriteResult () -> IO ()
forall a. WriteResult a -> IO ()
put WriteResult ()
rv
  GcMark SlotNo
_ WriteResult ()
rv -> WriteResult () -> IO ()
forall a. WriteResult a -> IO ()
put WriteResult ()
rv
  Shutdown WriteResult ()
rv -> WriteResult () -> IO ()
forall a. WriteResult a -> IO ()
put WriteResult ()
rv
 where
  put :: WriteResult a -> IO ()
  put :: forall a. WriteResult a -> IO ()
put WriteResult a
rv = 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
$ WriteResult a -> Either SomeException a -> STM IO ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTMVar m a -> a -> STM m ()
putTMVar WriteResult a
rv (SomeException -> Either SomeException a
forall a b. a -> Either a b
Left SomeException
cause)

-- | How many queued jobs the writer serves back-to-back before it takes a
-- turn of maintenance anyway.
--
-- Maintenance runs when the queue drains, which is the common case; this is
-- the floor under a queue that never does. One queue-full: under saturation
-- maintenance still gets a turn as often as the producers can refill, and
-- the resulting share (one turn in @'writerQueueDepth' + 1@) is far above
-- what copying and eviction ask for -- they are per promoted or expired EB,
-- not per write. 'TraceLeiosDbStats' is where a volatile partition that
-- still grows would show up.
maxJobsBetweenMaintenance :: Int
maxJobsBetweenMaintenance :: Int
maxJobsBetweenMaintenance = Natural -> Int
forall a b. (Integral a, Num b) => a -> b
fromIntegral Natural
writerQueueDepth

-- | Depth of the write queue. One slot per producer that can be mid-write --
-- each upstream peer's fetch client, the forge, and the maintenance
-- schedulers (copier, sweeper, the ChainDB's GC and promote calls) -- and a
-- little slack. Deliberately shallow: depth beyond that adds no throughput
-- (there is one worker) and only delays the commit-time notifications that
-- gate vote scheduling and chain selection. A producer that outruns the disk
-- blocks on submission, which is the backpressure.
writerQueueDepth :: Natural
writerQueueDepth :: Natural
writerQueueDepth = Natural
numUpstreamPeers Natural -> Natural -> Natural
forall a. Num a => a -> a -> a
+ Natural
forge Natural -> Natural -> Natural
forall a. Num a => a -> a -> a
+ Natural
maintenance Natural -> Natural -> Natural
forall a. Num a => a -> a -> a
+ Natural
slack
 where
  numUpstreamPeers :: Natural
numUpstreamPeers = Natural
20
  forge :: Natural
forge = Natural
1
  maintenance :: Natural
maintenance = Natural
4
  slack :: Natural
slack = Natural
2