git.iko.soy/distributed-process
on master help sign in
Implement bidirectional multimaps for links and monitors.

This solves a leak that occured when a processed died. It wouldn't be removed from any of the monitor lists of the processes which it was monitoring. With a bidirectional multimap we can implement the removal with reasonable performance. Kudos to Alexander Vershilov for investigating and uncovering the leak.

commiteca0b26d26b0b4094145d9d2486500da978f0634 authorFacundo Domínguez <facundo.dominguez@tweag.io> dateThu 18 Aug 2016, 11:49 (10 years ago) parentdfff79f
3 files changed+159−26
compare with master
unifiedsplit
distributed-process.cabal+1 src/Control/Distributed/Process/Internal/BiMultiMap.hs+123 src/Control/Distributed/Process/Node.hs+35−26
distributed-process.cabal+1
@@ -58,6 +58,7 @@ Library
5858 Exposed-modules: Control.Distributed.Process,
5959 Control.Distributed.Process.Closure,
6060 Control.Distributed.Process.Debug,
61+ Control.Distributed.Process.Internal.BiMultiMap,
6162 Control.Distributed.Process.Internal.Closure.BuiltIn,
6263 Control.Distributed.Process.Internal.Closure.Explicit,
6364 Control.Distributed.Process.Internal.CQueue,
src/Control/Distributed/Process/Internal/BiMultiMap.hsnew file+123
@@ -0,0 +1,123 @@
1+-- | This is an implementation of bidirectional multimaps.
2+module Control.Distributed.Process.Internal.BiMultiMap
3+ ( BiMultiMap
4+ , empty
5+ , singleton
6+ , size
7+ , insert
8+ , lookup
9+ , delete
10+ , deleteAllBy1st
11+ , deleteAllBy2nd
12+ , partitionWithKey
13+ , flip
14+ ) where
15+
16+import Data.List (foldl')
17+import Data.Map.Strict (Map)
18+import qualified Data.Map.Strict as Map
19+import Data.Set (Set)
20+import qualified Data.Set as Set
21+import Prelude hiding (flip, lookup)
22+
23+-- | A bidirectional multimaps @BiMultiMap a b v@ is a set of triplets of type
24+-- @(a, b, v)@.
25+--
26+-- It is possible to lookup values by using either @a@ or @b@ as keys.
27+--
28+data BiMultiMap a b v = BiMultiMap !(Map a (Set (b, v))) !(Map b (Set (a, v)))
29+
30+-- The bidirectional multimap is implemented with a pair of multimaps.
31+--
32+-- Each multimap represents a set of triples, and one invariant is that both
33+-- multimaps should represent exactly the same set of triples.
34+--
35+-- Each of the multimaps, however, uses a different component of the triplets
36+-- as key. This allows to do efficient deletions by any of the two components.
37+
38+-- | The empty bidirectional multimap.
39+empty :: BiMultiMap a b v
40+empty = BiMultiMap Map.empty Map.empty
41+
42+-- | A bidirectional multimap containing a single triplet.
43+singleton :: (Ord a, Ord b, Ord v) => a -> b -> v -> BiMultiMap a b v
44+singleton a b v = insert a b v empty
45+
46+-- | Yields the amount of triplets in the multimap.
47+size :: BiMultiMap a b v -> Int
48+size (BiMultiMap m _) = foldl' (+) 0 $ map Set.size $ Map.elems m
49+
50+-- | Inserts a triplet in the multimap.
51+insert :: (Ord a, Ord b, Ord v)
52+ => a -> b -> v -> BiMultiMap a b v -> BiMultiMap a b v
53+insert a b v (BiMultiMap m r) =
54+ BiMultiMap (Map.insertWith (\_new old -> Set.insert (b, v) old)
55+ a
56+ (Set.singleton (b, v))
57+ m)
58+ (Map.insertWith (\_new old -> Set.insert (a, v) old)
59+ b
60+ (Set.singleton (a, v))
61+ r)
62+
63+-- | Looks up all the triplets whose first component is the given value.
64+--
65+-- See 'flip' in order to look up by the second component.
66+lookup :: Ord a => a -> BiMultiMap a b v -> Set (b, v)
67+lookup a (BiMultiMap m _) = maybe Set.empty id $ Map.lookup a m
68+
69+-- | Deletes a triplet. It yields the original multimap if the triplet is
70+-- not present.
71+delete :: (Ord a, Ord b, Ord v)
72+ => a -> b -> v -> BiMultiMap a b v -> BiMultiMap a b v
73+delete a b v (BiMultiMap m r) =
74+ let m' = Map.update (nothingWhen Set.null . Set.delete (b, v)) a m
75+ r' = Map.update (nothingWhen Set.null . Set.delete (a, v)) b r
76+ in BiMultiMap m' r'
77+
78+-- | Deletes all triplets whose first component is the given value.
79+deleteAllBy1st :: (Ord a, Ord b, Ord v) => a -> BiMultiMap a b v -> BiMultiMap a b v
80+deleteAllBy1st a (BiMultiMap m r) =
81+ let (mm, m') = Map.updateLookupWithKey (\_ _ -> Nothing) a m
82+ r' = case mm of
83+ Nothing -> r
84+ Just mb -> reverseDelete a (Set.toList mb) r
85+ in BiMultiMap m' r'
86+
87+-- | Like 'deleteAllBy1st' but deletes by the second component of the triplets.
88+deleteAllBy2nd :: (Ord a, Ord b, Ord v)
89+ => b -> BiMultiMap a b v -> BiMultiMap a b v
90+deleteAllBy2nd b = flip . deleteAllBy1st b . flip
91+
92+-- | Yields the triplets satisfying the given predicate, and a multimap
93+-- with all this triplets removed.
94+partitionWithKey :: (Ord a, Ord b, Ord v)
95+ => (a -> Set (b, v) -> Bool) -> BiMultiMap a b v
96+ -> (Map a (Set (b, v)), BiMultiMap a b v)
97+partitionWithKey f (BiMultiMap m r) =
98+ let (m0, m1) = Map.partitionWithKey f m
99+ r1 = foldl' (\rr (a, mb) -> reverseDelete a (Set.toList mb) rr) r $
100+ Map.toList m0
101+ in (m0, BiMultiMap m1 r1)
102+
103+-- | Exchange the first and the second components of all triplets.
104+flip :: BiMultiMap a b v -> BiMultiMap b a v
105+flip (BiMultiMap m r) = BiMultiMap r m
106+
107+-- Internal functions
108+
109+-- | @reverseDelete a bs m@ removes from @m@ all the triplets wich have @a@ as
110+-- first component and second and third components in @bs@.
111+--
112+-- The @m@ map is in reversed form, meaning that the second component of the
113+-- triplets is used as key.
114+reverseDelete :: (Ord a, Ord b, Ord v)
115+ => a -> [(b, v)] -> Map b (Set (a, v)) -> Map b (Set (a, v))
116+reverseDelete a bs r = foldl' (\rr (b, v) -> Map.update (rmb v) b rr) r bs
117+ where
118+ rmb v = nothingWhen Set.null . Set.delete (a, v)
119+
120+-- | @nothingWhen p a@ is @Just a@ when @a@ satisfies predicate @p@.
121+-- Yields @Nothing@ otherwise.
122+nothingWhen :: (a -> Bool) -> a -> Maybe a
123+nothingWhen p a = if p a then Nothing else Just a
src/Control/Distributed/Process/Node.hs+35−26
@@ -42,10 +42,12 @@ import qualified Data.Set as Set
4242 ( empty
4343 , insert
4444 , delete
45+ , map
4546 , member
4647 , toList
4748 )
4849import Data.Foldable (forM_)
50+import Data.List (foldl')
4951import Data.Maybe (isJust, fromJust, isNothing, catMaybes)
5052import Data.Typeable (Typeable)
5153import Control.Category ((>>>))
@@ -71,6 +73,8 @@ import qualified Control.Exception as Exception
7173 , finally
7274 )
7375import Control.Concurrent (forkIO, killThread)
76+import Control.Distributed.Process.Internal.BiMultiMap (BiMultiMap)
77+import qualified Control.Distributed.Process.Internal.BiMultiMap as BiMultiMap
7478import Control.Distributed.Process.Internal.StrictMVar
7579 ( newMVar
7680 , withMVar
@@ -656,9 +660,9 @@ runNodeController node =
656660
657661data NCState = NCState
658662 { -- Mapping from remote processes to linked local processes
659− _links :: !(Map Identifier (Set ProcessId))
663+ _links :: !(BiMultiMap Identifier ProcessId ())
660664 -- Mapping from remote processes to monitoring local processes
661− , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
665+ , _monitors :: !(BiMultiMap Identifier ProcessId MonitorRef)
662666 -- Process registry: names and where they live, mapped to the PIDs
663667 , _registeredHere :: !(Map String ProcessId)
664668 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
@@ -674,8 +678,8 @@ newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
674678 )
675679
676680initNCState :: NCState
677−initNCState = NCState { _links = Map.empty
678− , _monitors = Map.empty
681+initNCState = NCState { _links = BiMultiMap.empty
682+ , _monitors = BiMultiMap.empty
679683 , _registeredHere = Map.empty
680684 , _registeredOnNodes = Map.empty
681685 }
@@ -818,8 +822,8 @@ ncEffectMonitor from them mRef = do
818822 case (shouldLink, isLocal node (ProcessIdentifier from)) of
819823 (True, _) -> -- [Unified: first rule]
820824 case mRef of
821− Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)
822− Nothing -> modify' $ linksFor them ^: Set.insert from
825+ Just ref -> modify' $ monitors ^: BiMultiMap.insert them from ref
826+ Nothing -> modify' $ links ^: BiMultiMap.insert them from ()
823827 (False, True) -> -- [Unified: second rule]
824828 notifyDied from them DiedUnknownId mRef
825829 (False, False) -> -- [Unified: third rule]
@@ -843,7 +847,7 @@ ncEffectUnlink from them = do
843847 postAsMessage from $ DidUnlinkNode nid
844848 SendPortIdentifier cid ->
845849 postAsMessage from $ DidUnlinkPort cid
846− modify' $ linksFor them ^: Set.delete from
850+ modify' $ links ^: BiMultiMap.delete them from ()
847851
848852-- [Unified: Table 11]
849853ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
@@ -851,7 +855,7 @@ ncEffectUnmonitor from ref = do
851855 node <- ask
852856 when (isLocal node (ProcessIdentifier from)) $
853857 postAsMessage from $ DidUnmonitor ref
854− modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
858+ modify' $ monitors ^: BiMultiMap.delete (monitorRefIdent ref) from ref
855859
856860-- [Unified: Table 12]
857861ncEffectDied :: Identifier -> DiedReason -> NC ()
@@ -866,7 +870,7 @@ ncEffectDied ident reason = do
866870 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
867871
868872 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
869− forM_ uss $ \us ->
873+ forM_ uss $ \(us, _) ->
870874 when (localOnly <= isLocal node (ProcessIdentifier us)) $
871875 notifyDied us them reason Nothing
872876
@@ -875,7 +879,15 @@ ncEffectDied ident reason = do
875879 when (localOnly <= isLocal node (ProcessIdentifier us)) $
876880 notifyDied us them reason (Just ref)
877881
878− modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
882+ let (unaffectedLinks', unaffectedMons') =
883+ case ident of
884+ ProcessIdentifier pid ->
885+ ( BiMultiMap.deleteAllBy2nd pid unaffectedLinks
886+ , BiMultiMap.deleteAllBy2nd pid unaffectedMons
887+ )
888+ _ -> (unaffectedLinks, unaffectedMons)
889+
890+ modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')
879891
880892 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
881893
@@ -1045,8 +1057,9 @@ ncEffectGetInfo from pid =
10451057 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
10461058 from (ProcessInfoNone DiedUnknownId)
10471059 Just proc -> do
1048− itsLinks <- gets (^. linksFor them)
1049− itsMons <- gets (^. monitorsFor them)
1060+ itsLinks <- Set.map fst . BiMultiMap.lookup them <$>
1061+ gets (^. links)
1062+ itsMons <- BiMultiMap.lookup them <$> gets (^. monitors)
10501063 registered <- gets (^. registeredHere)
10511064 size <- liftIO $ queueSize $ processQueue $ proc
10521065
@@ -1082,8 +1095,8 @@ ncEffectGetNodeStats from _nid = do
10821095 NodeStats {
10831096 nodeStatsNode = localNodeId node
10841097 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
1085− , nodeStatsMonitors = Map.size $ ncState ^. monitors
1086− , nodeStatsLinks = Map.size $ ncState ^. links
1098+ , nodeStatsMonitors = BiMultiMap.size $ ncState ^. monitors
1099+ , nodeStatsLinks = BiMultiMap.size $ ncState ^. links
10871100 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
10881101 }
10891102 postAsMessage from stats
@@ -1204,10 +1217,10 @@ withLocalProc node pid p =
12041217-- Accessors --
12051218--------------------------------------------------------------------------------
12061219
1207−links :: Accessor NCState (Map Identifier (Set ProcessId))
1220+links :: Accessor NCState (BiMultiMap Identifier ProcessId ())
12081221links = accessor _links (\ls st -> st { _links = ls })
12091222
1210−monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
1223+monitors :: Accessor NCState (BiMultiMap Identifier ProcessId MonitorRef)
12111224monitors = accessor _monitors (\ms st -> st { _monitors = ms })
12121225
12131226registeredHere :: Accessor NCState (Map String ProcessId)
@@ -1216,12 +1229,6 @@ registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry }
12161229registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
12171230registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
12181231
1219−linksFor :: Identifier -> Accessor NCState (Set ProcessId)
1220−linksFor ident = links >>> DAC.mapDefault Set.empty ident
1221−
1222−monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
1223−monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
1224−
12251232registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
12261233registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
12271234
@@ -1248,10 +1255,12 @@ registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
12481255-- * the notifications for typed channels to that process.
12491256--
12501257-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1251−splitNotif :: Identifier
1252− -> Map Identifier a
1253− -> (Map Identifier a, Map Identifier a)
1254−splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
1258+splitNotif :: (Ord a, Ord v)
1259+ => Identifier
1260+ -> BiMultiMap Identifier a v
1261+ -> (Map Identifier (Set (a,v)), BiMultiMap Identifier a v)
1262+splitNotif ident =
1263+ BiMultiMap.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
12551264
12561265--------------------------------------------------------------------------------
12571266-- Auxiliary --