simulation: made header validation blocking in praos-diffusion

No significant change in adoption time benchmark.

Author: Saizan

simulation/src/LeiosProtocol/Common.hs

@@ -11,6 +11,7 @@
 
 module LeiosProtocol.Common (
   module PraosProtocol.Common,
+  module TaskMultiQueue,
   RankingBlockHeader,
   RankingBlockBody (..),
   RankingBlock,
@@ -49,6 +50,7 @@ import GHC.Generics
 import GHC.Records
 import PraosProtocol.Common
 import SimTypes
+import TaskMultiQueue
 
 {-
   Note [size of blocks/messages]: we add a `size` field to most

simulation/src/LeiosProtocol/Short/SimP2P.hs

@@ -125,8 +125,10 @@ exampleTrace2' rng0 leios p2pTopography@P2PNetwork{..} =
       , baseChain = Genesis
       , slotConfig
       , leios
-      , processingQueueBound = 100
-      , processingCores = fromMaybe undefined $ Map.lookup nodeId p2pNodeCores
+      , processingQueueBound = defaultQueueBound processingCores
+      , processingCores
       , nodeId
       , rng
       }
+   where
+    processingCores = fromMaybe undefined $ Map.lookup nodeId p2pNodeCores

simulation/src/PraosProtocol/PraosNode.hs

@@ -161,11 +161,10 @@ 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
-        atomically $
-          curry
-            queue
-            (CPUTask delay $ T.pack $ "ValidateHeader " ++ show (coerce @_ @Int $ blockHash h))
-            (return ())
+        curry
+          (queueAndWait (atomically . queue))
+          (CPUTask delay $ T.pack $ "ValidateHeader " ++ show (coerce @_ @Int $ blockHash h))
+          (return ())
   (map Concurrently ts ++) <$> setupPraosThreads' tracer cfg valHeader f st0 followers peers
 
 setupPraosThreads' ::
@@ -208,7 +207,7 @@ praosNode ::
   m [m ()]
 praosNode tracer cfg followers peers = do
   st0 <- PraosNodeState <$> newBlockFetchControllerState cfg.chain <*> pure Map.empty
-  taskQueue <- atomically $ newTaskMultiQueue @() 100
+  taskQueue <- atomically $ newTaskMultiQueue @() (defaultQueueBound cfg.processingCores)
   let queue = writeTMQueue taskQueue ()
   praosThreads <- setupPraosThreads tracer cfg.praosConfig queue st0 followers peers
   let cpuTasksProcessors = processCPUTasks cfg.processingCores (contramap PraosNodeEventCPU tracer) taskQueue
@@ -221,5 +220,17 @@ praosNode tracer cfg followers peers = do
           cfg.blockMarker
           st0.blockFetchControllerState.cpsVar
           (BlockFetch.addProducedBlock st0.blockFetchControllerState)
-          (atomically . queue)
+          (queueAndWait (atomically . queue))
   return $ cpuTasksProcessors : generationThread : map runConcurrently praosThreads
+
+queueAndWait ::
+  MonadSTM m =>
+  ((a, m ()) -> m ()) ->
+  (a, m ()) ->
+  m ()
+queueAndWait queue (d, m) = do
+  sem <- newEmptyTMVarIO
+  curry queue d $ do
+    m
+    atomically $ putTMVar sem ()
+  atomically $ takeTMVar sem

simulation/src/TaskMultiQueue.hs

@@ -81,3 +81,9 @@ processCPUTasks (Finite n) tracer queue = newBoundedWorkerPool n [taskSource l |
           m
     -- TODO: read from var and log exception.
     return $ Task action var
+
+defaultQueueBound :: NumCores -> Natural
+defaultQueueBound processingCores = do
+  fromIntegral $ case processingCores of
+    Infinite -> 100
+    Finite n -> n * 2