Handle process deaths without scanning unrelated monitors and connections

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

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
+  ]

packages/distributed-process/ChangeLog

@@ -1,3 +1,11 @@
+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 <laurent.decotret@outlook.com> 0.7.9
 
 * `receiveTimeout`, `expectTimeout` and `receiveChanTimeout` now arm the

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.
 --

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

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.
 --