distributed-process 0.7.9 → 0.7.10

raw patch · 7 files changed

+198/−33 lines, 7 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• ChangeLog +5/−0
• benchmarks/Main.hs +70/−7
• distributed-process.cabal +1/−1
• src/Control/Distributed/Process/Internal/BiMultiMap.hs +48/−3
• src/Control/Distributed/Process/Internal/Messaging.hs +43/−0
• src/Control/Distributed/Process/Internal/Types.hs +8/−0
• src/Control/Distributed/Process/Node.hs +23/−22

@@ -1,3 +1,8 @@+2026-10-13 Laurent P. René de Cotret <laurent.decotret@outlook.com> 0.7.10++* Improved the performance of a process death, as it were, to not scale with the+  number of monitors, typed channels, or connections to/from it (#495)+ 2026-09-21 Laurent P. René de Cotret <laurent.decotret@outlook.com> 0.7.9  * `receiveTimeout`, `expectTimeout` and `receiveChanTimeout` now arm the

@@ -102,7 +102,7 @@   ) import Control.Distributed.Process.Serializable (Serializable) import qualified Control.Exception as E-import Control.Monad (forever, replicateM, replicateM_, void, when)+import Control.Monad (forM_, forever, replicateM, replicateM_, void, when) import qualified Control.Monad.Catch as Catch import Data.Binary (Binary) import qualified Data.ByteString.Char8 as BS@@ -118,6 +118,7 @@     bench,     bgroup,     defaultMain,+    envWithCleanup,     whnfIO,   ) @@ -165,7 +166,8 @@         introspection fx,         ring fx       ],-    remote fx+    remote fx,+    deaths fx   ]  -- | Cost of the harness alone, and so the floor below which the other numbers@@ -416,11 +418,7 @@         (getSelfNode >>= monitorNode) >>= unmonitor,       repsBench fx "monitorPort/unmonitor" 100 $         monitorPort (fst (fxChan fx)) >>= unmonitor,-      repsBench fx "notification (normal exit)" 100 $ do-        pid <- spawnLocal (expect :: Process ())-        ref <- monitor pid-        send pid ()-        awaitDown ref,+      repsBench fx "notification (normal exit)" 100 exitAndAwait,       repsBench fx "notification (kill)" 100 $ do         pid <- spawnLocal (expect :: Process ())         ref <- monitor pid@@ -576,6 +574,71 @@   where     echo = fxRemoteEcho fx     nid = fxRemoteNodeId fx++-- | A process death while the node holds many unrelated monitors or+-- connections. Its cost should not depend on their number.+deaths :: Fixture -> Benchmark+deaths fx =+  bgroup+    "deaths"+    [ bgroup+        (show n ++ " " ++ what)+        [ envWithCleanup (start fx n) (stopBackground fx) $ \_ ->+            repsBench fx "notification (normal exit)" 100 exitAndAwait+        ]+    | (what, start) <-+        [ ("monitors", monitoredBackground),+          ("connections", connectedBackground)+        ],+      n <- [0, 1000, 10000]+    ]++-- | @n@ idle processes and, first in the list, one process that monitors them.+monitoredBackground :: Fixture -> Int -> IO [ProcessId]+monitoredBackground fx n = runWith fx $ do+  us <- getSelfPid+  pids <- replicateM n (spawnLocal (expect :: Process ()))+  watcher <- spawnLocal $ do+    mapM_ monitor pids+    send us ()+    expect :: Process ()+  expect :: Process ()+  return (watcher : pids)++-- | @n@ idle processes, each with a connection to the other node.+connectedBackground :: Fixture -> Int -> IO [ProcessId]+connectedBackground fx n = runWith fx $ do+  us <- getSelfPid+  pids <- replicateM n $ spawnLocal $ do+    getSelfPid >>= send (fxRemoteEcho fx)+    expect :: Process ()+    send us ()+    expect :: Process ()+  replicateM_ n (expect :: Process ())+  return pids++-- | Kills the processes one by one, in order, so that their deaths do not+-- overlap with later benchmarks. Killing the watcher of 'monitoredBackground'+-- first keeps the other deaths cheap.+stopBackground :: Fixture -> [ProcessId] -> IO ()+stopBackground fx pids = fxRun fx $ forM_ pids $ \pid -> do+  ref <- monitor pid+  kill pid "benchmark done"+  awaitDown ref++exitAndAwait :: Process ()+exitAndAwait = do+  pid <- spawnLocal (expect :: Process ())+  ref <- monitor pid+  send pid ()+  awaitDown ref++-- | Like 'fxRun', but returns the result.+runWith :: Fixture -> Process a -> IO a+runWith fx act = do+  var <- newEmptyMVar+  fxRun fx (act >>= liftIO . putMVar var)+  takeMVar var  -- | tasty-bench already repeats the body of the benchmark, but the benchmark -- fixture adds a baseline amount of time which drowns some of the faster benchmarks.

