block-fetch: test polite protocol termination

ouroboros-network/test/Ouroboros/Network/BlockFetch/Examples.hs

@@ -6,6 +6,7 @@
 {-# LANGUAGE TypeFamilies        #-}
 
 module Ouroboros.Network.BlockFetch.Examples (
+    blockFetchExample0,
     blockFetchExample1,
     mockBlockFetchServer1,
     exampleFixedPeerGSVs,
@@ -57,6 +58,91 @@ import           Ouroboros.Network.BlockFetch.Client
 import           Ouroboros.Network.Testing.ConcreteBlock
 
 
+-- | Run a single block fetch protocol until the chain is downloaded.
+--
+blockFetchExample0 :: forall m.
+                      (MonadSTM m, MonadST m, MonadAsync m, MonadFork m,
+                       MonadTime m, MonadTimer m, MonadMask m, MonadThrow (STM m))
+                   => Tracer m [TraceLabelPeer Int
+                                 (FetchDecision [Point BlockHeader])]
+                   -> Tracer m (TraceLabelPeer Int
+                                 (TraceFetchClientState BlockHeader))
+                   -> Tracer m (TraceLabelPeer Int
+                                 (TraceSendRecv (BlockFetch Block)))
+                   -> Maybe DiffTime -- ^ client's channel delay
+                   -> Maybe DiffTime -- ^ servers's channel delay
+                   -> ControlMessageSTM m
+                   -> AnchoredFragment Block -- ^ Fixed current chain
+                   -> AnchoredFragment Block -- ^ Fixed candidate chain
+                   -> m ()
+blockFetchExample0 decisionTracer clientStateTracer clientMsgTracer
+                   clientDelay serverDelay
+                   controlMessageSTM
+                   currentChain candidateChain = do
+
+    registry    <- newFetchClientRegistry
+    blockHeap   <- mkTestFetchedBlockHeap (anchoredChainPoints currentChain)
+
+    (clientAsync, serverAsync, syncClientAsync, keepAliveAsync)
+                <- runFetchClientAndServerAsync
+                    (contramap (TraceLabelPeer peerno) clientMsgTracer)
+                    (contramap (TraceLabelPeer peerno) serverMsgTracer)
+                    clientDelay serverDelay
+                    registry peerno
+                    (blockFetchClient NodeToNodeV_1 controlMessageSTM)
+                    (mockBlockFetchServer1 (unanchorFragment candidateChain))
+
+    fetchAsync  <- async $ blockFetch registry blockHeap
+    driverAsync <- async $ driver blockHeap
+
+    -- Order of shutdown here is important for this example: must kill off the
+    -- fetch thread before the peer threads.
+    _ <- waitAnyCancel $ [ fetchAsync, driverAsync,
+                           clientAsync, serverAsync,
+                           syncClientAsync, keepAliveAsync]
+    return ()
+
+  where
+    peerno = 1 :: Int
+
+    serverMsgTracer = nullTracer
+
+    currentChainHeaders =
+      AnchoredFragment.mapAnchoredFragment blockHeader currentChain
+
+    candidateChainHeaders =
+      Map.fromList $ zip [1..] $
+      map (AnchoredFragment.mapAnchoredFragment blockHeader) [candidateChain]
+
+    anchoredChainPoints c = anchorPoint c
+                          : map blockPoint (AnchoredFragment.toOldestFirst c)
+
+    blockFetch :: FetchClientRegistry Int BlockHeader Block m
+               -> TestFetchedBlockHeap m Block
+               -> m ()
+    blockFetch registry blockHeap =
+        blockFetchLogic
+          decisionTracer clientStateTracer
+          (sampleBlockFetchPolicy1 blockHeap currentChainHeaders candidateChainHeaders)
+          registry
+          (BlockFetchConfiguration {
+            bfcMaxConcurrencyBulkSync = 1,
+            bfcMaxConcurrencyDeadline = 2,
+            bfcMaxRequestsInflight    = 10,
+            bfcDecisionLoopInterval   = 0.01,
+            bfcSalt                   = 0
+          })
+        >> return ()
+
+    driver :: TestFetchedBlockHeap m Block -> m ()
+    driver blockHeap = do
+      atomically $ do
+        heap <- getTestFetchedBlocks blockHeap
+        check $
+          all (\c -> AnchoredFragment.headPoint c `Set.member` heap)
+              [candidateChain]
+
+
 --
 -- Sample setups of block fetch logic with fetch clients and peers
 --

ouroboros-network/test/Test/Ouroboros/Network/BlockFetch.hs

@@ -21,7 +21,7 @@ import           Data.Set (Set)
 import qualified Data.Set as Set
 import           Data.Typeable (Typeable)
 
-import           Control.Exception (AssertionFailed (..))
+import           Control.Exception (AssertionFailed (..), throw)
 import           Control.Monad (unless)
 import           Control.Monad.Class.MonadAsync
 import           Control.Monad.Class.MonadFork
@@ -41,10 +41,12 @@ import           Ouroboros.Network.BlockFetch.ClientRegistry
 import           Ouroboros.Network.BlockFetch.ClientState
 import           Ouroboros.Network.BlockFetch.Examples
 import qualified Ouroboros.Network.MockChain.Chain as Chain
-import           Ouroboros.Network.Mux (continueForever)
+import           Ouroboros.Network.Mux (ControlMessage (..), continueForever)
 import           Ouroboros.Network.Protocol.BlockFetch.Type (BlockFetch)
 import           Ouroboros.Network.Testing.ConcreteBlock
 
+import           Test.Ouroboros.Network.Utils
+
 
 --
 -- The list of all tests
@@ -64,6 +66,8 @@ tests = testGroup "BlockFetch"
   --TODO: test where for any given delta-Q, check that we do achieve full
   -- pipelining to keep the server busy and get decent enough batching of
   -- requests (testing the high/low watermark mechanism).
+  , testProperty "termination"
+                 prop_terminate
   ]
 
 
@@ -607,6 +611,50 @@ _unit_bracketSyncWithFetchClient step = do
                 _             -> return ()
             return res
 
+-- | Check that the client can terminate using `ControlMessage` mechanism.
+--
+-- The 'awaitDelay' of @100 * delay@ is a bit arbitrary.  It would be nicer to
+-- make a proper calucation what should it be.  At the moment this test shows
+-- that the block fetch protocol can exit within some large time limit.
+--
+prop_terminate :: TestChainFork -> Delay -> Property
+prop_terminate (TestChainFork _commonChain forkChain _forkChain) (Delay delay) =
+    let tr = runSimTrace simulation
+        trace :: [FetchRequestTrace]
+        trace  = selectTraceEventsDynamic tr
+    in counterexample
+        ("Trace: \n" ++ unlines (map show trace))
+        (case traceResult True tr of
+           Left e  -> throw e
+           Right x -> counterexample "block-fetch was unstoppable" x)
+  where
+    simulation :: forall s. IOSim s Bool
+    simulation = do
+      controlMessageVar <- newTVarIO Continue
+      result <-
+        race
+          (do
+            let terminateDelay =
+                  realToFrac (Chain.length forkChain) * delay / 2
+            threadDelay terminateDelay
+            atomically (writeTVar controlMessageVar Terminate)
+            let awaitDelay = delay * 100
+            threadDelay awaitDelay)
+          (blockFetchExample0
+            (contramap TraceFetchDecision       dynamicTracer)
+            (contramap TraceFetchClientState    dynamicTracer)
+            (contramap TraceFetchClientSendRecv dynamicTracer)
+            (Just delay) (Just delay)
+            (readTVar controlMessageVar)
+            (AnchoredFragment.Empty AnchoredFragment.AnchorGenesis)
+            fork')
+      return $ case result of
+        Left _ -> False
+        Right _ -> True
+
+    fork'  = chainToAnchoredFragment forkChain
+
+
 
 --
 -- Trace utils