simulation: integrated unbounded parallelism worker thread

simulation/src/LeiosProtocol/Relay.hs

@@ -497,8 +497,7 @@ data SubmitPolicy = SubmitInOrder | SubmitAll
 
 data RelayConsumerConfig id header body m = RelayConsumerConfig
   { relay :: !RelayConfig
-  , headerValidationDelay :: header -> DiffTime
-  , threadDelayParallel :: [DiffTime] -> m ()
+  , validateHeaders :: [header] -> m ()
   , headerId :: !(header -> id)
   , prioritize :: !(Map id header -> [header])
   -- ^ returns a subset of headers, in order of what should be fetched first.
@@ -785,7 +784,7 @@ relayConsumerPipelined config sst =
       unless (Seq.length idsSeq <= fromIntegral windowExpand) $
         throw IdsNotRequested
 
-      config.threadDelayParallel $ map config.headerValidationDelay headers
+      config.validateHeaders headers
 
       -- Upon receiving a batch of new headers we extend our available set,
       -- and extend the unacknowledged sequence.

simulation/src/LeiosProtocol/Short/Generate.hs

@@ -92,7 +92,7 @@ data BlockGeneratorConfig m = BlockGeneratorConfig
   , nodeId :: NodeId
   , buffers :: BuffersView m
   , schedule :: SlotNo -> m [(SomeRole, Word64)]
-  , submit :: [([CPUTask], SomeAction)] -> m ()
+  , submit :: [(Maybe CPUTask, SomeAction)] -> m ()
   }
 
 blockGenerator ::
@@ -109,28 +109,28 @@ blockGenerator BlockGeneratorConfig{..} = go (0, 0)
     submit actions
     go (blkId', slot + 1)
   execute slot (SomeRole r, wins) = assert (wins >= 1) $ second (SomeAction r) <$> execute' slot r wins
-  execute' :: SlotNo -> Role a -> Word64 -> StateT Int m ([CPUTask], a)
+  execute' :: SlotNo -> Role a -> Word64 -> StateT Int m (Maybe CPUTask, a)
   execute' slot Base _wins = do
     rbData <- lift $ atomically buffers.newRBData
     let meb = rbData.freshestCertifiedEB
     let !task = CPUTask $ maybe 0 (leios.delays.certificateCreation . snd) meb
     let body = mkRankingBlockBody leios nodeId meb rbData.txsPayload
     let !rb = mkPartialBlock slot body
-    return ([task], rb)
+    return (Just task, rb)
   execute' slot Propose wins =
-    ([],) <$> do
+    (Nothing,) <$> do
       ibData <- lift $ atomically buffers.newIBData
       forM [toEnum $ fromIntegral sub | sub <- [0 .. wins - 1]] $ \sub -> do
         i <- nextBlkId InputBlockId
         let header = mkInputBlockHeader leios i slot sub nodeId ibData.referenceRankingBlock
         return $! mkInputBlock leios header ibData.txsPayload
   execute' slot Endorse _wins =
-    ([],) <$> do
+    (Nothing,) <$> do
       i <- nextBlkId EndorseBlockId
       ibs <- lift $ atomically buffers.ibs
       return $! mkEndorseBlock leios i slot nodeId $ inputBlocksToEndorse leios slot ibs
   execute' slot Vote votes =
-    ([],) <$> do
+    (Nothing,) <$> do
       votingFor <- lift $ atomically $ do
         ibs <- buffers.ibs
         ebs <- buffers.ebs