simulation: handle NumCores in praosNode

Author: Saizan

simulation/ouroboros-leios-sim.cabal

@@ -61,7 +61,6 @@ library
     LeiosProtocol.Short.VizSim
     LeiosProtocol.Short.VizSimP2P
     LeiosProtocol.SimTestRelay
-    LeiosProtocol.TaskMultiQueue
     LeiosProtocol.VizSimTestRelay
     ModelTCP
     JSONCompat
@@ -91,6 +90,7 @@ library
     STMCompat
     SimTCPLinks
     SimTypes
+    TaskMultiQueue
     TimeCompat
     Topology
     Viz

simulation/src/LeiosProtocol/Short/Generate.hs

@@ -1,6 +1,7 @@
 {-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RecordWildCards #-}
@@ -14,16 +15,13 @@ import Control.Monad (forM)
 import Control.Monad.State (
   MonadState (get, put),
   MonadTrans (lift),
-  StateT (runStateT),
-  gets,
-  runState,
+  StateT,
  )
 import Data.Bifunctor (Bifunctor (..))
 import Data.Kind (Type)
 import LeiosProtocol.Common
 import LeiosProtocol.Short hiding (Stage (..))
 import STMCompat
-import System.Random (StdGen, uniformR)
 
 --------------------------------------------------------------------------------
 
@@ -50,42 +48,7 @@ data SomeRole :: Type where
 data SomeAction :: Type where
   SomeAction :: Role a -> a -> SomeAction
 
-mkScheduler :: MonadSTM m => StdGen -> (SlotNo -> [(a, Maybe (Double -> Word64))]) -> m (SlotNo -> m [(a, Word64)])
-mkScheduler rng0 rates = do
-  let
-    sampleRates (_role, Nothing) = return []
-    sampleRates (role, Just f) = do
-      (sample, rng') <- gets $ uniformR (0, 1)
-      put $! rng'
-      let wins = f sample
-      return [(role, wins) | wins >= 1]
-  rngVar <- newTVarIO rng0
-  let sched slot = atomically $ do
-        rng <- readTVar rngVar
-        let (acts, rng1) = flip runState rng . fmap concat . mapM sampleRates $ rates slot
-        writeTVar rngVar rng1
-        return acts
-  return sched
-
--- | @waitNextSlot cfg targetSlot@ waits until the beginning of
--- @targetSlot@ if that's now or in the future, otherwise the closest slot.
-waitNextSlot :: (Monad m, MonadTime m, MonadDelay m) => SlotConfig -> SlotNo -> m SlotNo
-waitNextSlot slotConfig targetSlot = do
-  now <- getCurrentTime
-  let targetSlotTime = slotTime slotConfig targetSlot
-  let slot
-        | now <= targetSlotTime = targetSlot
-        | otherwise = assert (nextSlotIndex >= 0) $ toEnum nextSlotIndex
-       where
-        nextSlotIndex =
-          assert (slotConfig.duration == 1) $
-            ceiling $
-              now `diffUTCTime` slotConfig.start
-  let tgt = slotTime slotConfig slot
-  threadDelayNDT (tgt `diffUTCTime` now)
-  return slot
-
-data BlockGeneratorConfig m = BlockGeneratorConfig
+data LeiosGeneratorConfig m = LeiosGeneratorConfig
   { leios :: LeiosConfig
   , slotConfig :: SlotConfig
   , nodeId :: NodeId
@@ -94,19 +57,21 @@ data BlockGeneratorConfig m = BlockGeneratorConfig
   , submit :: [(DiffTime, SomeAction)] -> m ()
   }
 
-blockGenerator ::
+leiosBlockGenerator ::
   forall m.
   (MonadSTM m, MonadDelay m, MonadTime m) =>
-  BlockGeneratorConfig m ->
+  LeiosGeneratorConfig m ->
   m ()
-blockGenerator BlockGeneratorConfig{..} = go (0, 0)
+leiosBlockGenerator LeiosGeneratorConfig{..} =
+  blockGenerator $
+    BlockGeneratorConfig
+      { execute = \slot -> do
+          roles <- lift $ schedule slot
+          actions <- concat <$> mapM (execute slot) roles
+          lift $ submit actions
+      , slotConfig
+      }
  where
-  go (!blkId, !tgtSlot) = do
-    slot <- waitNextSlot slotConfig tgtSlot
-    roles <- schedule slot
-    (actions, blkId') <- runStateT (concat <$> mapM (execute slot) roles) blkId
-    submit actions
-    go (blkId', slot + 1)
   execute slot (SomeRole r, wins) = assert (wins >= 1) $ (map . second) (SomeAction r) <$> execute' slot r wins
   execute' :: SlotNo -> Role a -> Word64 -> StateT Int m [(DiffTime, a)]
   execute' slot Base _wins = do

simulation/src/LeiosProtocol/Short/Node.hs

@@ -24,7 +24,7 @@ import Control.Monad.Class.MonadThrow
 import Control.Tracer
 import Data.Coerce (coerce)
 import Data.Foldable (forM_)
-import Data.Ix (Ix, range)
+import Data.Ix (Ix)
 import Data.List (sort, sortOn)
 import Data.Map (Map)
 import qualified Data.Map.Strict as Map
@@ -39,7 +39,6 @@ import qualified LeiosProtocol.RelayBuffer as RB
 import LeiosProtocol.Short
 import LeiosProtocol.Short.Generate
 import qualified LeiosProtocol.Short.Generate as Generate
-import LeiosProtocol.TaskMultiQueue
 import Numeric.Natural (Natural)
 import PraosProtocol.BlockFetch (
   BlockFetchControllerState (blocksVar),
@@ -51,7 +50,7 @@ import qualified PraosProtocol.PraosNode as PraosNode
 import STMCompat
 import SimTypes (cpuTask)
 import System.Random
-import WorkerPool
+import TaskMultiQueue
 
 --------------------------------------------------------------
 ---- Events
@@ -99,9 +98,9 @@ data LeiosNodeState m = LeiosNodeState
   , relayVoteState :: !(RelayVoteState m)
   , ibDeliveryTimesVar :: !(TVar m (Map InputBlockId UTCTime))
   , taskQueue :: !(TaskMultiQueue LeiosNodeTask m)
-  , waitingForRBVar :: !(TVar m (Map (HeaderHash RankingBlock) [m ()]))
+  , waitingForRBVar :: !(TVar m (Map (HeaderHash RankingBlock) [STM m ()]))
   -- ^ waiting for RB block itself to be validated.
-  , waitingForLedgerStateVar :: !(TVar m (Map (HeaderHash RankingBlock) [m ()]))
+  , waitingForLedgerStateVar :: !(TVar m (Map (HeaderHash RankingBlock) [STM m ()]))
   -- ^ waiting for ledger state of RB block to be validated.
   , ledgerStateVar :: !(TVar m (Map (HeaderHash RankingBlock) LedgerState))
   , ibsNeededForEBVar :: !(TVar m (Map EndorseBlockId (Set InputBlockId)))
@@ -400,25 +399,6 @@ leiosNode tracer cfg followers peers = do
       , pruningThreads
       ]
 
-processCPUTasks ::
-  (MonadSTM m, MonadDelay m, MonadMonotonicTimeNSec m, MonadFork m, MonadAsync m, MonadCatch m) =>
-  NumCores ->
-  Tracer m CPUTask ->
-  TaskMultiQueue LeiosNodeTask m ->
-  m ()
-processCPUTasks Infinite tracer queue = forever $ runInfParallelBlocking tracer queue
-processCPUTasks (Finite n) tracer queue = newBoundedWorkerPool n [taskSource l | l <- range (minBound, maxBound)]
- where
-  taskSource l = do
-    (cpu, m) <- readTMQueue queue l
-    var <- newEmptyTMVar
-    let action = do
-          traceWith tracer cpu
-          threadDelay (cpuTaskDuration cpu)
-          m
-    -- TODO: read from var and log exception.
-    return $ Task action var
-
 computeLedgerStateThread ::
   forall m.
   (MonadMVar m, MonadFork m, MonadAsync m, MonadSTM m, MonadTime m, MonadDelay m) =>
@@ -456,7 +436,7 @@ dispatchValidation ::
 dispatchValidation tracer cfg leiosState req =
   atomically $ mapM_ (uncurry $ writeTMQueue leiosState.taskQueue) =<< go req
  where
-  queue = atomically . mapM_ (uncurry $ writeTMQueue leiosState.taskQueue)
+  queue = mapM_ (uncurry $ writeTMQueue leiosState.taskQueue)
   labelTask (tag, (f, m)) = let !task = f (show tag) in (tag, (task, m))
   valRB rb m = do
     let task prefix = cpuTask prefix cfg.leios.praos.blockValidationDelay rb
@@ -561,7 +541,7 @@ generator tracer cfg st = do
           atomically $ modifyTVar' st.relayVoteState.relayBufferVar (RB.snoc v.id (v.id, v))
           traceWith tracer (LeiosNodeEvent Generate (EventVote v))
   let LeiosNodeConfig{..} = cfg
-  blockGenerator $ BlockGeneratorConfig{submit = mapM_ submitOne, ..}
+  leiosBlockGenerator $ LeiosGeneratorConfig{submit = mapM_ submitOne, ..}
 
 mkBuffersView :: forall m. MonadSTM m => LeiosNodeConfig -> LeiosNodeState m -> BuffersView m
 mkBuffersView cfg st = BuffersView{..}

simulation/src/PraosProtocol/BlockFetch.hs

@@ -26,9 +26,8 @@ import Chan (Chan)
 import ChanDriver (ProtocolMessage, chanDriver)
 import Control.Exception (assert)
 import Control.Monad (forM, forever, guard, join, unless, void, when, (<=<))
-import Control.Tracer (Contravariant (contramap), Tracer, traceWith)
+import Control.Tracer (Tracer, traceWith)
 import Data.Bifunctor (second)
-import Data.Foldable (forM_)
 import Data.Kind (Type)
 import qualified Data.List as List
 import Data.Map.Strict (Map)
@@ -47,7 +46,6 @@ import Network.TypedProtocol (
 import Network.TypedProtocol.Driver (runPeerWithDriver)
 import qualified Network.TypedProtocol.Peer.Client as TC
 import qualified Network.TypedProtocol.Peer.Server as TS
-import Numeric.Natural (Natural)
 import PraosProtocol.Common
 import qualified PraosProtocol.Common.AnchoredFragment as AnchoredFragment
 import qualified PraosProtocol.Common.Chain as Chain
@@ -602,82 +600,30 @@ initBlockFetchConsumerStateForPeerId tracer peerId blockFetchControllerState sub
 
 setupValidatorThreads ::
   (MonadSTM m, MonadDelay m) =>
-  Tracer m (PraosNodeEvent BlockBody) ->
   PraosConfig BlockBody ->
   BlockFetchControllerState BlockBody m ->
-  -- | bound on queue length.
-  Natural ->
+  ((CPUTask, m ()) -> STM m ()) ->
   m ([m ()], Block BlockBody -> m () -> m ())
-setupValidatorThreads tracer cfg st n = do
-  queue <- newTBQueueIO n
-  (waitingVar, processWaitingThread) <- setupProcessWaitingThread (contramap PraosNodeEventCPU tracer) (Just 1) st.blocksVar
-  let doTask (cpuTask, m) = do
-        traceWith tracer . PraosNodeEventCPU $ cpuTask
-        threadDelay cpuTask.cpuTaskDuration
-        m
-
-  -- if we have the previous block, we process the task sequentially to provide back pressure on the queue.
-  let waitForPrev block task = case blockPrevHash block of
-        GenesisHash -> doTask task
+setupValidatorThreads cfg st queue = do
+  waitingVar <- newTVarIO Map.empty
+  let processWaitingThread = processWaiting' st.blocksVar waitingVar
+
+  let waitForPrev block task = atomically $ case blockPrevHash block of
+        GenesisHash -> queue task
         BlockHash prev -> do
-          havePrev <- Map.member prev <$> readTVarIO st.blocksVar
-          -- Note: for pure praos this also means we have the ledger state.
-          if havePrev
-            then doTask task
-            else atomically $ modifyTVar' waitingVar (Map.insertWith (++) prev [task])
-      fetch = forever $ do
-        (block, completion) <- atomically $ readTBQueue queue
+          modifyTVar' waitingVar (Map.insertWith (++) prev [queue task])
+      add block completion = do
         assert (blockInvariant block) $ do
           waitForPrev block $
             let !cpuTask = CPUTask (cfg.blockValidationDelay block) (T.pack $ "Validate " ++ show (blockHash block))
              in (cpuTask, completion)
-      add block completion = atomically $ writeTBQueue queue (block, completion)
-  return ([fetch, processWaitingThread], add)
-
-setupProcessWaitingThread ::
-  forall m a b.
-  (MonadSTM m, MonadDelay m) =>
-  Tracer m CPUTask ->
-  -- | how many waiting to process in parallel
-  Maybe Int ->
-  TVar m (Map ConcreteHeaderHash a) ->
-  m (TVar m (Map ConcreteHeaderHash [(CPUTask, m b)]), m ())
-setupProcessWaitingThread tracer npar blocksVar = do
-  waitingVar <- newTVarIO Map.empty
-  return (waitingVar, processWaiting tracer npar blocksVar waitingVar)
-
-processWaiting ::
-  forall m a b.
-  (MonadSTM m, MonadDelay m) =>
-  Tracer m CPUTask ->
-  -- | how many waiting to process in parallel
-  Maybe Int ->
-  TVar m (Map ConcreteHeaderHash a) ->
-  TVar m (Map ConcreteHeaderHash [(CPUTask, m b)]) ->
-  m ()
-processWaiting tracer npar blocksVar waitingVar = go
- where
-  parallelDelay xs = do
-    let !d = maximum $ map (cpuTaskDuration . fst) xs
-    forM_ xs $ traceWith tracer . fst
-    threadDelay d
-    mapM_ snd xs
-  go = forever $ join $ atomically $ do
-    waiting <- readTVar waitingVar
-    when (Map.null waiting) retry
-    blocks <- readTVar blocksVar
-    let toValidate = Map.intersection waiting blocks
-    when (Map.null toValidate) retry
-    writeTVar waitingVar $! waiting Map.\\ toValidate
-    let chunks Nothing xs = [xs]
-        chunks (Just m) xs = map (take m) . takeWhile (not . null) . iterate (drop m) $ xs
-    return . mapM_ parallelDelay . chunks npar . concat . Map.elems $ toValidate
+  return ([processWaitingThread], add)
 
 processWaiting' ::
   forall m a b.
   (MonadSTM m, MonadDelay m) =>
   TVar m (Map ConcreteHeaderHash a) ->
-  TVar m (Map ConcreteHeaderHash [m b]) ->
+  TVar m (Map ConcreteHeaderHash [STM m b]) ->
   m ()
 processWaiting' blocksVar waitingVar = go
  where
@@ -688,4 +634,4 @@ processWaiting' blocksVar waitingVar = go
     let toValidate = Map.intersection waiting blocks
     when (Map.null toValidate) retry
     writeTVar waitingVar $! waiting Map.\\ toValidate
-    return . sequence_ . concat . Map.elems $ toValidate
+    return . mapM_ atomically . concat . Map.elems $ toValidate

simulation/src/PraosProtocol/BlockGeneration.hs

@@ -1,105 +1,69 @@
 {-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE NondecreasingIndentation #-}
 {-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeOperators #-}
 
 module PraosProtocol.BlockGeneration where
 
 import Cardano.Slotting.Slot (WithOrigin (..))
-import Control.Monad (forever)
+import Control.Monad.Trans
 import Control.Tracer
 import Data.ByteString as BS
 import Data.ByteString.Char8 as BS8
+import Data.Foldable (forM_)
 import Data.Function (fix)
-import Data.Word (Word64)
 import PraosProtocol.Common
 import qualified PraosProtocol.Common.Chain as Chain
 import STMCompat
-import System.Random (StdGen, uniformR)
+import System.Random (StdGen)
 
 -- | Returns a block that can extend the chain.
 --   PRECONDITION: the SlotNo is ahead of the chain tip.
 mkBlock :: IsBody body => Chain (Block body) -> SlotNo -> body -> Block body
 mkBlock c sl body = fixupBlock (Chain.headAnchor c) (mkPartialBlock sl body)
 
-type SlotGap = Word64
-
-data PacketGenerationPattern
-  = NoPacketGeneration
-  | UniformGenerationPattern SlotGap
-  | PoissonGenerationPattern StdGen Double
-
 mkBody :: PraosConfig BlockBody -> ByteString -> SlotNo -> BlockBody
 mkBody cfg prefix (SlotNo w) = fix $ \b ->
   BlockBody
     { bodyTag = BS.append prefix $ BS8.pack (show w)
     , bodyMessageSize = cfg.bodySize b
     }
 
-mkNextBlock ::
-  forall m.
-  MonadSTM m =>
-  PraosConfig BlockBody ->
-  PacketGenerationPattern ->
-  ByteString ->
-  m (Maybe (m (SlotNo, BlockBody)))
-mkNextBlock _cfg NoPacketGeneration _ = return Nothing
-mkNextBlock cfg (UniformGenerationPattern gap) prefix = do
-  stVar <- newTVarIO (SlotNo 0)
-  let
-    go = atomically $ do
-      last_sl <- readTVar stVar
-      let
-        !sl = SlotNo (unSlotNo last_sl + gap :: Word64)
-      writeTVar stVar sl
-      let body = mkBody cfg prefix sl
-      return (sl, body)
-  return $ Just go
-mkNextBlock cfg (PoissonGenerationPattern rng0 lambda) prefix = do
-  stVar <- newTVarIO (SlotNo 0, rng0)
-  let go = atomically $ do
-        (last_sl, rng) <- readTVar stVar
-
-        let (u, !rng') = uniformR (0, 1) rng
-            gap = round ((-log u) * lambda :: Double) :: Word64
-
-        let !sl' = SlotNo $ unSlotNo last_sl + gap
-        writeTVar stVar (sl', rng')
-        let body = mkBody cfg prefix sl'
-        return (sl', body)
-  return $ Just go
-
-blockGenerator ::
-  (IsBody body, MonadSTM m, MonadDelay m, MonadTime m) =>
+praosBlockGenerator ::
+  (IsBody body, MonadSTM m, MonadDelay m, MonadTime m, body ~ BlockBody) =>
+  StdGen ->
   Tracer m (PraosNodeEvent body) ->
   PraosConfig body ->
   SlotConfig ->
+  ByteString ->
   TVar m (ChainProducerState (Block body)) ->
   (Block body -> STM m ()) ->
-  Maybe (m (SlotNo, body)) ->
+  ((CPUTask, m ()) -> m ()) ->
   m ()
-blockGenerator _tracer _praosConfig _ _cpsVar _addBlockSt Nothing = return ()
-blockGenerator tracer praosConfig slotConfig cpsVar addBlockSt (Just nextBlock) = forever go
+praosBlockGenerator rng tracer praosConfig slotConfig prefix cpsVar addBlockSt queue = do
+  sched <- mkScheduler rng (const [((), Just $ \p -> if p <= praosConfig.blockFrequencyPerSlot then 1 else 0)])
+  blockGenerator
+    BlockGeneratorConfig{slotConfig, execute = execute sched}
  where
-  go = do
-    (sl, body) <- nextBlock
-    waitForSlot sl
-    let !delay = praosConfig.blockGenerationDelay $ mkPartialBlock sl body
-    traceWith tracer (PraosNodeEventCPU $ CPUTask delay "Block generation")
-    threadDelay delay
-    mblk <- atomically $ do
-      chain <- chainState <$> readTVar cpsVar
-      let block = case mkBlock chain sl body of
-            Block h b -> Block (h{headerMessageSize = praosConfig.headerSize}) b
-      if Chain.headSlot chain <= At sl
-        then addBlockSt block >> return (Just (block, chain))
-        else return Nothing
-    case mblk of
-      Nothing -> return ()
-      Just (blk, chain) -> do
-        traceWith tracer (PraosNodeEventGenerate blk)
-        traceWith tracer (PraosNodeEventNewTip (chain Chain.:> blk))
-  waitForSlot sl = do
-    let tgt = slotTime slotConfig sl
-    now <- getCurrentTime
-    threadDelayNDT (tgt `diffUTCTime` now)
+  execute sched sl = lift $ do
+    wins <- sched sl
+    forM_ wins $ \_ -> do
+      let body = mkBody praosConfig prefix sl
+      let !delay = praosConfig.blockGenerationDelay $ mkPartialBlock sl body
+      let !cpuTask = CPUTask delay "Block generation"
+      curry queue cpuTask $ do
+        mblk <- atomically $ do
+          chain <- chainState <$> readTVar cpsVar
+          let block = case mkBlock chain sl body of
+                Block h b -> Block (h{headerMessageSize = praosConfig.headerSize}) b
+          if Chain.headSlot chain <= At sl
+            then addBlockSt block >> return (Just (block, chain))
+            else return Nothing
+        case mblk of
+          Nothing -> return ()
+          Just (blk, chain) -> do
+            traceWith tracer (PraosNodeEventGenerate blk)
+            traceWith tracer (PraosNodeEventNewTip (chain Chain.:> blk))

simulation/src/PraosProtocol/ChainSync.hs

@@ -152,7 +152,7 @@ data ChainConsumerState m = ChainConsumerState
   }
 
 runChainConsumer ::
-  (MonadSTM m, MonadDelay m) =>
+  MonadSTM m =>
   Tracer m (PraosNodeEvent body) ->
   PraosConfig body ->
   Chan m ChainSyncMessage ->
@@ -165,7 +165,7 @@ type ChainConsumer st m a = TC.Client ChainSyncState 'NonPipelined st m a
 
 chainConsumer ::
   forall m body.
-  (MonadSTM m, MonadDelay m) =>
+  MonadSTM m =>
   Tracer m (PraosNodeEvent body) ->
   PraosConfig body ->
   ChainConsumerState m ->

simulation/src/PraosProtocol/Common.hs

@@ -1,7 +1,9 @@
+{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeOperators #-}
+{-# LANGUAGE NoFieldSelectors #-}
 {-# OPTIONS_GHC -Wno-orphans #-}
 
 module PraosProtocol.Common (
@@ -31,22 +33,29 @@ module PraosProtocol.Common (
   defaultPraosConfig,
   CPUTask (..),
   hashToColor,
+  blockGenerator,
+  BlockGeneratorConfig (..),
+  waitNextSlot,
+  mkScheduler,
 ) where
 
 import ChanTCP (Bytes, MessageSize (..))
 import Control.Exception (assert)
+import Control.Monad.State
 import Data.Coerce (coerce)
 import Data.Default
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as Map
 import Data.Word (Word8)
+import GHC.Word (Word64)
 import Ouroboros.Network.Mock.ProducerState as ProducerState
 import PraosProtocol.Common.AnchoredFragment (Anchor (..), AnchoredFragment)
 import PraosProtocol.Common.Chain (Chain (..), foldChain, pointOnChain)
 import PraosProtocol.Common.ConcreteBlock as ConcreteBlock
+import STMCompat
 import SimTCPLinks (kilobytes)
 import SimTypes (CPUTask (..))
-import System.Random (mkStdGen, uniform)
+import System.Random (StdGen, mkStdGen, uniform, uniformR)
 import TimeCompat
 
 --------------------------------
@@ -158,3 +167,55 @@ defaultPraosConfig =
 
 instance Default (PraosConfig body) where
   def = defaultPraosConfig
+
+data BlockGeneratorConfig m = BlockGeneratorConfig
+  { slotConfig :: SlotConfig
+  , execute :: SlotNo -> StateT Int m ()
+  }
+
+blockGenerator ::
+  forall m.
+  (MonadSTM m, MonadDelay m, MonadTime m) =>
+  BlockGeneratorConfig m ->
+  m ()
+blockGenerator BlockGeneratorConfig{..} = go (0, 0)
+ where
+  go (!blkId, !tgtSlot) = do
+    slot <- waitNextSlot slotConfig tgtSlot
+    blkId' <- execStateT (execute slot) blkId
+    go (blkId', slot + 1)
+
+-- | @waitNextSlot cfg targetSlot@ waits until the beginning of
+-- @targetSlot@ if that's now or in the future, otherwise the closest slot.
+waitNextSlot :: (Monad m, MonadTime m, MonadDelay m) => SlotConfig -> SlotNo -> m SlotNo
+waitNextSlot slotConfig targetSlot = do
+  now <- getCurrentTime
+  let targetSlotTime = slotTime slotConfig targetSlot
+  let slot
+        | now <= targetSlotTime = targetSlot
+        | otherwise = assert (nextSlotIndex >= 0) $ toEnum nextSlotIndex
+       where
+        nextSlotIndex =
+          assert (slotConfig.duration == 1) $
+            ceiling $
+              now `diffUTCTime` slotConfig.start
+  let tgt = slotTime slotConfig slot
+  threadDelayNDT (tgt `diffUTCTime` now)
+  return slot
+
+mkScheduler :: MonadSTM m => StdGen -> (SlotNo -> [(a, Maybe (Double -> Word64))]) -> m (SlotNo -> m [(a, Word64)])
+mkScheduler rng0 rates = do
+  let
+    sampleRates (_role, Nothing) = return []
+    sampleRates (role, Just f) = do
+      (sample, rng') <- gets $ uniformR (0, 1)
+      put $! rng'
+      let wins = f sample
+      return [(role, wins) | wins >= 1]
+  rngVar <- newTVarIO rng0
+  let sched slot = atomically $ do
+        rng <- readTVar rngVar
+        let (acts, rng1) = flip runState rng . fmap concat . mapM sampleRates $ rates slot
+        writeTVar rngVar rng1
+        return acts
+  return sched

simulation/src/PraosProtocol/ExamplesPraosP2P.hs

@@ -30,7 +30,6 @@ import qualified LeiosProtocol.Config as OnDisk
 import Network.TypedProtocol
 import P2P (P2PTopography, P2PTopographyCharacteristics (..), genArbitraryP2PTopography, p2pNodes)
 import PraosProtocol.BlockFetch
-import PraosProtocol.BlockGeneration (PacketGenerationPattern (..))
 import PraosProtocol.Common
 import qualified PraosProtocol.Common.Chain as Chain
 import PraosProtocol.PraosNode
@@ -263,7 +262,7 @@ example1000Diffusion ::
   -- | file to write data to.
   FilePath ->
   IO ()
-example1000Diffusion rng0 cfg p2pNetwork@P2P.P2PNetwork{p2pNodes, p2pNodeStakes} stop fp
+example1000Diffusion rng0 cfg p2pNetwork@P2P.P2PNetwork{p2pNodeStakes, p2pNodeCores} stop fp
   | length (List.group (Map.elems p2pNodeStakes)) /= 1 = error "Only uniform stake distribution supported for this sim."
   | otherwise =
       runSampleModel traceFile logEvent (diffusionSampleModel (P2P.networkToTopology p2pNetwork) fp) stop $
@@ -292,50 +291,42 @@ example1000Diffusion rng0 cfg p2pNetwork@P2P.P2PNetwork{p2pNodes, p2pNodeStakes}
   logMsg ((PraosMessage _)) = Nothing
 
   traceFile = dropExtension fp <.> "log"
-  blockInterval = 1 / praosConfig.blockFrequencyPerSlot
   praosConfig = maybe defaultPraosConfig convertConfig cfg
-  p2pNumNodes = Map.size $ p2pNodes
+  stake nid = maybe undefined coerce $ Map.lookup nid p2pNodeStakes
   trace =
     tracePraosP2P
       rng0
       p2pNetwork
       (\latency -> mkTcpConnProps latency . fromMaybe (error "Only finite bandwidth supported for this sim."))
       ( \slotConfig nid rng ->
           PraosNodeConfig
-            { blockGeneration =
-                PoissonGenerationPattern
-                  rng
-                  -- average seconds between blocks:
-                  (realToFrac blockInterval * fromIntegral p2pNumNodes)
-            , praosConfig
+            { rng
+            , praosConfig = praosConfig{blockFrequencyPerSlot = praosConfig.blockFrequencyPerSlot * stake nid}
             , blockMarker = BS8.pack $ show nid ++ ": "
             , chain = Genesis
             , slotConfig
+            , processingCores = fromMaybe undefined $ Map.lookup nid p2pNodeCores
             }
       )
 
 example1Trace :: StdGen -> PraosConfig BlockBody -> P2P.P2PNetwork -> PraosTrace
-example1Trace rng0 praosConfig p2pNetwork@P2P.P2PNetwork{p2pNodes} =
+example1Trace rng0 praosConfig p2pNetwork@P2P.P2PNetwork{p2pNodeStakes, p2pNodeCores} =
   tracePraosP2P
     rng0
     p2pNetwork
     (\latency -> mkTcpConnProps latency . fromMaybe undefined)
     ( \slotConfig nid rng ->
         PraosNodeConfig
-          { blockGeneration =
-              PoissonGenerationPattern
-                rng
-                -- average seconds between blocks:
-                (realToFrac blockInterval * fromIntegral p2pNumNodes)
+          { rng
           , slotConfig
-          , praosConfig
+          , praosConfig = praosConfig{blockFrequencyPerSlot = praosConfig.blockFrequencyPerSlot * stake nid}
           , blockMarker = BS8.pack $ show nid ++ ": "
           , chain = Genesis
+          , processingCores = fromMaybe undefined $ Map.lookup nid p2pNodeCores
           }
     )
  where
-  blockInterval = 1 / praosConfig.blockFrequencyPerSlot
-  p2pNumNodes = Map.size p2pNodes
+  stake nid = maybe undefined coerce $ Map.lookup nid p2pNodeStakes
 
 example2 :: Visualization
 example2 =
@@ -401,15 +392,12 @@ example2 =
         (\latency -> mkTcpConnProps latency . fromMaybe undefined)
         ( \slotConfig nid rng ->
             PraosNodeConfig
-              { blockGeneration =
-                  PoissonGenerationPattern
-                    rng
-                    -- average seconds between blocks:
-                    (5 * fromIntegral p2pNumNodes)
-              , praosConfig = defaultPraosConfig
+              { rng
+              , praosConfig = defaultPraosConfig{blockFrequencyPerSlot = 5 / fromIntegral p2pNumNodes}
               , slotConfig
               , chain = Genesis
               , blockMarker = BS8.pack $ show nid ++ ": "
+              , processingCores = Finite 1
               }
         )
     p2pTopography =

simulation/src/PraosProtocol/PraosNode.hs

@@ -14,7 +14,9 @@ import ChanDriver (protocolMessage)
 import ChanMux
 import Control.Exception (assert)
 import Control.Monad.Class.MonadAsync (Concurrently (..), MonadAsync (..))
-import Control.Tracer (Tracer, traceWith)
+import Control.Monad.Class.MonadFork
+import Control.Monad.Class.MonadThrow
+import Control.Tracer (Tracer, contramap)
 import Data.ByteString (ByteString)
 import Data.Coerce (coerce)
 import Data.Either (fromLeft, fromRight)
@@ -28,6 +30,9 @@ import PraosProtocol.ChainSync (ChainConsumerState (..), ChainSyncMessage, chain
 import PraosProtocol.Common
 import qualified PraosProtocol.Common.Chain as Chain (Chain (..))
 import STMCompat
+import SimTypes
+import System.Random (StdGen)
+import TaskMultiQueue
 
 data Praos body f = Praos
   { protocolChainSync :: f ChainSyncMessage
@@ -135,7 +140,9 @@ runPraosNode ::
   m ()
 runPraosNode tracer cfg chain followers peers = do
   st0 <- PraosNodeState <$> newBlockFetchControllerState chain <*> pure Map.empty
-  concurrentlyMany . map runConcurrently =<< setupPraosThreads tracer cfg st0 followers peers
+  taskQueue <- atomically $ newTaskMultiQueue @() 100
+  let queue = writeTMQueue taskQueue ()
+  concurrentlyMany . map runConcurrently =<< setupPraosThreads tracer cfg queue st0 followers peers
  where
   -- Nested children threads are slow with IOSim, this impl forks them all as direct children.
   concurrentlyMany :: MonadAsync m => [m ()] -> m ()
@@ -145,16 +152,20 @@ setupPraosThreads ::
   (MonadAsync m, MonadSTM m, MonadDelay m) =>
   Tracer m (PraosNodeEvent BlockBody) ->
   PraosConfig BlockBody ->
+  ((CPUTask, m ()) -> STM m ()) ->
   PraosNodeState BlockBody m ->
   [Praos BlockBody (Chan m)] ->
   [Praos BlockBody (Chan m)] ->
   m [Concurrently m ()]
-setupPraosThreads tracer cfg st0 followers peers = do
-  (ts, f) <- BlockFetch.setupValidatorThreads tracer cfg st0.blockFetchControllerState 1 -- TODO: parameter
+setupPraosThreads tracer cfg queue st0 followers peers = do
+  (ts, f) <- BlockFetch.setupValidatorThreads cfg st0.blockFetchControllerState queue
   let valHeader h = assert (blockInvariant h) $ do
         let !delay = cfg.headerValidationDelay h
-        traceWith tracer (PraosNodeEventCPU (CPUTask delay $ T.pack $ "ValidateHeader " ++ show (coerce @_ @Int $ blockHash h)))
-        threadDelay delay
+        atomically $
+          curry
+            queue
+            (CPUTask delay $ T.pack $ "ValidateHeader " ++ show (coerce @_ @Int $ blockHash h))
+            (return ())
   (map Concurrently ts ++) <$> setupPraosThreads' tracer cfg valHeader f st0 followers peers
 
 setupPraosThreads' ::
@@ -178,32 +189,37 @@ setupPraosThreads' tracer cfg valHeader submitFetchedBlock st0 followers peers =
 data PraosNodeConfig = PraosNodeConfig
   { praosConfig :: PraosConfig BlockBody
   , slotConfig :: SlotConfig
-  , blockGeneration :: PacketGenerationPattern
   , chain :: Chain (Block BlockBody)
   , blockMarker :: ByteString
   -- ^ bytes to include in block bodies.
+  , rng :: StdGen
+  , processingCores :: NumCores
   }
 
 newPraosNodeState :: MonadSTM m => Chain (Block body) -> m (PraosNodeState body m)
 newPraosNodeState chain = PraosNodeState <$> newBlockFetchControllerState chain <*> pure Map.empty
 
 praosNode ::
-  (MonadAsync m, MonadSTM m, MonadTime m, MonadDelay m) =>
+  (MonadAsync m, MonadSTM m, MonadTime m, MonadDelay m, MonadCatch m, MonadFork m, MonadMonotonicTimeNSec m) =>
   Tracer m (PraosNodeEvent BlockBody) ->
   PraosNodeConfig ->
   [Praos BlockBody (Chan m)] ->
   [Praos BlockBody (Chan m)] ->
   m [m ()]
 praosNode tracer cfg followers peers = do
   st0 <- PraosNodeState <$> newBlockFetchControllerState cfg.chain <*> pure Map.empty
-  praosThreads <- setupPraosThreads tracer cfg.praosConfig st0 followers peers
-  nextBlock <- mkNextBlock cfg.praosConfig cfg.blockGeneration cfg.blockMarker
+  taskQueue <- atomically $ newTaskMultiQueue @() 100
+  let queue = writeTMQueue taskQueue ()
+  praosThreads <- setupPraosThreads tracer cfg.praosConfig queue st0 followers peers
+  let cpuTasksProcessors = processCPUTasks cfg.processingCores (contramap PraosNodeEventCPU tracer) taskQueue
   let generationThread =
-        blockGenerator
+        praosBlockGenerator
+          cfg.rng
           tracer
           cfg.praosConfig
           cfg.slotConfig
+          cfg.blockMarker
           st0.blockFetchControllerState.cpsVar
           (BlockFetch.addProducedBlock st0.blockFetchControllerState)
-          nextBlock
-  return $ map runConcurrently $ Concurrently generationThread : praosThreads
+          (atomically . queue)
+  return $ cpuTasksProcessors : generationThread : map runConcurrently praosThreads

simulation/src/PraosProtocol/SimBlockFetch.hs

@@ -5,6 +5,7 @@ module PraosProtocol.SimBlockFetch where
 import Chan (Chan)
 import ChanDriver (ProtocolMessage)
 import ChanTCP
+import Control.Monad
 import Control.Monad.Class.MonadAsync (
   MonadAsync (..),
   mapConcurrently_,
@@ -82,14 +83,19 @@ traceRelayLink1 tcpprops =
   nodeA praosConfig chan = do
     peerChainVar <- newTVarIO (blockHeader <$> bchain)
     (st, peerId) <- newBlockFetchControllerState Genesis >>= addPeer (asReadOnly peerChainVar)
-    (ts, submitFetchedBlock) <- setupValidatorThreads nullTracer praosConfig st 1
-    concurrently_ (mapConcurrently_ id ts) $
-      concurrently_
-        ( blockFetchController nullTracer st
-        )
-        ( runBlockFetchConsumer nullTracer praosConfig chan $
-            initBlockFetchConsumerStateForPeerId nullTracer peerId st submitFetchedBlock
-        )
+    taskTMVar <- newEmptyTMVarIO
+    let queue (_, t) = putTMVar taskTMVar t
+    let processingThread = forever $ do
+          join $ atomically $ takeTMVar taskTMVar
+    (ts, submitFetchedBlock) <- setupValidatorThreads praosConfig st queue
+    concurrently_ processingThread $
+      concurrently_ (mapConcurrently_ id ts) $
+        concurrently_
+          ( blockFetchController nullTracer st
+          )
+          ( runBlockFetchConsumer nullTracer praosConfig chan $
+              initBlockFetchConsumerStateForPeerId nullTracer peerId st submitFetchedBlock
+          )
   -- Block-Fetch Producer
   nodeB chan = do
     st <- BlockFetchProducerState . asReadOnly <$> newTVarIO (toBlocks bchain)

simulation/src/LeiosProtocol/TaskMultiQueue.hs simulation/src/TaskMultiQueue.hs

@@ -4,16 +4,20 @@
 {-# LANGUAGE RecordWildCards #-}
 {-# LANGUAGE TupleSections #-}
 
-module LeiosProtocol.TaskMultiQueue where
+module TaskMultiQueue where
 
 import Control.Monad
+import Control.Monad.Class.MonadAsync
 import Control.Monad.Class.MonadFork (MonadFork (forkIO))
+import Control.Monad.Class.MonadThrow
 import Control.Tracer
 import Data.Array
 import qualified Data.Map.Strict as Map
 import GHC.Natural
-import LeiosProtocol.Common
 import STMCompat
+import SimTypes
+import TimeCompat
+import WorkerPool
 
 type IsLabel lbl = (Ix lbl, Bounded lbl)
 
@@ -23,7 +27,7 @@ newTaskMultiQueue' :: (MonadSTM m, Ix l) => (l, l) -> Natural -> STM m (TaskMult
 newTaskMultiQueue' (a, b) n =
   TaskMultiQueue . listArray (a, b) <$> mapM (const $ newTBQueue n) (range (a, b))
 
-newTaskMultiQueue :: (MonadSTM m, IsLabel l) => Natural -> STM m (TaskMultiQueue l m)
+newTaskMultiQueue :: forall l m. (MonadSTM m, IsLabel l) => Natural -> STM m (TaskMultiQueue l m)
 newTaskMultiQueue = newTaskMultiQueue' (minBound, maxBound)
 
 writeTMQueue :: (MonadSTM m, IsLabel l) => TaskMultiQueue l m -> l -> (CPUTask, m ()) -> STM m ()
@@ -58,3 +62,22 @@ runInfParallelBlocking tracer mq = do
     forM_ (Map.toAscList tasksByEnd) $ \(end, ms) -> do
       waitUntil end
       sequence_ ms
+
+processCPUTasks ::
+  (MonadSTM m, MonadDelay m, MonadMonotonicTimeNSec m, MonadFork m, MonadAsync m, MonadCatch m, IsLabel lbl) =>
+  NumCores ->
+  Tracer m CPUTask ->
+  TaskMultiQueue lbl m ->
+  m ()
+processCPUTasks Infinite tracer queue = forever $ runInfParallelBlocking tracer queue
+processCPUTasks (Finite n) tracer queue = newBoundedWorkerPool n [taskSource l | l <- range (minBound, maxBound)]
+ where
+  taskSource l = do
+    (cpu, m) <- readTMQueue queue l
+    var <- newEmptyTMVar
+    let action = do
+          traceWith tracer cpu
+          threadDelay (cpuTaskDuration cpu)
+          m
+    -- TODO: read from var and log exception.
+    return $ Task action var