Track the last time the ChainDB thread was starved

Author: Niols

ouroboros-consensus-diffusion/test/consensus-test/Test/Consensus/PeerSimulator/Trace.hs

@@ -370,6 +370,10 @@ traceChainDBEventTestBlockWith tracer = \case
         AddedReprocessLoEBlocksToQueue ->
           trace $ "Requested ChainSel run"
         _ -> pure ()
+    ChainDB.TraceChainSelStarvationEvent (ChainDB.ChainSelStarvationStarted time) ->
+      trace $ "ChainSel starvation started at " ++ prettyTime time
+    ChainDB.TraceChainSelStarvationEvent (ChainDB.ChainSelStarvationEnded time pt) ->
+      trace $ "ChainSel starvation ended at " ++ prettyTime time ++ " thanks to " ++ terseRealPoint pt
     _ -> pure ()
   where
     trace = traceUnitWith tracer "ChainDB"

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/MiniProtocol/BlockFetch/ClientInterface.hs

@@ -42,7 +42,8 @@ import           Ouroboros.Network.AnchoredFragment (AnchoredFragment)
 import qualified Ouroboros.Network.AnchoredFragment as AF
 import           Ouroboros.Network.Block (MaxSlotNo)
 import           Ouroboros.Network.BlockFetch.ConsensusInterface