@@ -1,6 +1,6 @@ cabal-version: 3.0 Name:          distributed-process-Version:       0.7.9+Version:       0.7.10 Build-Type:    Simple License:       BSD-3-Clause License-File:  LICENSE

@@ -12,6 +12,9 @@   , deleteAllBy2nd   , partitionWithKeyBy1st   , partitionWithKeyBy2nd+  , partitionOnBy1st+  , partitionOnBy2nd+  , partitionOn   , flip   ) where @@ -100,9 +103,7 @@                       -> (Map a (Set (b, v)), BiMultiMap a b v) partitionWithKeyBy1st p (BiMultiMap m r) =     let (m0, m1) = Map.partitionWithKey p m-        r1 = foldl' (\rr (a, mb) -> reverseDelete a (Set.toList mb) rr) r $-               Map.toList m0-     in (m0, BiMultiMap m1 r1)+     in (m0, BiMultiMap m1 (reverseDeleteAll m0 r))  -- | Like 'partitionWithKeyBy1st' but the predicates takes the second component -- of the triplets as first argument.@@ -112,6 +113,42 @@ partitionWithKeyBy2nd p b = let (m, b') = partitionWithKeyBy1st p $ flip b                              in (m, flip b') +-- | Like 'partitionOn' on the first component of the triplets.+--+-- @since 0.7.10+partitionOnBy1st :: (Ord a, Ord b, Ord v, Ord p)+                 => (a -> p) -> [p] -> BiMultiMap a b v+                 -> (Map a (Set (b, v)), BiMultiMap a b v)+partitionOnBy1st f ps (BiMultiMap m r) =+    let (m0, m1) = partitionOn f ps m+     in (m0, BiMultiMap m1 (reverseDeleteAll m0 r))++-- | Like 'partitionOn' on the second component of the triplets.+--+-- @since 0.7.10+partitionOnBy2nd :: (Ord a, Ord b, Ord v, Ord p)+                 => (b -> p) -> [p] -> BiMultiMap a b v+                 -> (Map b (Set (a, v)), BiMultiMap a b v)+partitionOnBy2nd f ps b = let (m, b') = partitionOnBy1st f ps $ flip b+                           in (m, flip b')++-- | @partitionOn f ps m@ is @Map.partitionWithKey (\\k _ -> elem (f k) ps) m@,+-- for a monotone @f@: @k1 <= k2@ implies @f k1 <= f k2@.+--+-- It takes O((length ps + r) * (log n)^2) time for a result of size r, instead+-- of the O(n) of 'Map.partitionWithKey'.+--+-- @since 0.7.10+partitionOn :: (Ord k, Ord p) => (k -> p) -> [p] -> Map k a -> (Map k a, Map k a)+partitionOn f ps m0 = foldl' step (Map.empty, m0) ps+  where+    step (inside, m) p+      | Map.null atP = (inside, m)+      | otherwise    = (Map.union inside atP, Map.union below above)+      where+        (below, rest) = Map.spanAntitone ((< p) . f) m+        (atP, above)  = Map.spanAntitone ((== p) . f) rest+ -- | Exchange the first and the second components of all triplets. flip :: BiMultiMap a b v -> BiMultiMap b a v flip (BiMultiMap m r) = BiMultiMap r m@@ -128,6 +165,14 @@ reverseDelete a bs r = foldl' (\rr (b, v) -> Map.update (rmb v) b rr) r bs   where     rmb v = nothingWhen Set.null . Set.delete (a, v)++-- | @reverseDeleteAll m r@ removes from @r@ the triplets of @m@.+--+-- The @r@ map is in reversed form, like in 'reverseDelete'.+reverseDeleteAll :: (Ord a, Ord b, Ord v)+                 => Map a (Set (b, v)) -> Map b (Set (a, v)) -> Map b (Set (a, v))+reverseDeleteAll m r =+    foldl' (\rr (a, mb) -> reverseDelete a (Set.toList mb) rr) r $ Map.toList m  -- | @nothingWhen p a@ is @Just a@ when @a@ satisfies predicate @p@. -- Yields @Nothing@ otherwise.

@@ -5,6 +5,10 @@   , disconnect   , closeImplicitReconnections   , impliesDeathOf+  , IdentifierKind(..)+  , identifierKind+  , deathKey+  , deathKeys   , sendCtrlMsg   ) where @@ -181,6 +185,45 @@   cid' == cid _ `impliesDeathOf` _ =   False++-- | The constructors of 'Identifier', in the same order.+--+-- @since 0.7.10+data IdentifierKind = NodeKind | ProcessKind | SendPortKind+  deriving (Eq, Ord, Enum, Bounded, Show)++-- | The constructor of an identifier.+--+-- @since 0.7.10+identifierKind :: Identifier -> IdentifierKind+identifierKind NodeIdentifier{}     = NodeKind+identifierKind ProcessIdentifier{}  = ProcessKind+identifierKind SendPortIdentifier{} = SendPortKind++-- | @impliesDeathOf ident i@ if and only if @deathKey ident i@ is in+-- @deathKeys ident@.+--+-- @deathKey ident@ is monotone (see the notes at 'Identifier', 'ProcessId' and+-- 'SendPortId'), so the identifiers that die with @ident@ can be found with+-- 'Control.Distributed.Process.Internal.BiMultiMap.partitionOn'.+--+-- @since 0.7.10+deathKey :: Identifier -> Identifier -> (IdentifierKind, Identifier)+deathKey ident i = (identifierKind i, owner i)+  where+    -- @i@ cut to the level of @ident@: its node, its process, or @i@ itself.+    owner = case ident of+      NodeIdentifier _     -> NodeIdentifier . nodeOf+      ProcessIdentifier _  -> \i' -> case i' of+        SendPortIdentifier cid -> ProcessIdentifier (sendPortProcessId cid)+        _                      -> i'+      SendPortIdentifier _ -> id++-- | See 'deathKey'.+--+-- @since 0.7.10+deathKeys :: Identifier -> [(IdentifierKind, Identifier)]+deathKeys ident = [ (kind, ident) | kind <- [identifierKind ident ..] ]   -- Send a control message. Evaluates the message to HNF before sending it.

