Track the last time the ChainDB thread was starved

Author: Niols

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

@@ -52,6 +52,7 @@ data ChainDbView m blk = ChainDbView {
    , getIsFetched              :: STM m (Point blk -> Bool)
    , getMaxSlotNo              :: STM m MaxSlotNo
    , addBlockWaitWrittenToDisk :: InvalidBlockPunishment m -> blk -> m Bool
+   , getLastTimeStarved        :: STM m Time
    }
 
 defaultChainDbView :: IOLike m => ChainDB m blk -> ChainDbView m blk
@@ -60,6 +61,7 @@ defaultChainDbView chainDB = ChainDbView {
   , getIsFetched              = ChainDB.getIsFetched chainDB
   , getMaxSlotNo              = ChainDB.getMaxSlotNo chainDB
   , addBlockWaitWrittenToDisk = ChainDB.addBlockWaitWrittenToDisk chainDB
+  , getLastTimeStarved        = ChainDB.getLastTimeStarved chainDB
   }
 
 -- | How to get the wall-clock time of a slot. Note that this is a very
@@ -340,3 +342,5 @@ mkBlockFetchConsensusInterface
 
     headerForgeUTCTime = slotForgeTime . headerRealPoint . unFromConsensus
     blockForgeUTCTime  = slotForgeTime . blockRealPoint  . unFromConsensus
+
+    lastChainSelStarvation = getLastTimeStarved chainDB

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

@@ -334,6 +334,10 @@ data ChainDB m blk = ChainDB {
       -- invalid block is detected. These blocks are likely to be valid.
     , getIsInvalidBlock :: STM m (WithFingerprint (HeaderHash blk -> Maybe (InvalidBlockReason blk)))
 
+      -- | The last time we starved the chainsel thread. this is used by the
+      -- blockfetch decision logic to demote peers.
+    , getLastTimeStarved :: STM m Time
+
     , closeDB            :: m ()
 
       -- | Return 'True' when the database is open.

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

@@ -176,6 +176,7 @@ openDBInternal args launchBgTasks = runWithTempRegistry $ do
       copyFuse           <- newFuse "copy to immutable db"
       chainSelFuse       <- newFuse "chain selection"
       chainSelQueue      <- newChainSelQueue (Args.cdbsBlocksToAddSize cdbSpecificArgs)
+      varLastTimeStarved <- newTVarIO =<< getMonotonicTime
 
       let env = CDB { cdbImmutableDB     = immutableDB
                     , cdbVolatileDB      = volatileDB
@@ -200,6 +201,7 @@ openDBInternal args launchBgTasks = runWithTempRegistry $ do
                     , cdbChainSelQueue   = chainSelQueue
                     , cdbFutureBlocks    = varFutureBlocks
                     , cdbLoE             = Args.cdbsLoE cdbSpecificArgs
+                    , cdbLastTimeStarved = varLastTimeStarved
                     }
       h <- fmap CDBHandle $ newTVarIO $ ChainDbOpen env
       let chainDB = API.ChainDB
@@ -217,6 +219,7 @@ openDBInternal args launchBgTasks = runWithTempRegistry $ do
             , stream                = Iterator.stream  h
             , newFollower           = Follower.newFollower h
             , getIsInvalidBlock     = getEnvSTM  h Query.getIsInvalidBlock
+            , getLastTimeStarved    = getEnvSTM  h Query.getLastTimeStarved
             , 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 (writeTVar cdbLastTimeStarved) cdbChainSelQueue)
         (\message -> lift $ atomically $ do
           case message of
             ChainSelReprocessLoEBlocks -> pure ()

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

@@ -21,6 +21,7 @@ module Ouroboros.Consensus.Storage.ChainDB.Impl.Query (
   , getAnyBlockComponent
   , getAnyKnownBlock
   , getAnyKnownBlockComponent
+  , getLastTimeStarved
   ) where
 
 import qualified Data.Map.Strict as Map
@@ -148,6 +149,9 @@ getIsInvalidBlock ::
 getIsInvalidBlock CDB{..} =
   fmap (fmap (fmap invalidBlockReason) . flip Map.lookup) <$> readTVar cdbInvalid
 
+getLastTimeStarved :: forall m blk. IOLike m => ChainDbEnv m blk -> STM m Time
+getLastTimeStarved CDB{..} = readTVar cdbLastTimeStarved
+
 getIsValid ::
      forall m blk. (IOLike m, HasHeader blk)
   => ChainDbEnv m blk

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

@@ -63,6 +63,7 @@ module Ouroboros.Consensus.Storage.ChainDB.Impl.Types (
   , TraceValidationEvent (..)
   ) where
 
+import           Cardano.Prelude (whenM)
 import           Control.Tracer
 import           Data.Foldable (traverse_)
 import           Data.Map.Strict (Map)
@@ -275,6 +276,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.
+  , cdbLastTimeStarved :: !(StrictTVar m Time)
+    -- ^ The last time we starved the ChainSel thread. This is used by the
+    -- BlockFetch decision logic to demote peers.
   } deriving (Generic)
 
 -- | We include @blk@ in 'showTypeOf' because it helps resolving type families
@@ -509,9 +513,13 @@ addReprocessLoEBlocks tracer (ChainSelQueue queue) = do
   atomically $ writeTBQueue queue ChainSelReprocessLoEBlocks
 
 -- | 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 current time to the given callback.
+getChainSelMessage :: IOLike m => (Time -> STM m ()) -> ChainSelQueue m blk -> m (ChainSelMessage m blk)
+getChainSelMessage whenEmpty (ChainSelQueue queue) = do
+  time <- getMonotonicTime
+  -- NOTE: The two following lines are in different `atomically` on purpose.
+  atomically $ whenM (isEmptyTBQueue queue) (whenEmpty time)
+  atomically $ readTBQueue queue
 
 -- | Flush the 'ChainSelQueue' queue and notify the waiting threads.
 --