pass resources through

Author: isovector

ouroboros-consensus-diffusion/app/conformance-test-runner/Main.hs

@@ -1,5 +1,9 @@
 module Main (main) where
 
+import qualified Data.Map.Merge.Lazy as M
+import Test.Consensus.PeerSimulator.Resources (PeerSimulatorResources(..), makePeerSimulatorResources)
+import Control.Tracer (nullTracer)
+import qualified Data.List.NonEmpty as NonEmpty
 import Data.Aeson (encode, throwDecode)
 import qualified Data.ByteString.Lazy.Char8 as BSL8
 import Data.Coerce
@@ -77,29 +81,34 @@ main = do
   let simPeerMap = buildPeerMap (optPort opts) pointSchedule
   BSL8.writeFile (optOutputTopologyFile opts) (encode $ makeTopology simPeerMap)
 
+zipMaps :: Ord k => Map k a -> Map k b -> Map k (a, b)
+zipMaps = M.merge M.dropMissing M.dropMissing $ M.zipWithMatched $ const (,)
+
 runServer :: IO ()
 runServer = do
   let peerMap = buildPeerMap 6001 testPointSchedule
 
+  peerSim <- makePeerSimulatorResources nullTracer undefined $ NonEmpty.fromList $ M.keys peerMap
+
   peerServers <-
-    for peerMap $ \port -> do
+    for (zipMaps peerMap $ psrPeers peerSim) $ \(port, res) -> do
       -- Make a TMVar for the chainsync and blockfetch channels exposed through
       -- the miniprotocols. These get threaded into the server, which will fill
       -- them once the NUT has connected.
-      csChannelTMV <- newEmptyTMVarIO
-      bfChannelTMV <- newEmptyTMVarIO
+      csChannelTMV <- newTVarIO False
+      bfChannelTMV <- newTVarIO False
 
       putStrLn $ "starting server on " <> show port
       let sockAddr = Socket.SockAddrInet port $ Socket.tupleToHostAddress (127, 0, 0, 1)
-      thread <- async $ run csChannelTMV bfChannelTMV sockAddr
+      thread <- async $ run res csChannelTMV bfChannelTMV sockAddr
       pure ((csChannelTMV, bfChannelTMV), thread)
 
   -- Now, take each of the resulting TMVars. This effectively blocks until the
   -- NUT has connected.
   _peerChannels <- atomically $ do
     for peerServers $ \((csChanTMV, bfChanTMV), _thread) -> do
-      csChan <- readTMVar csChanTMV
-      bfChan <- readTMVar bfChanTMV
+      csChan <- readTVar csChanTMV
+      bfChan <- readTVar bfChanTMV
       pure (csChan, bfChan)
 
   for_ peerServers $ uninterruptibleCancel . snd

ouroboros-consensus-diffusion/app/conformance-test-runner/MiniProtocols.hs

@@ -16,6 +16,10 @@
 -- picked up by the peer simulator.
 module MiniProtocols (peerSimServer) where
 
+import Ouroboros.Network.Protocol.BlockFetch.Server
+import Ouroboros.Network.Util.ShowProxy (ShowProxy)
+import Ouroboros.Network.Protocol.ChainSync.Server
+import Test.Consensus.PeerSimulator.Resources (PeerResources (..), ChainSyncResources (..), BlockFetchResources (..))
 import qualified Codec.CBOR.Decoding as CBOR
 import qualified Codec.CBOR.Encoding as CBOR
 import Control.Monad (forever)
@@ -59,10 +63,13 @@ peerSimServer ::
   ( IOLike m
   , SerialiseNodeToNodeConstraints blk
   , SupportedNetworkProtocolVersion blk
+  , ShowProxy blk
+  , ShowProxy (Header blk)
   , MonadSay m
   ) =>