@@ -176,6 +176,8 @@   , processLocalId :: {-# UNPACK #-} !LocalProcessId   }   deriving (Eq, Ord, Typeable, Data, Generic)+-- 'Control.Distributed.Process.Internal.Messaging.deathKey' relies on the+-- order of the fields and on the derived 'Ord'.  instance Binary ProcessId where instance NFData ProcessId where rnf (ProcessId n _) = rnf n `seq` ()@@ -191,6 +193,10 @@   | ProcessIdentifier !ProcessId   | SendPortIdentifier !SendPortId   deriving (Eq, Ord, Generic)+-- The constructors are in the order of+-- 'Control.Distributed.Process.Internal.Messaging.IdentifierKind'.+-- 'Control.Distributed.Process.Internal.Messaging.deathKey' relies on that and+-- on the derived 'Ord'.  instance Hashable Identifier where instance NFData Identifier where@@ -412,6 +418,8 @@   , sendPortLocalId   :: {-# UNPACK #-} !LocalSendPortId   }   deriving (Eq, Ord, Typeable, Generic)+-- 'Control.Distributed.Process.Internal.Messaging.deathKey' relies on the+-- order of the fields and on the derived 'Ord'.  instance Hashable SendPortId where instance Show SendPortId where

@@ -29,9 +29,7 @@ import qualified Data.Map as Map   ( empty   , toList-  , fromList   , partition-  , partitionWithKey   , elems   , size   , filterWithKey@@ -50,7 +48,7 @@   , union   ) import Data.Foldable (forM_)-import Data.Maybe (isJust, fromJust, isNothing, catMaybes)+import Data.Maybe (isJust, fromJust, isNothing) import Data.Typeable (Typeable) import Control.Category ((>>>)) import Control.Applicative@@ -199,6 +197,9 @@   ( sendBinary   , closeImplicitReconnections   , impliesDeathOf+  , IdentifierKind(ProcessKind)+  , deathKey+  , deathKeys   ) import Control.Distributed.Process.Internal.Primitives   ( register@@ -457,7 +458,9 @@                    -> IO (ValidLocalNodeState, [NT.Connection])     cleanupProcess pid vst = do       let pid' = ProcessIdentifier pid-      let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)+      let (affected, unaffected) =+            BiMultiMap.partitionOn (deathKey pid' . fst) (deathKeys pid')+                                   (vst ^. localConnections)       return ( (localProcessWithId (processLocalId pid) ^= Nothing)                . (localConnections ^= unaffected)                $ vst@@ -890,14 +893,15 @@           , nid /= localNodeId node     ] +  -- Watchers and registered processes are keyed by ProcessId, so only the+  -- death keys of processes can match.+  let pidDeathKey = deathKey ident . ProcessIdentifier+      pidDeathKeys = filter ((== ProcessKind) . fst) (deathKeys ident)+   -- Delete monitors in the local node.   let deleteDeads :: (Ord a, Ord v)                   => BiMultiMap a ProcessId v -> BiMultiMap a ProcessId v-      deleteDeads = case ident of-                      -- deleteAllBy2nd is faster than partitionWithKeyBy2nd-                      ProcessIdentifier pid -> BiMultiMap.deleteAllBy2nd pid-                      _ -> snd . BiMultiMap.partitionWithKeyBy2nd-                        (\pid _ -> ident `impliesDeathOf` ProcessIdentifier pid)+      deleteDeads = snd . BiMultiMap.partitionOnBy2nd pidDeathKey pidDeathKeys       unaffectedLinks' = deleteDeads unaffectedLinks       unaffectedMons' = deleteDeads unaffectedMons @@ -909,16 +913,13 @@   mapM_ (\(p, l) -> liftIO $ trace node (MxUnRegistered l p)) (Map.toList dropNames)   modify' $ registeredHere ^= keepNames -  remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=-      mapM (\(pid,nidlist) ->-        case ident `impliesDeathOf` ProcessIdentifier pid of-           True ->-              do forM_ nidlist $ \(nid,_) ->-                   when (not $ isLocal node (NodeIdentifier nid))-                      (forwardDeath node nid)-                 return Nothing-           False -> return $ Just (pid,nidlist)  )-  modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))+  (deadRegs, liveRegs) <- gets $+    BiMultiMap.partitionOn pidDeathKey pidDeathKeys . (^. registeredOnNodes)+  forM_ (Map.elems deadRegs) $ \nidlist ->+    forM_ nidlist $ \(nid,_) ->+      when (not $ isLocal node (NodeIdentifier nid))+        (forwardDeath node nid)+  modify' $ registeredOnNodes ^= liveRegs     where        forwardDeath node nid = ncSendToNode nid            NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node)@@ -1277,13 +1278,13 @@ -- * the notifications for that process specifically and -- * the notifications for typed channels to that process. ----- See https://github.com/haskell/containers/issues/14 for the bang on _v.+-- Each set is a range of keys, so the cost does not depend on the number of+-- unaffected notifications. splitNotif :: (Ord a, Ord v)            => Identifier            -> BiMultiMap Identifier a v            -> (Map Identifier (Set (a,v)), BiMultiMap Identifier a v)-splitNotif ident =-    BiMultiMap.partitionWithKeyBy1st (\k !_v -> ident `impliesDeathOf` k)+splitNotif ident = BiMultiMap.partitionOnBy1st (deathKey ident) (deathKeys ident)  -------------------------------------------------------------------------------- -- Auxiliary                                                                  --