Improve protocol stopping in smoke tests for ObjectDiffusion

Author: tbagrel1

ouroboros-consensus/src/ouroboros-consensus/Ouroboros/Consensus/MiniProtocol/ObjectDiffusion/ObjectPool/PerasCert.hs

@@ -1,4 +1,3 @@
-{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE StandaloneDeriving #-}
 

ouroboros-consensus/test/consensus-test/Test/Consensus/MiniProtocol/ObjectDiffusion/Smoke.hs

@@ -5,6 +5,7 @@
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE GeneralizedNewtypeDeriving #-}
 {-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
@@ -21,8 +22,9 @@ module Test.Consensus.MiniProtocol.ObjectDiffusion.Smoke
   , prop_smoke_object_diffusion
   ) where
 
+import Control.Monad (void)
 import Control.Monad.IOSim (runSimStrictShutdown)
-import Control.ResourceRegistry (forkLinkedThread, waitAnyThread, withRegistry)
+import Control.ResourceRegistry (forkLinkedThread, withRegistry)
 import Control.Tracer (Tracer, nullTracer, traceWith)
 import Data.Containers.ListUtils (nubOrdOn)
 import Data.Data (Typeable)
@@ -257,9 +259,27 @@ prop_smoke_object_diffusion
         let tracer = nullTracer
 
         traceWith tracer "========== [ Starting ObjectDiffusion smoke test ] =========="
+        traceWith tracer "objects: "
         traceWith tracer (show objects)
 
         (outboundPoolReader, inboundPoolWriter, getAllInboundPoolContent) <- mkPoolInterfaces
+
+        -- We wait until the inbound pool content stabilizes
+        -- Caveat: in the case where objects are continuously added to the
+        -- outbound pool, this may never terminate.
+        let waitUntilSettlement prevValue = do
+              -- TODO: should have a delay value equal to 4·Δ + Ɛ
+              -- were Δ is the delay in which any message is delivered on the
+              -- network and Ɛ is a small margin to encompass computation time;
+              -- as in the worst case, we need 4 echanged messages
+              -- (+ computation time, assumed negligible w.r.t. network delays)
+              -- to see a state update on the inbound side
+              threadDelay 10_000
+              newValue <- getAllInboundPoolContent
+              if newValue == prevValue
+                then pure ()
+                else waitUntilSettlement newValue
+
         controlMessage <- uncheckedNewTVarM Continue
 
         let
@@ -283,22 +303,20 @@ prop_smoke_object_diffusion
 
         withRegistry $ \reg -> do
           (outboundChannel, inboundChannel) <- createConnectedChannels
-          outboundThread <-
+          _outboundThread <-
             forkLinkedThread reg "ObjectDiffusion Outbound peer thread" $
               runOutboundPeer outbound outboundChannel tracer
-          inboundThread <-
+          _inboundThread <-
             forkLinkedThread reg "ObjectDiffusion Inbound peer thread" $
               runInboundPeer inbound inboundChannel tracer
-          controlMessageThread <- forkLinkedThread reg "ObjectDiffusion Control thread" $ do
-            threadDelay 1000 -- give a head start to the other threads
-            atomically $ writeTVar controlMessage Terminate
-            threadDelay 1000 -- wait for the other threads to finish
-          waitAnyThread [outboundThread, inboundThread, controlMessageThread]
+
+          void $ waitUntilSettlement []
+          atomically $ writeTVar controlMessage Terminate
 
         traceWith tracer "========== [ ObjectDiffusion smoke test finished ] =========="
-        poolContent <- getAllInboundPoolContent
 
-        traceWith tracer "inboundPoolContent:"
+        poolContent <- getAllInboundPoolContent
+        traceWith tracer "inboundPoolContent: "
         traceWith tracer (show poolContent)
         traceWith tracer "========== ======================================= =========="
         pure poolContent