-                     (BlockFetchConsensusInterface (..), FetchMode (..),
+                     (BlockFetchConsensusInterface (..),
+                     ChainSelStarvation (..), FetchMode (..),
                      FromConsensus (..))
 import           Ouroboros.Network.PeerSelection.Bootstrap (UseBootstrapPeers,
                      requiresBootstrapPeers)
@@ -56,6 +57,7 @@ data ChainDbView m blk = ChainDbView {
    , getIsFetched              :: STM m (Point blk -> Bool)
    , getMaxSlotNo              :: STM m MaxSlotNo
    , addBlockWaitWrittenToDisk :: InvalidBlockPunishment m -> blk -> m Bool
+   , getChainSelStarvation     :: STM m ChainSelStarvation
    }
 
 defaultChainDbView :: IOLike m => ChainDB m blk -> ChainDbView m blk
@@ -64,6 +66,7 @@ defaultChainDbView chainDB = ChainDbView {
   , getIsFetched              = ChainDB.getIsFetched chainDB
   , getMaxSlotNo              = ChainDB.getMaxSlotNo chainDB
   , addBlockWaitWrittenToDisk = ChainDB.addBlockWaitWrittenToDisk chainDB
+  , getChainSelStarvation     = ChainDB.getChainSelStarvation chainDB
   }
 
 -- | How to get the wall-clock time of a slot. Note that this is a very
@@ -338,5 +341,7 @@ mkBlockFetchConsensusInterface
     headerForgeUTCTime = slotForgeTime . headerRealPoint . unFromConsensus
     blockForgeUTCTime  = slotForgeTime . blockRealPoint  . unFromConsensus
 
+    readChainSelStarvation = getChainSelStarvation chainDB
+
     demoteCSJDynamo :: peer -> m ()
     demoteCSJDynamo = void . atomically . Jumping.rotateDynamo csHandlesCol

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/API.hs

@@ -91,6 +91,8 @@ import qualified Ouroboros.Network.AnchoredFragment as AF
 import           Ouroboros.Network.Block (ChainUpdate, MaxSlotNo,
                      Serialised (..))
 import qualified Ouroboros.Network.Block as Network
+import           Ouroboros.Network.BlockFetch.ConsensusInterface
+                     (ChainSelStarvation (..))
 import           Ouroboros.Network.Mock.Chain (Chain (..))
 import qualified Ouroboros.Network.Mock.Chain as Chain
 import           System.FS.API.Types (FsError)
@@ -347,6 +349,10 @@ data ChainDB m blk = ChainDB {
       -- invalid block is detected. These blocks are likely to be valid.
     , getIsInvalidBlock :: STM m (WithFingerprint (HeaderHash blk -> Maybe (ExtValidationError blk)))
 
+      -- | Whether ChainSel is currently starved, or when was last time it
+      -- stopped being starved.
+    , getChainSelStarvation :: STM m ChainSelStarvation
+
     , closeDB            :: m ()
 
       -- | Return 'True' when the database is open.

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/Impl.hs

@@ -16,6 +16,7 @@ module Ouroboros.Consensus.Storage.ChainDB.Impl (
   , LgrDB.TraceReplayEvent
   , SelectionChangedInfo (..)
   , TraceAddBlockEvent (..)
+  , TraceChainSelStarvationEvent (..)
   , TraceCopyToImmutableDBEvent (..)
   , TraceEvent (..)
   , TraceFollowerEvent (..)
@@ -69,6 +70,8 @@ import           Ouroboros.Consensus.Util.IOLike
 import           Ouroboros.Consensus.Util.STM (Fingerprint (..),
                      WithFingerprint (..))
 import qualified Ouroboros.Network.AnchoredFragment as AF
+import           Ouroboros.Network.BlockFetch.ConsensusInterface
+                     (ChainSelStarvation (..))
 
 {-------------------------------------------------------------------------------
   Initialization
@@ -174,6 +177,7 @@ openDBInternal args launchBgTasks = runWithTempRegistry $ do
       copyFuse           <- newFuse "copy to immutable db"
       chainSelFuse       <- newFuse "chain selection"
       chainSelQueue      <- newChainSelQueue (Args.cdbsBlocksToAddSize cdbSpecificArgs)
+      varChainSelStarvation <- newTVarIO ChainSelStarvationOngoing
 
       let env = CDB { cdbImmutableDB     = immutableDB
                     , cdbVolatileDB      = volatileDB
@@ -196,6 +200,7 @@ openDBInternal args launchBgTasks = runWithTempRegistry $ do
                     , cdbKillBgThreads   = varKillBgThreads
                     , cdbChainSelQueue   = chainSelQueue
                     , cdbLoE             = Args.cdbsLoE cdbSpecificArgs
+                    , cdbChainSelStarvation = varChainSelStarvation
                     }
       h <- fmap CDBHandle $ newTVarIO $ ChainDbOpen env
       let chainDB = API.ChainDB
@@ -214,6 +219,7 @@ openDBInternal args launchBgTasks = runWithTempRegistry $ do
             , stream                = Iterator.stream  h
             , newFollower           = Follower.newFollower h
             , getIsInvalidBlock     = getEnvSTM  h Query.getIsInvalidBlock
+            , getChainSelStarvation = getEnvSTM  h Query.getChainSelStarvation
             , closeDB               = closeDB h
             , isOpen                = isOpen  h
             }

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/Impl/Background.hs

@@ -522,7 +522,7 @@ addBlockRunner fuse cdb@CDB{..} = forever $ do
     -- exception (or it errored), notify the blocked thread
     withFuse fuse $
       bracketOnError
-        (lift $ getChainSelMessage cdbChainSelQueue)
+        (lift $ getChainSelMessage starvationTracer cdbChainSelStarvation cdbChainSelQueue)
         (\message -> lift $ atomically $ do
           case message of
             ChainSelReprocessLoEBlocks varProcessed ->
@@ -542,3 +542,5 @@ addBlockRunner fuse cdb@CDB{..} = forever $ do
               trace $ PoppedBlockFromQueue $ FallingEdgeWith $
                       blockRealPoint blockToAdd
           chainSelSync cdb message)
+  where
+    starvationTracer = Tracer $ traceWith cdbTracer . TraceChainSelStarvationEvent

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/Impl/Query.hs

@@ -22,6 +22,7 @@ module Ouroboros.Consensus.Storage.ChainDB.Impl.Query (
   , getAnyBlockComponent
   , getAnyKnownBlock
   , getAnyKnownBlockComponent
+  , getChainSelStarvation
   ) where
 
 import qualified Data.Map.Strict as Map
@@ -50,6 +51,8 @@ import           Ouroboros.Consensus.Util.STM (WithFingerprint (..))
 import           Ouroboros.Network.AnchoredFragment (AnchoredFragment)
 import qualified Ouroboros.Network.AnchoredFragment as AF
 import           Ouroboros.Network.Block (MaxSlotNo, maxSlotNoFromWithOrigin)
+import           Ouroboros.Network.BlockFetch.ConsensusInterface
+                     (ChainSelStarvation (..))
 
 -- | Return the last @k@ headers.
 --
@@ -181,6 +184,12 @@ getIsInvalidBlock ::
 getIsInvalidBlock CDB{..} =
   fmap (fmap (fmap invalidBlockReason) . flip Map.lookup) <$> readTVar cdbInvalid
 
+getChainSelStarvation ::
+     forall m blk. IOLike m
+  => ChainDbEnv m blk
+  -> STM m ChainSelStarvation
+getChainSelStarvation CDB {..} = readTVar cdbChainSelStarvation
+
 getIsValid ::
      forall m blk. (IOLike m, HasHeader blk)
   => ChainDbEnv m blk

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/Storage/ChainDB/Impl/Types.hs

@@ -51,6 +51,7 @@ module Ouroboros.Consensus.Storage.ChainDB.Impl.Types (
     -- * Trace types
   , SelectionChangedInfo (..)
   , TraceAddBlockEvent (..)
+  , TraceChainSelStarvationEvent (..)
   , TraceCopyToImmutableDBEvent (..)
   , TraceEvent (..)
   , TraceFollowerEvent (..)
@@ -62,6 +63,7 @@ module Ouroboros.Consensus.Storage.ChainDB.Impl.Types (
   , TraceValidationEvent (..)
   ) where
 
+import           Cardano.Prelude (whenM)
 import           Control.ResourceRegistry
 import           Control.Tracer
 import           Data.Foldable (traverse_)
@@ -105,6 +107,8 @@ import           Ouroboros.Consensus.Util.IOLike
 import           Ouroboros.Consensus.Util.STM (WithFingerprint)
 import           Ouroboros.Network.AnchoredFragment (AnchoredFragment)
 import           Ouroboros.Network.Block (MaxSlotNo)
+import           Ouroboros.Network.BlockFetch.ConsensusInterface
+                     (ChainSelStarvation (..))
 
 -- | All the serialisation related constraints needed by the ChainDB.
 class ( ImmutableDbSerialiseConstraints blk
@@ -254,6 +258,9 @@ data ChainDbEnv m blk = CDB
     -- switch back to a chain containing it. The fragment is usually anchored at
     -- a recent immutable tip; if it does not, it will conservatively be treated
     -- as the empty fragment anchored in the current immutable tip.
+  , cdbChainSelStarvation :: !(StrictTVar m ChainSelStarvation)
+    -- ^ Information on the last starvation of ChainSel, whether ongoing or
+    -- ended recently.
   } deriving (Generic)
 
 -- | We include @blk@ in 'showTypeOf' because it helps resolving type families
@@ -483,9 +490,42 @@ addReprocessLoEBlocks tracer (ChainSelQueue queue) = do
   return $ ChainSelectionPromise waitUntilRan
 
 -- | Get the oldest message from the 'ChainSelQueue' queue. Can block when the
--- queue is empty.
-getChainSelMessage :: IOLike m => ChainSelQueue m blk -> m (ChainSelMessage m blk)
-getChainSelMessage (ChainSelQueue queue) = atomically $ readTBQueue queue
+-- queue is empty; in that case, reports the starvation (and its end) via the
+-- given tracer.
+getChainSelMessage
+  :: forall m blk. (HasHeader blk, IOLike m)
+  => Tracer m (TraceChainSelStarvationEvent blk)
+  -> StrictTVar m ChainSelStarvation
+  -> ChainSelQueue m blk
+  -> m (ChainSelMessage m blk)
+getChainSelMessage starvationTracer starvationVar (ChainSelQueue queue) =
+    atomically (tryReadTBQueue' queue) >>= \case
+      Just msg -> pure msg
+      Nothing  -> do
+        startStarvationMeasure
+        msg <- atomically $ readTBQueue queue
+        terminateStarvationMeasure msg
+        pure msg
+  where
+    startStarvationMeasure :: m ()
+    startStarvationMeasure = do
+      prevStarvation <- atomically $ swapTVar starvationVar ChainSelStarvationOngoing
+      when (prevStarvation /= ChainSelStarvationOngoing) $
+        traceWith starvationTracer . ChainSelStarvationStarted =<< getMonotonicTime
+
+    terminateStarvationMeasure :: ChainSelMessage m blk -> m ()
+    terminateStarvationMeasure = \case
+      ChainSelAddBlock BlockToAdd{blockToAdd=block} -> do
+        tf <- getMonotonicTime
+        let pt = blockRealPoint block
+        traceWith starvationTracer $ ChainSelStarvationEnded tf pt
+        atomically $ writeTVar starvationVar (ChainSelStarvationEndedAt tf)
+      ChainSelReprocessLoEBlocks{} -> pure ()
+
+-- TODO Can't use tryReadTBQueue from io-classes because it is broken for IOSim
+-- (but not for IO). https://github.com/input-output-hk/io-sim/issues/195
+tryReadTBQueue' :: MonadSTM m => TBQueue m a -> STM m (Maybe a)
+tryReadTBQueue' q = (Just <$> readTBQueue q) `orElse` pure Nothing
 
 -- | Flush the 'ChainSelQueue' queue and notify the waiting threads.
 --
@@ -519,6 +559,7 @@ data TraceEvent blk
   | TraceImmutableDBEvent       (ImmutableDB.TraceEvent       blk)
   | TraceVolatileDBEvent        (VolatileDB.TraceEvent        blk)
   | TraceLastShutdownUnclean
+  | TraceChainSelStarvationEvent(TraceChainSelStarvationEvent blk)
   deriving (Generic)
 
 
@@ -827,3 +868,16 @@ data TraceIteratorEvent blk
     -- next block we're looking for.
   | SwitchBackToVolatileDB
   deriving (Generic, Eq, Show)
+
+-- | Chain selection is /starved/ when the background thread runs out of work.
+-- This is the usual case and innocent while caught-up; but while syncing, it
+-- means that we are downloading blocks at a smaller rate than we can validate
+-- them, even though we generally expect to be CPU-bound.
+data TraceChainSelStarvationEvent blk
+    -- | A ChainSel starvation started at the given time.
+  = ChainSelStarvationStarted Time
+
+    -- | The last ChainSel starvation ended at the given time as a block wth the
+    -- given point has been received.
+  | ChainSelStarvationEnded Time (RealPoint blk)
+  deriving (Generic, Eq, Show)

ouroboros-consensus/test/storage-test/Test/Ouroboros/Storage/ChainDB/StateMachine.hs

@@ -1241,6 +1241,8 @@ deriving instance SOP.Generic         (ImmutableDB.TraceEvent blk)
 deriving instance SOP.HasDatatypeInfo (ImmutableDB.TraceEvent blk)
 deriving instance SOP.Generic         (VolatileDB.TraceEvent blk)
 deriving instance SOP.HasDatatypeInfo (VolatileDB.TraceEvent blk)
+deriving instance SOP.Generic         (TraceChainSelStarvationEvent blk)
+deriving instance SOP.HasDatatypeInfo (TraceChainSelStarvationEvent blk)
 
 data Tag =
     TagGetIsValidJust
@@ -1635,7 +1637,7 @@ traceEventName = \case
     TraceImmutableDBEvent       ev    -> "ImmutableDB."       <> constrName ev
     TraceVolatileDBEvent        ev    -> "VolatileDB."        <> constrName ev
     TraceLastShutdownUnclean          -> "LastShutdownUnclean"
-    TraceChainSelStarvationEvent _    -> "TraceChainSelStarvationEvent"
+    TraceChainSelStarvationEvent ev   -> "ChainSelStarvation." <> constrName ev
 
 mkArgs :: IOLike m
        => TopLevelConfig Blk