TX Submission Logic

network-mux/demo/mux-demo.hs

@@ -133,7 +133,7 @@ serverWorker bearer = do
       putStrLn $ "Result: " ++ show result
       Mx.stop mux
 
-    Mx.run nullTracer mux bearer
+    Mx.run nullTracer 1 mux bearer
   where
     ptcls :: [MiniProtocolInfo ResponderMode]
     ptcls = [ MiniProtocolInfo {
@@ -192,7 +192,7 @@ clientWorker bearer n msg = do
       putStrLn $ "Result: " ++ show result
       Mx.stop mux
 
-    Mx.run nullTracer mux bearer
+    Mx.run nullTracer 0 mux bearer
   where
     ptcls :: [MiniProtocolInfo Mx.InitiatorMode]
     ptcls = [ MiniProtocolInfo {

network-mux/src/Control/Concurrent/JobPool.hs

@@ -12,6 +12,7 @@ module Control.Concurrent.JobPool
   , Job (..)
   , withJobPool
   , forkJob
+  , forkJobOn
   , readSize
   , readGroupSize
   , waitForJob
@@ -29,6 +30,9 @@ import Control.Monad.Class.MonadAsync
 import Control.Monad.Class.MonadFork (MonadThread (..))
 import Control.Monad.Class.MonadThrow
 
+import Control.Concurrent (getNumCapabilities)
+import System.IO.Unsafe (unsafePerformIO)
+
 -- | JobPool allows to submit asynchronous jobs, wait for their completion or
 -- cancel.  Jobs are grouped, each group can be cancelled separately.
 --
@@ -69,16 +73,18 @@ withJobPool =
       jobs <- readTVarIO jobsVar
       mapM_ uninterruptibleCancel jobs
 
-forkJob :: forall group m a.
-           ( MonadAsync m, MonadMask m
-           , Ord group
-           )
-        => JobPool group m a
-        -> Job     group m a
-        -> m ()
-forkJob JobPool{jobsVar, completionQueue} (Job action handler group label) =
+
+forkJob' :: forall group m a.
+            ( MonadAsync m, MonadMask m
+            , Ord group
+            )
+         => (m () -> m (Async m ()))
+         -> JobPool group m a
+         -> Job     group m a
+         -> m ()
+forkJob' doFork JobPool{jobsVar, completionQueue} (Job action handler group label) =
     mask $ \restore -> do
-      jobAsync <- async $ do
+      jobAsync <- doFork $ do
         tid <- myThreadId
         io tid restore
           `onException`
@@ -104,6 +110,35 @@ forkJob JobPool{jobsVar, completionQueue} (Job action handler group label) =
               restore action
       atomically $ writeTQueue completionQueue res
 
+
+
+forkJob :: forall group m a.
+           ( MonadAsync m, MonadMask m
+           , Ord group
+           )
+        => JobPool group m a
+        -> Job     group m a
+        -> m ()
+forkJob = forkJob' async
+
+
+forkJobOn :: forall group m a.
+             ( MonadAsync m, MonadMask m
+             , Ord group
+             )
+          => Int
+          -> JobPool group m a
+          -> Job     group m a
+          -> m ()
+forkJobOn c = forkJob' (asyncOn limitCapability)
+  where
+    limitCapability :: Int
+    limitCapability =
+      -- TODO: add `getNumCapabilities` to `MonadFork`
+      let sysCap = unsafePerformIO getNumCapabilities in
+      c `mod` (max 1 $ sysCap - 2)
+
+
 readSize :: MonadSTM m => JobPool group m a -> STM m Int
 readSize JobPool{jobsVar} = Map.size <$> readTVar jobsVar
 

network-mux/src/Network/Mux.hs

@@ -211,10 +211,11 @@ run :: forall m mode.
        , MonadMask m
        )
     => Tracer m Trace
+    -> Int
     -> Mux mode m
     -> Bearer m
     -> m ()
-run tracer Mux {muxMiniProtocols, muxControlCmdQueue, muxStatus} bearer@Bearer {name} = do
+run tracer peerHash Mux {muxMiniProtocols, muxControlCmdQueue, muxStatus} bearer@Bearer{name} = do
     egressQueue <- atomically $ newTBQueue 100
 
     -- label shared variables
@@ -231,7 +232,8 @@ run tracer Mux {muxMiniProtocols, muxControlCmdQueue, muxStatus} bearer@Bearer {
         -- Wait for someone to shut us down by calling muxStop or an error.
         -- Outstanding jobs are shut down Upon completion of withJobPool.
         withTimeoutSerial $ \timeout ->
-          monitor tracer
+          monitor peerHash
+                  tracer
                   timeout
                   jobpool
                   egressQueue
@@ -375,14 +377,15 @@ monitor :: forall mode m.
            , Alternative (STM m)
            , MonadThrow (STM m)
            )
-        => Tracer m Trace
+        => Int
+        -> Tracer m Trace
         -> TimeoutFn m
         -> JobPool.JobPool Group m JobResult
         -> EgressQueue m
         -> StrictTQueue m (ControlCmd mode m)
         -> StrictTVar m Status
         -> m ()
-monitor tracer timeout jobpool egressQueue cmdQueue muxStatus =
+monitor peerHash tracer timeout jobpool egressQueue cmdQueue muxStatus =
     go (MonitorCtx Map.empty)
   where
     go :: MonitorCtx m mode -> m ()
@@ -451,7 +454,7 @@ monitor tracer timeout jobpool egressQueue cmdQueue muxStatus =
                            ptclAction) -> do
           traceWith tracer (TraceStartEagerly miniProtocolNum
                              (protocolDirEnum miniProtocolDir))
-          JobPool.forkJob jobpool $
+          JobPool.forkJobOn peerHash jobpool $
             miniProtocolJob
               tracer
               egressQueue

network-mux/test/Test/Mux.hs

@@ -358,8 +358,8 @@ prop_mux_snd_recv (DummyRun messages) = ioProperty $ do
 
     serverMux <- Mx.new [serverApp]
 
-    withAsync (Mx.run clientTracer clientMux clientBearer) $ \clientAsync ->
-      withAsync (Mx.run serverTracer serverMux serverBearer) $ \serverAsync -> do
+    withAsync (Mx.run clientTracer 0 clientMux clientBearer) $ \clientAsync ->
+      withAsync (Mx.run serverTracer 0 serverMux serverBearer) $ \serverAsync -> do
 
         r <- step clientMux clientApp serverMux serverApp messages
         Mx.stop serverMux
@@ -434,10 +434,10 @@ prop_mux_snd_recv_bi (DummyRun messages) = ioProperty $ do
 
 
     clientMux <- Mx.new clientApps
-    clientAsync <- async $ Mx.run clientTracer clientMux clientBearer
+    clientAsync <- async $ Mx.run clientTracer 0 clientMux clientBearer
 
     serverMux <- Mx.new serverApps
-    serverAsync <- async $ Mx.run serverTracer serverMux serverBearer
+    serverAsync <- async $ Mx.run serverTracer 1 serverMux serverBearer
 
     r <- step clientMux clientApps serverMux serverApps messages
     Mx.stop clientMux
@@ -541,7 +541,7 @@ prop_mux_snd_recv_compat messages = ioProperty $ do
         )
 
       -- Wait for the first MuxApplication to finish, then stop the mux.
-      withAsync (Mx.run clientTracer clientMux clientBearer) $ \aid -> do
+      withAsync (Mx.run clientTracer 0 clientMux clientBearer) $ \aid -> do
         _ <- atomically res
         Mx.stop clientMux
         wait aid
@@ -559,7 +559,7 @@ prop_mux_snd_recv_compat messages = ioProperty $ do
         )
 
       -- Wait for the first MuxApplication to finish, then stop the mux.