-  StrictTMVar m (Mux.Channel m BL.ByteString) ->
-  StrictTMVar m (Mux.Channel m BL.ByteString) ->
+  PeerResources m blk ->
+  StrictTVar m Bool ->
+  StrictTVar m Bool ->
   CodecConfig blk ->
   (NodeToNodeVersion -> addr -> CBOR.Encoding) ->
   (NodeToNodeVersion -> forall s. CBOR.Decoder s addr) ->
@@ -71,7 +78,7 @@ peerSimServer ::
     NodeToNodeVersion
     NodeToNodeVersionData
     (OuroborosApplicationWithMinimalCtx 'Mux.ResponderMode addr BL.ByteString m Void ())
-peerSimServer csChanTMV bfChanTMV codecCfg encAddr decAddr networkMagic = do
+peerSimServer res csChanTMV bfChanTMV codecCfg encAddr decAddr networkMagic = do
   forAllVersions application
  where
   forAllVersions ::
@@ -115,19 +122,19 @@ peerSimServer csChanTMV bfChanTMV codecCfg encAddr decAddr networkMagic = do
           $ MiniProtocolCb
           $ \_ctx channel -> do
             say "hello from cs"
-            atomically $
-              putTMVar csChanTMV channel
-            pure ((), Nothing)
+            atomically $ writeTVar csChanTMV True
+            runPeer nullTracer cChainSyncCodec channel
+              $ chainSyncServerPeer $ csrServer $ prChainSync res
       , mkMiniProtocol
           Mux.StartOnDemand
           N2N.blockFetchMiniProtocolNum
           N2N.blockFetchProtocolLimits
           $ MiniProtocolCb
           $ \_ctx channel -> do
             say "hello from bf"
-            atomically $
-              putTMVar bfChanTMV channel
-            pure ((), Nothing)
+            atomically $ writeTVar bfChanTMV True
+            runPeer nullTracer cBlockFetchCodec channel
+              $ blockFetchServerPeer $ bfrServer $ prBlockFetch res
       , mkMiniProtocol
           Mux.StartOnDemand
           N2N.txSubmissionMiniProtocolNum
@@ -138,8 +145,8 @@ peerSimServer csChanTMV bfChanTMV codecCfg encAddr decAddr networkMagic = do
      where
       Consensus.N2N.Codecs
         { cKeepAliveCodec
-        -- , cChainSyncCodecSerialised
-        -- , cBlockFetchCodecSerialised
+        , cChainSyncCodec
+        , cBlockFetchCodec
         } =
           Consensus.N2N.defaultCodecs codecCfg blockVersion encAddr decAddr version
 

ouroboros-consensus-diffusion/app/conformance-test-runner/Server.hs

@@ -7,6 +7,7 @@
 
 module Server (run) where
 
+import Test.Consensus.PeerSimulator.Resources (PeerResources)
 import Control.ResourceRegistry
 import Control.Tracer
 import qualified Data.ByteString.Lazy as BL
@@ -68,15 +69,17 @@ run ::
   , ConfigSupportsNode blk
   , blk ~ SimpleBlock SimpleMockCrypto SimplePraosRuleExt
   ) =>
+  PeerResources IO blk ->
   -- | A TMVar for the chainsync channel that we will fill in once the node connects.
-  StrictTMVar IO (Mux.Channel IO BL.ByteString) ->
+  StrictTVar IO Bool ->
   -- | A TMVar for the blockfetch channel that we will fill in once the node connects.
-  StrictTMVar IO (Mux.Channel IO BL.ByteString) ->
+  StrictTVar IO Bool ->
   SockAddr ->
   IO Void
-run csChanTMV bfChanTMV sockAddr = withRegistry \_registry ->
+run res csChanTMV bfChanTMV sockAddr = withRegistry \_registry ->
   serve sockAddr
     $ peerSimServer @_ @(SimpleBlock SimpleMockCrypto SimplePraosRuleExt)
+      res
       csChanTMV
       bfChanTMV
       SimpleCodecConfig