From aecd7035e75e7b4d9fa7591811d2f90ebe716300 Mon Sep 17 00:00:00 2001 From: Vasiliy Dommes Date: Fri, 9 Oct 2026 11:20:57 -0400 Subject: [PATCH 1/4] Benchmarks: process deaths with many unrelated monitors or connections The cost of a process death should not depend on the monitors and connections of other processes on the node. Measure it with 0, 1,000 and 10,000 of each. Co-Authored-By: Claude Opus 5.5 --- .../distributed-process/benchmarks/Main.hs | 77 +++++++++++++++++-- 1 file changed, 70 insertions(+), 7 deletions(-) diff --git a/packages/distributed-process/benchmarks/Main.hs b/packages/distributed-process/benchmarks/Main.hs index 5b7c9f11..179c6152 100644 --- a/packages/distributed-process/benchmarks/Main.hs +++ b/packages/distributed-process/benchmarks/Main.hs @@ -102,7 +102,7 @@ import Control.Distributed.Process.Node ) 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 @@ import Test.Tasty.Bench bench, bgroup, defaultMain, + envWithCleanup, whnfIO, ) @@ -165,7 +166,8 @@ benchmarks fx = 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 @@ monitoring fx = (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 @@ -577,6 +575,71 @@ remote fx = 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. -- From 44ea46a98180c822dc5dfef3fcae27bc45df171d Mon Sep 17 00:00:00 2001 From: Vasiliy Dommes Date: Wed, 7 Oct 2026 18:22:15 -0400 Subject: [PATCH 2/4] Node controller: find the links and monitors a death affects by key range MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ncEffectDied scanned all links and monitors of the node on every death (splitNotif used partitionWithKeyBy1st with impliesDeathOf), for a node death also all watchers, and it rebuilt registeredOnNodes. So each death took time proportional to all monitors and links of the node, and local sends, which also go through the node controller, waited behind it. Add deathKey, a monotone projection of identifiers such that impliesDeathOf a b if and only if deathKey a b is in deathKeys a. The identifiers that die with a are then a few ranges of keys, which the new BiMultiMap.partitionOn finds with Map.spanAntitone in time polylogarithmic in the size of the map, plus the size of the result. ncEffectDied still scans registeredHere, which is keyed by name. In the deaths benchmark, a process death with 10,000 unrelated monitors on the node went from about 2,900 to 45 µs (about 15 µs with none). Co-Authored-By: Claude Opus 5.5 --- .../distributed-process-tests.cabal | 15 +++ .../tests/TestDeathKeys.hs | 117 ++++++++++++++++++ packages/distributed-process/ChangeLog | 6 + .../Process/Internal/BiMultiMap.hs | 45 ++++++- .../Distributed/Process/Internal/Messaging.hs | 25 ++++ .../Distributed/Process/Internal/Types.hs | 18 +++ .../src/Control/Distributed/Process/Node.hs | 40 +++--- 7 files changed, 243 insertions(+), 23 deletions(-) create mode 100644 packages/distributed-process-tests/tests/TestDeathKeys.hs diff --git a/packages/distributed-process-tests/distributed-process-tests.cabal b/packages/distributed-process-tests/distributed-process-tests.cabal index 8f17bb10..c28763c4 100644 --- a/packages/distributed-process-tests/distributed-process-tests.cabal +++ b/packages/distributed-process-tests/distributed-process-tests.cabal @@ -193,3 +193,18 @@ Test-Suite TestMxInTCP default-language: Haskell98 ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests + +Test-Suite TestDeathKeys + import: warnings + Type: exitcode-stdio-1.0 + Main-Is: TestDeathKeys.hs + Build-Depends: base >= 4.14 && < 5, + bytestring >= 0.10 && < 0.13, + containers >= 0.6 && < 0.9, + distributed-process >= 0.6.0 && < 0.8, + network-transport >= 0.4.1.0 && < 0.6, + tasty >= 1.5 && <1.6, + tasty-hunit >=0.10 && <0.11, + tasty-quickcheck >=0.11 && <0.12 + default-language: Haskell2010 + HS-Source-Dirs: tests diff --git a/packages/distributed-process-tests/tests/TestDeathKeys.hs b/packages/distributed-process-tests/tests/TestDeathKeys.hs new file mode 100644 index 00000000..80638404 --- /dev/null +++ b/packages/distributed-process-tests/tests/TestDeathKeys.hs @@ -0,0 +1,117 @@ +module Main where + +import Control.Distributed.Process.Internal.BiMultiMap (BiMultiMap) +import qualified Control.Distributed.Process.Internal.BiMultiMap as BiMultiMap +import Control.Distributed.Process.Internal.Messaging + ( deathKey + , deathKeys + , impliesDeathOf + ) +import Control.Distributed.Process.Internal.Types + ( Identifier(..) + , LocalProcessId(..) + , NodeId(..) + , ProcessId(..) + , SendPortId(..) + ) +import qualified Data.ByteString.Char8 as BSC +import Data.Map (Map) +import qualified Data.Map as Map +import Data.Set (Set) +import Network.Transport (EndPointAddress(..)) +import Test.Tasty (TestTree, defaultMain, testGroup) +import Test.Tasty.HUnit (assertEqual, testCase) +import Test.Tasty.QuickCheck + ( Arbitrary(..) + , Positive(..) + , elements + , testProperty + ) + +-- All identifiers with 2 nodes, 4 processes per node and 2 ports per process. +universe :: [Identifier] +universe = map NodeIdentifier nodes + ++ map ProcessIdentifier processes + ++ [ SendPortIdentifier (SendPortId pid c) | pid <- processes, c <- [0, 1] ] + where + nodes = [ NodeId (EndPointAddress (BSC.pack a)) | a <- ["a", "b"] ] + processes = [ ProcessId nid (LocalProcessId u c) + | nid <- nodes, u <- [0, 1], c <- [0, 1] ] + +newtype Ident = Ident Identifier + +instance Show Ident where + show (Ident i) = show i + +instance Arbitrary Ident where + arbitrary = Ident <$> elements universe + +testDeathKeyImpliesDeathOf :: TestTree +testDeathKeyImpliesDeathOf = + testCase "deathKey agrees with impliesDeathOf" $ + assertEqual "counterexamples" [] + [ (a, b) | a <- universe, b <- universe + , (a `impliesDeathOf` b) /= (deathKey a b `elem` deathKeys a) ] + +testDeathKeyMonotone :: TestTree +testDeathKeyMonotone = + testCase "deathKey is monotone" $ + assertEqual "counterexamples" [] + [ (a, b1, b2) | a <- universe, b1 <- universe, b2 <- universe + , b1 <= b2, deathKey a b1 > deathKey a b2 ] + +-- Projections on Int: @(`div` d)@ is monotone. +prop_partitionOn :: Positive Int -> [Int] -> Map Int Char -> Bool +prop_partitionOn (Positive d) ps m = + BiMultiMap.partitionOn (`div` d) ps m + == Map.partitionWithKey (\k _ -> (k `div` d) `elem` ps) m + +prop_partitionOnDeathKey :: Ident -> [(Ident, Char)] -> Bool +prop_partitionOnDeathKey (Ident a) kvs = + BiMultiMap.partitionOn (deathKey a) (deathKeys a) m + == Map.partitionWithKey (\k _ -> a `impliesDeathOf` k) m + where + m = Map.fromList [ (k, v) | (Ident k, v) <- kvs ] + +prop_partitionOnBy1st :: Positive Int -> [Int] -> [(Int, Int, Char)] -> Bool +prop_partitionOnBy1st (Positive d) ps ts = + sameResult (BiMultiMap.partitionOnBy1st (`div` d) ps bm) + (BiMultiMap.partitionWithKeyBy1st (\k _ -> (k `div` d) `elem` ps) bm) + where + bm = fromTriplets ts + +prop_partitionOnBy2nd :: Positive Int -> [Int] -> [(Int, Int, Char)] -> Bool +prop_partitionOnBy2nd (Positive d) ps ts = + sameResult (BiMultiMap.partitionOnBy2nd (`div` d) ps bm) + (BiMultiMap.partitionWithKeyBy2nd (\k _ -> (k `div` d) `elem` ps) bm) + where + bm = fromTriplets ts + +sameResult :: (Eq k, Eq s) + => (Map k s, BiMultiMap Int Int Char) + -> (Map k s, BiMultiMap Int Int Char) + -> Bool +sameResult (m, bm) (m', bm') = m == m' && sides bm == sides bm' + +fromTriplets :: [(Int, Int, Char)] -> BiMultiMap Int Int Char +fromTriplets = foldr (\(a, b, v) -> BiMultiMap.insert a b v) BiMultiMap.empty + +-- Both sides of a multimap: by the first and by the second component. +sides :: BiMultiMap Int Int Char + -> (Map Int (Set (Int, Char)), Map Int (Set (Int, Char))) +sides bm = ( fst (BiMultiMap.partitionWithKeyBy1st (\_ _ -> True) bm) + , fst (BiMultiMap.partitionWithKeyBy2nd (\_ _ -> True) bm) + ) + +main :: IO () +main = defaultMain tests + +tests :: TestTree +tests = testGroup "Death keys" + [ testDeathKeyImpliesDeathOf + , testDeathKeyMonotone + , testProperty "partitionOn" prop_partitionOn + , testProperty "partitionOn with deathKey" prop_partitionOnDeathKey + , testProperty "partitionOnBy1st" prop_partitionOnBy1st + , testProperty "partitionOnBy2nd" prop_partitionOnBy2nd + ] diff --git a/packages/distributed-process/ChangeLog b/packages/distributed-process/ChangeLog index 6b9603ee..6d9ea20d 100644 --- a/packages/distributed-process/ChangeLog +++ b/packages/distributed-process/ChangeLog @@ -1,3 +1,9 @@ +Unreleased + +* The node controller handles the death of a node, process or typed channel in + time that depends on the number of links and monitors it affects, not on + their total number. + 2026-09-21 Laurent P. René de Cotret 0.7.9 * `receiveTimeout`, `expectTimeout` and `receiveChanTimeout` now arm the diff --git a/packages/distributed-process/src/Control/Distributed/Process/Internal/BiMultiMap.hs b/packages/distributed-process/src/Control/Distributed/Process/Internal/BiMultiMap.hs index 7b97b447..c614389e 100644 --- a/packages/distributed-process/src/Control/Distributed/Process/Internal/BiMultiMap.hs +++ b/packages/distributed-process/src/Control/Distributed/Process/Internal/BiMultiMap.hs @@ -12,6 +12,9 @@ module Control.Distributed.Process.Internal.BiMultiMap , deleteAllBy2nd , partitionWithKeyBy1st , partitionWithKeyBy2nd + , partitionOnBy1st + , partitionOnBy2nd + , partitionOn , flip ) where @@ -100,9 +103,7 @@ partitionWithKeyBy1st :: (Ord a, Ord b, Ord v) -> (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,36 @@ partitionWithKeyBy2nd :: (Ord a, Ord b, Ord v) partitionWithKeyBy2nd p b = let (m, b') = partitionWithKeyBy1st p $ flip b in (m, flip b') +-- | Like 'partitionOn' on the first component of the triplets. +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. +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'. +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 @@ -129,6 +160,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. nothingWhen :: (a -> Bool) -> a -> Maybe a diff --git a/packages/distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs b/packages/distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs index f7e5874f..0d996183 100644 --- a/packages/distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs +++ b/packages/distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs @@ -5,6 +5,8 @@ module Control.Distributed.Process.Internal.Messaging , disconnect , closeImplicitReconnections , impliesDeathOf + , deathKey + , deathKeys , sendCtrlMsg ) where @@ -52,6 +54,8 @@ import Control.Distributed.Process.Internal.Types , Process(..) , SendPortId(sendPortProcessId) , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier) + , IdentifierKind + , identifierKind ) import Control.Distributed.Process.Serializable (Serializable) import Data.Foldable (forM_) @@ -182,6 +186,27 @@ SendPortIdentifier cid `impliesDeathOf` SendPortIdentifier cid' = _ `impliesDeathOf` _ = False +-- | @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'. +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'. +deathKeys :: Identifier -> [(IdentifierKind, Identifier)] +deathKeys ident = [ (kind, ident) | kind <- [identifierKind ident ..] ] + -- Send a control message. Evaluates the message to HNF before sending it. -- diff --git a/packages/distributed-process/src/Control/Distributed/Process/Internal/Types.hs b/packages/distributed-process/src/Control/Distributed/Process/Internal/Types.hs index 126ae3b2..0af7796f 100644 --- a/packages/distributed-process/src/Control/Distributed/Process/Internal/Types.hs +++ b/packages/distributed-process/src/Control/Distributed/Process/Internal/Types.hs @@ -20,6 +20,8 @@ module Control.Distributed.Process.Internal.Types , ProcessId(..) , Identifier(..) , nodeOf + , IdentifierKind(..) + , identifierKind , firstNonReservedProcessId , nullProcessId -- * Local nodes and processes @@ -176,6 +178,8 @@ data ProcessId = ProcessId , 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 +195,9 @@ data Identifier = | ProcessIdentifier !ProcessId | SendPortIdentifier !SendPortId deriving (Eq, Ord, Generic) +-- The constructors are in the order of 'IdentifierKind'. +-- 'Control.Distributed.Process.Internal.Messaging.deathKey' relies on that and +-- on the derived 'Ord'. instance Hashable Identifier where instance NFData Identifier where @@ -208,6 +215,15 @@ nodeOf (NodeIdentifier nid) = nid nodeOf (ProcessIdentifier pid) = processNodeId pid nodeOf (SendPortIdentifier cid) = processNodeId (sendPortProcessId cid) +-- | The constructors of 'Identifier', in the same order. +data IdentifierKind = NodeKind | ProcessKind | SendPortKind + deriving (Eq, Ord, Enum, Bounded, Show) + +identifierKind :: Identifier -> IdentifierKind +identifierKind NodeIdentifier{} = NodeKind +identifierKind ProcessIdentifier{} = ProcessKind +identifierKind SendPortIdentifier{} = SendPortKind + -------------------------------------------------------------------------------- -- Special PIDs -- -------------------------------------------------------------------------------- @@ -412,6 +428,8 @@ data SendPortId = SendPortId { , 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 diff --git a/packages/distributed-process/src/Control/Distributed/Process/Node.hs b/packages/distributed-process/src/Control/Distributed/Process/Node.hs index 8487d894..c3c51421 100644 --- a/packages/distributed-process/src/Control/Distributed/Process/Node.hs +++ b/packages/distributed-process/src/Control/Distributed/Process/Node.hs @@ -29,7 +29,6 @@ import Data.Map (Map) import qualified Data.Map as Map ( empty , toList - , fromList , partition , partitionWithKey , elems @@ -50,7 +49,7 @@ import qualified Data.Set as Set , 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 @@ -123,6 +122,7 @@ import qualified Control.Distributed.Static as Static ) import Control.Distributed.Process.Internal.Types ( NodeId(..) + , IdentifierKind(ProcessKind) , LocalProcessId(..) , ProcessId(..) , LocalNode(..) @@ -199,6 +199,8 @@ import Control.Distributed.Process.Internal.Messaging ( sendBinary , closeImplicitReconnections , impliesDeathOf + , deathKey + , deathKeys ) import Control.Distributed.Process.Internal.Primitives ( register @@ -890,14 +892,15 @@ ncEffectDied ident reason = do , 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 +912,13 @@ ncEffectDied ident reason = do 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 +1277,13 @@ registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident -- * 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 -- From d98c0f6aa579674f6ea08079236603eb1766a2ea Mon Sep 17 00:00:00 2001 From: Vasiliy Dommes Date: Wed, 7 Oct 2026 18:24:28 -0400 Subject: [PATCH 3/4] Process exit: remove the process's connections by key range MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cleanupProcess scanned all of localConnections on every process exit. The result is part of the node state, which local and remote sends, spawns and the node controller also use. Connections are keyed by (from, to), so the connections of the process and its typed channels are found with partitionOn and deathKey. In the deaths benchmark, a process death with 10,000 unrelated connections on the node went from about 2,300 to 37 µs (about 15 µs with none). Co-Authored-By: Claude Opus 5.5 --- packages/distributed-process/ChangeLog | 2 ++ .../src/Control/Distributed/Process/Node.hs | 5 +++-- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/packages/distributed-process/ChangeLog b/packages/distributed-process/ChangeLog index 6d9ea20d..30696d14 100644 --- a/packages/distributed-process/ChangeLog +++ b/packages/distributed-process/ChangeLog @@ -3,6 +3,8 @@ Unreleased * The node controller handles the death of a node, process or typed channel in time that depends on the number of links and monitors it affects, not on their total number. +* A process exit takes time that depends on the number of connections of + the process, not on the number of connections of the node. 2026-09-21 Laurent P. René de Cotret 0.7.9 diff --git a/packages/distributed-process/src/Control/Distributed/Process/Node.hs b/packages/distributed-process/src/Control/Distributed/Process/Node.hs index c3c51421..cebcc414 100644 --- a/packages/distributed-process/src/Control/Distributed/Process/Node.hs +++ b/packages/distributed-process/src/Control/Distributed/Process/Node.hs @@ -30,7 +30,6 @@ import qualified Data.Map as Map ( empty , toList , partition - , partitionWithKey , elems , size , filterWithKey @@ -459,7 +458,9 @@ forkProcess node proc = do -> 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 From cf2592b030baca7aa947c8308d7d7cfd03fac4f7 Mon Sep 17 00:00:00 2001 From: Vasiliy Dommes Date: Tue, 13 Oct 2026 14:22:27 -0400 Subject: [PATCH 4/4] Address review: version 0.7.10, @since, move IdentifierKind - Bump distributed-process to 0.7.10, and require it in TestDeathKeys. - Add @since 0.7.10 to the new exports of BiMultiMap and Messaging. - Move IdentifierKind and identifierKind to Messaging, next to deathKey, their only user. Co-Authored-By: Claude Opus 5.5 --- .../distributed-process-tests.cabal | 2 +- .../distributed-process.cabal | 2 +- .../Process/Internal/BiMultiMap.hs | 6 +++++ .../Distributed/Process/Internal/Messaging.hs | 22 +++++++++++++++++-- .../Distributed/Process/Internal/Types.hs | 14 ++---------- .../src/Control/Distributed/Process/Node.hs | 2 +- 6 files changed, 31 insertions(+), 17 deletions(-) diff --git a/packages/distributed-process-tests/distributed-process-tests.cabal b/packages/distributed-process-tests/distributed-process-tests.cabal index c28763c4..ef09311f 100644 --- a/packages/distributed-process-tests/distributed-process-tests.cabal +++ b/packages/distributed-process-tests/distributed-process-tests.cabal @@ -201,7 +201,7 @@ Test-Suite TestDeathKeys Build-Depends: base >= 4.14 && < 5, bytestring >= 0.10 && < 0.13, containers >= 0.6 && < 0.9, - distributed-process >= 0.6.0 && < 0.8, + distributed-process >= 0.7.10 && < 0.8, network-transport >= 0.4.1.0 && < 0.6, tasty >= 1.5 && <1.6, tasty-hunit >=0.10 && <0.11, diff --git a/packages/distributed-process/distributed-process.cabal b/packages/distributed-process/distributed-process.cabal index bfcfd718..a08237b5 100644 --- a/packages/distributed-process/distributed-process.cabal +++ b/packages/distributed-process/distributed-process.cabal @@ -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 diff --git a/packages/distributed-process/src/Control/Distributed/Process/Internal/BiMultiMap.hs b/packages/distributed-process/src/Control/Distributed/Process/Internal/BiMultiMap.hs index c614389e..08be7acf 100644 --- a/packages/distributed-process/src/Control/Distributed/Process/Internal/BiMultiMap.hs +++ b/packages/distributed-process/src/Control/Distributed/Process/Internal/BiMultiMap.hs @@ -114,6 +114,8 @@ 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) @@ -122,6 +124,8 @@ partitionOnBy1st f ps (BiMultiMap m r) = 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) @@ -133,6 +137,8 @@ partitionOnBy2nd f ps b = let (m, b') = partitionOnBy1st f ps $ flip b -- -- 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 diff --git a/packages/distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs b/packages/distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs index 0d996183..08835747 100644 --- a/packages/distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs +++ b/packages/distributed-process/src/Control/Distributed/Process/Internal/Messaging.hs @@ -5,6 +5,8 @@ module Control.Distributed.Process.Internal.Messaging , disconnect , closeImplicitReconnections , impliesDeathOf + , IdentifierKind(..) + , identifierKind , deathKey , deathKeys , sendCtrlMsg @@ -54,8 +56,6 @@ import Control.Distributed.Process.Internal.Types , Process(..) , SendPortId(sendPortProcessId) , Identifier(NodeIdentifier, ProcessIdentifier, SendPortIdentifier) - , IdentifierKind - , identifierKind ) import Control.Distributed.Process.Serializable (Serializable) import Data.Foldable (forM_) @@ -186,12 +186,28 @@ SendPortIdentifier cid `impliesDeathOf` SendPortIdentifier 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 @@ -204,6 +220,8 @@ deathKey ident i = (identifierKind i, owner i) SendPortIdentifier _ -> id -- | See 'deathKey'. +-- +-- @since 0.7.10 deathKeys :: Identifier -> [(IdentifierKind, Identifier)] deathKeys ident = [ (kind, ident) | kind <- [identifierKind ident ..] ] diff --git a/packages/distributed-process/src/Control/Distributed/Process/Internal/Types.hs b/packages/distributed-process/src/Control/Distributed/Process/Internal/Types.hs index 0af7796f..21fcc706 100644 --- a/packages/distributed-process/src/Control/Distributed/Process/Internal/Types.hs +++ b/packages/distributed-process/src/Control/Distributed/Process/Internal/Types.hs @@ -20,8 +20,6 @@ module Control.Distributed.Process.Internal.Types , ProcessId(..) , Identifier(..) , nodeOf - , IdentifierKind(..) - , identifierKind , firstNonReservedProcessId , nullProcessId -- * Local nodes and processes @@ -195,7 +193,8 @@ data Identifier = | ProcessIdentifier !ProcessId | SendPortIdentifier !SendPortId deriving (Eq, Ord, Generic) --- The constructors are in the order of 'IdentifierKind'. +-- 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'. @@ -215,15 +214,6 @@ nodeOf (NodeIdentifier nid) = nid nodeOf (ProcessIdentifier pid) = processNodeId pid nodeOf (SendPortIdentifier cid) = processNodeId (sendPortProcessId cid) --- | The constructors of 'Identifier', in the same order. -data IdentifierKind = NodeKind | ProcessKind | SendPortKind - deriving (Eq, Ord, Enum, Bounded, Show) - -identifierKind :: Identifier -> IdentifierKind -identifierKind NodeIdentifier{} = NodeKind -identifierKind ProcessIdentifier{} = ProcessKind -identifierKind SendPortIdentifier{} = SendPortKind - -------------------------------------------------------------------------------- -- Special PIDs -- -------------------------------------------------------------------------------- diff --git a/packages/distributed-process/src/Control/Distributed/Process/Node.hs b/packages/distributed-process/src/Control/Distributed/Process/Node.hs index cebcc414..c527e47d 100644 --- a/packages/distributed-process/src/Control/Distributed/Process/Node.hs +++ b/packages/distributed-process/src/Control/Distributed/Process/Node.hs @@ -121,7 +121,6 @@ import qualified Control.Distributed.Static as Static ) import Control.Distributed.Process.Internal.Types ( NodeId(..) - , IdentifierKind(ProcessKind) , LocalProcessId(..) , ProcessId(..) , LocalNode(..) @@ -198,6 +197,7 @@ import Control.Distributed.Process.Internal.Messaging ( sendBinary , closeImplicitReconnections , impliesDeathOf + , IdentifierKind(ProcessKind) , deathKey , deathKeys )