-      withAsync (Mx.run serverTracer serverMux serverBearer) $ \aid -> do
+      withAsync (Mx.run serverTracer 1 serverMux serverBearer) $ \aid -> do
         _ <- atomically res
         Mx.stop serverMux
         wait aid
@@ -719,7 +719,7 @@ runMuxApplication initApps initBearer respApps respBearer = do
     respMux <- Mx.new $ map (\(pn,_) ->
         MiniProtocolInfo (Mx.MiniProtocolNum pn) Mx.ResponderDirectionOnly defaultMiniProtocolLimits)
         respApps'
-    respAsync <- async $ Mx.run serverTracer respMux respBearer
+    respAsync <- async $ Mx.run serverTracer 1 respMux respBearer
     getRespRes <- sequence [ Mx.runMiniProtocol
                               respMux
                               (Mx.MiniProtocolNum pn)
@@ -732,7 +732,7 @@ runMuxApplication initApps initBearer respApps respBearer = do
     initMux <- Mx.new $ map (\(pn,_) ->
         MiniProtocolInfo (Mx.MiniProtocolNum pn) Mx.InitiatorDirectionOnly defaultMiniProtocolLimits)
         initApps'
-    initAsync <- async $ Mx.run clientTracer initMux initBearer
+    initAsync <- async $ Mx.run clientTracer 0 initMux initBearer
     getInitRes <- sequence [ Mx.runMiniProtocol
                               initMux
                               (Mx.MiniProtocolNum pn)
@@ -952,17 +952,17 @@ prop_mux_starvation (Uneven response0 response1) =
                        }
 
     serverMux <- Mx.new [serverApp2, serverApp3]
-    serverMux_aid <- async $ Mx.run serverTracer serverMux serverBearer
+    serverMux_aid <- async $ Mx.run serverTracer 0 serverMux serverBearer
     serverRes2 <- Mx.runMiniProtocol serverMux (miniProtocolNum serverApp2) (miniProtocolDir serverApp2)
                    Mx.StartOnDemand server_short
     serverRes3 <- Mx.runMiniProtocol serverMux (miniProtocolNum serverApp3) (miniProtocolDir serverApp3)
                    Mx.StartOnDemand server_long
 
     clientMux <- Mx.new [clientApp2, clientApp3]
-    clientMux_aid <- async $ Mx.run (clientTracer <> headerTracer) clientMux clientBearer
-    clientRes2 <- Mx.runMiniProtocol clientMux (Mx.miniProtocolNum clientApp2) (Mx.miniProtocolDir clientApp2)
+    clientMux_aid <- async $ Mx.run (clientTracer <> headerTracer) 1 clientMux clientBearer
+    clientRes2 <- Mx.runMiniProtocol clientMux (miniProtocolNum clientApp2) (miniProtocolDir clientApp2)
                    Mx.StartEagerly client_short
-    clientRes3 <- Mx.runMiniProtocol clientMux (Mx.miniProtocolNum clientApp3) (Mx.miniProtocolDir clientApp3)
+    clientRes3 <- Mx.runMiniProtocol clientMux (miniProtocolNum clientApp3) (miniProtocolDir clientApp3)
                    Mx.StartEagerly client_long
 
 
@@ -1157,7 +1157,7 @@ prop_demux_sdu a = do
         serverRes <- Mx.runMiniProtocol serverMux (Mx.miniProtocolNum serverApp) (Mx.miniProtocolDir serverApp)
                  Mx.StartEagerly server_mp
 
-        said <- async $ Mx.run serverTracer serverMux serverBearer
+        said <- async $ Mx.run serverTracer 1 serverMux serverBearer
         return (server_r, said, serverRes, serverMux)
 
     -- Server that expects to receive a specific ByteString.
@@ -1432,7 +1432,7 @@ prop_mux_restart_m (DummyRestartingInitiatorApps apps) = do
     let minis = map (appToInfo Mx.InitiatorDirectionOnly . fst) apps
 
     mux <- Mx.new minis
-    mux_aid <- async $ Mx.run nullTracer mux bearer
+    mux_aid <- async $ Mx.run nullTracer 0 mux bearer
     getRes <- sequence [ Mx.runMiniProtocol
                            mux
                           (daNum $ fst app)
@@ -1479,7 +1479,7 @@ prop_mux_restart_m (DummyRestartingResponderApps rapps) = do
         minis = map (appToInfo Mx.ResponderDirectionOnly) apps
 
     mux <- Mx.new minis
-    mux_aid <- async $ Mx.run nullTracer mux bearer
+    mux_aid <- async $ Mx.run nullTracer 1 mux bearer
     getRes <- sequence [ Mx.runMiniProtocol
                            mux
                           (daNum $ fst app)
@@ -1528,7 +1528,7 @@ prop_mux_restart_m (DummyRestartingInitiatorResponderApps rapps) = do
         respMinis = map (appToInfo Mx.ResponderDirection) apps
 
     mux <- Mx.new $ initMinis ++ respMinis
-    mux_aid <- async $ Mx.run nullTracer mux bearer
+    mux_aid <- async $ Mx.run nullTracer 1 mux bearer
     getInitRes <- sequence [ Mx.runMiniProtocol
                                mux
                                (daNum $ fst app)
@@ -1603,7 +1603,7 @@ prop_mux_start_m bearer _ checkRes (DummyInitiatorApps apps) runTime = do
         minRunTime = minimum $ runTime : (map daRunTime $ filter (\app -> daAction app == DummyAppFail) apps)
 
     mux <- Mx.new minis
-    mux_aid <- async $ Mx.run nullTracer mux bearer
+    mux_aid <- async $ Mx.run nullTracer 0 mux bearer
     killer <- async $ (threadDelay runTime) >> Mx.stop mux
     getRes <- sequence [ Mx.runMiniProtocol
                            mux
@@ -1624,7 +1624,7 @@ prop_mux_start_m bearer trigger checkRes (DummyResponderApps apps) runTime = do
         minRunTime = minimum $ runTime : (map (\a -> daRunTime a + daStartAfter a) $ filter (\app -> daAction app == DummyAppFail) apps)
 
     mux <- Mx.new minis
-    mux_aid <- async $ Mx.run verboseTracer mux bearer
+    mux_aid <- async $ Mx.run verboseTracer 0 mux bearer
     getRes <- sequence [ Mx.runMiniProtocol
                            mux
                           (daNum app)
@@ -1650,7 +1650,7 @@ prop_mux_start_m bearer _trigger _checkRes (DummyResponderAppsKillMux apps) runT
     let minis = map (appToInfo Mx.ResponderDirectionOnly) apps
 
     mux <- Mx.new minis
-    mux_aid <- async $ Mx.run verboseTracer mux bearer
+    mux_aid <- async $ Mx.run verboseTracer 1 mux bearer
     getRes <- sequence [ Mx.runMiniProtocol
                            mux
                           (daNum app)
@@ -1673,7 +1673,7 @@ prop_mux_start_m bearer trigger checkRes (DummyInitiatorResponderApps apps) runT
         minRunTime = minimum $ runTime : (map (\a -> daRunTime a) $ filter (\app -> daAction app == DummyAppFail) apps)
 
     mux <- Mx.new $ initMinis ++ respMinis
-    mux_aid <- async $ Mx.run verboseTracer mux bearer
+    mux_aid <- async $ Mx.run verboseTracer 0 mux bearer
     getInitRes <- sequence [ Mx.runMiniProtocol
                                mux
                                (daNum app)
@@ -1835,7 +1835,7 @@ close_experiment
                        ])
                 Mx.stop $ \mux ->
         withNetworkCtx clientCtx $ \clientBearer ->
-          withAsync (Mx.run ((Client,) `contramap` muxTracer) mux clientBearer) $ \_muxAsync ->
+          withAsync (Mx.run ((Client,) `contramap` muxTracer) 0 mux clientBearer) $ \_muxAsync ->
                 Mx.runMiniProtocol
                   mux miniProtocolNum
                   Mx.InitiatorDirectionOnly Mx.StartEagerly
@@ -1853,7 +1853,7 @@ close_experiment
                             ])
                     Mx.stop $ \mux ->
           withNetworkCtx serverCtx $ \serverBearer  ->
-            withAsync (Mx.run ((Server,) `contramap` muxTracer) mux serverBearer) $ \_muxAsync -> do
+            withAsync (Mx.run ((Server,) `contramap` muxTracer) 0 mux serverBearer) $ \_muxAsync -> do
                   Mx.runMiniProtocol
                     mux miniProtocolNum
                     Mx.ResponderDirectionOnly Mx.StartOnDemand

ouroboros-network-api/src/Ouroboros/Network/SizeInBytes.hs

@@ -17,6 +17,7 @@ import Quiet (Quiet (..))
 newtype SizeInBytes = SizeInBytes { getSizeInBytes :: Word32 }
   deriving (Eq, Ord)
   deriving Show      via Quiet SizeInBytes
+  deriving Bounded   via Word32
   deriving Enum      via Word32
   deriving Num       via Word32
   deriving Real      via Word32

ouroboros-network-framework/demo/connection-manager.hs

@@ -37,6 +37,7 @@ import Control.Tracer (Tracer (..), contramap, nullTracer, traceWith)
 import Data.ByteString.Lazy (ByteString)
 import Data.Either (partitionEithers)
 import Data.Functor (($>))
+import Data.Hashable (Hashable)
 import Data.List.NonEmpty (NonEmpty (..))
 import Data.Typeable (Typeable)
 
@@ -77,6 +78,7 @@ import Ouroboros.Network.Server.RateLimiting (AcceptedConnectionsLimit (..))
 import Ouroboros.Network.Server2 qualified as Server
 import Ouroboros.Network.Snocket (Snocket, socketSnocket)
 import Ouroboros.Network.Snocket qualified as Snocket
+import Ouroboros.Network.Socket () -- Hashable SockAddr
 import Ouroboros.Network.Util.ShowProxy
 
 
@@ -174,7 +176,10 @@ withBidirectionalConnectionManager
     :: forall peerAddr socket m a.
        ( ConnectionManagerMonad m
 
-       , Ord peerAddr, Show peerAddr, Typeable peerAddr
+       , Hashable peerAddr
+       , Ord peerAddr
+       , Show peerAddr
+       , Typeable peerAddr
 
        -- debugging
        , MonadFix m
@@ -441,7 +446,8 @@ runInitiatorProtocols
 --
 bidirectionalExperiment
     :: forall peerAddr socket.
-       ( Ord peerAddr
+       ( Hashable peerAddr
+       , Ord peerAddr
        , Show peerAddr
        , Typeable peerAddr
        , Eq peerAddr

ouroboros-network-framework/ouroboros-network-framework.cabal

@@ -140,6 +140,7 @@ library testlib
     cborg,
     containers,
     contra-tracer,
+    hashable,
     io-classes,
     io-sim,
     network-mux,
@@ -331,6 +332,7 @@ executable demo-connection-manager
     base >=4.14 && <4.21,
     bytestring,
     contra-tracer,
+    hashable,
     io-classes,
     network,
     network-mux,