eca0b26 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Implement bidirectional multimaps for links and monitors.

eca0b26Facundo Domínguez10 years ago

distributed-process.cabal+1

compare with master

unifiedsplit

@@ -58,6 +58,7 @@ Library

58 Exposed-modules: Control.Distributed.Process,

59 Control.Distributed.Process.Closure,

60 Control.Distributed.Process.Debug,

61+ Control.Distributed.Process.Internal.BiMultiMap,

61 Control.Distributed.Process.Internal.Closure.BuiltIn,

62 Control.Distributed.Process.Internal.Closure.BuiltIn,

62 Control.Distributed.Process.Internal.Closure.Explicit,

63 Control.Distributed.Process.Internal.Closure.Explicit,

63 Control.Distributed.Process.Internal.CQueue,

64 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

42 ( empty

43 , insert

44 , delete

45+ , map

45 , member

46 , member

46 , toList

47 , toList

47 )

48 )

48import Data.Foldable (forM_)

49import Data.Foldable (forM_)

50+import Data.List (foldl')

49import Data.Maybe (isJust, fromJust, isNothing, catMaybes)

51import Data.Maybe (isJust, fromJust, isNothing, catMaybes)

50import Data.Typeable (Typeable)

52import Data.Typeable (Typeable)

51import Control.Category ((>>>))

53import Control.Category ((>>>))

@@ -71,6 +73,8 @@ import qualified Control.Exception as Exception

71 , finally

73 , finally

72 )

74 )

73import Control.Concurrent (forkIO, killThread)

75import Control.Concurrent (forkIO, killThread)

76+import Control.Distributed.Process.Internal.BiMultiMap (BiMultiMap)

77+import qualified Control.Distributed.Process.Internal.BiMultiMap as BiMultiMap

74import Control.Distributed.Process.Internal.StrictMVar

78import Control.Distributed.Process.Internal.StrictMVar

75 ( newMVar

79 ( newMVar

76 , withMVar

80 , withMVar

@@ -656,9 +660,9 @@ runNodeController node =

656

660

657data NCState = NCState

661data NCState = NCState

658 { -- Mapping from remote processes to linked local processes

662 { -- Mapping from remote processes to linked local processes

659− _links :: !(Map Identifier (Set ProcessId))

663+ _links :: !(BiMultiMap Identifier ProcessId ())

660 -- Mapping from remote processes to monitoring local processes

664 -- Mapping from remote processes to monitoring local processes

661− , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))

665+ , _monitors :: !(BiMultiMap Identifier ProcessId MonitorRef)

662 -- Process registry: names and where they live, mapped to the PIDs

666 -- Process registry: names and where they live, mapped to the PIDs

663 , _registeredHere :: !(Map String ProcessId)

667 , _registeredHere :: !(Map String ProcessId)

664 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])

668 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])

@@ -674,8 +678,8 @@ newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }

674 )

678 )

675

679

676initNCState :: NCState

680initNCState :: NCState

677−initNCState = NCState { _links = Map.empty

681+initNCState = NCState { _links = BiMultiMap.empty

678− , _monitors = Map.empty

682+ , _monitors = BiMultiMap.empty

679 , _registeredHere = Map.empty

683 , _registeredHere = Map.empty

680 , _registeredOnNodes = Map.empty

684 , _registeredOnNodes = Map.empty

681 }

685 }

@@ -818,8 +822,8 @@ ncEffectMonitor from them mRef = do

818 case (shouldLink, isLocal node (ProcessIdentifier from)) of

822 case (shouldLink, isLocal node (ProcessIdentifier from)) of

819 (True, _) -> -- [Unified: first rule]

823 (True, _) -> -- [Unified: first rule]

820 case mRef of

824 case mRef of

821− Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)

825+ Just ref -> modify' $ monitors ^: BiMultiMap.insert them from ref

822− Nothing -> modify' $ linksFor them ^: Set.insert from

826+ Nothing -> modify' $ links ^: BiMultiMap.insert them from ()

823 (False, True) -> -- [Unified: second rule]

827 (False, True) -> -- [Unified: second rule]

824 notifyDied from them DiedUnknownId mRef

828 notifyDied from them DiedUnknownId mRef

825 (False, False) -> -- [Unified: third rule]

829 (False, False) -> -- [Unified: third rule]

@@ -843,7 +847,7 @@ ncEffectUnlink from them = do

843 postAsMessage from $ DidUnlinkNode nid

847 postAsMessage from $ DidUnlinkNode nid

844 SendPortIdentifier cid ->

848 SendPortIdentifier cid ->

845 postAsMessage from $ DidUnlinkPort cid

849 postAsMessage from $ DidUnlinkPort cid

846− modify' $ linksFor them ^: Set.delete from

850+ modify' $ links ^: BiMultiMap.delete them from ()

847

851

848-- [Unified: Table 11]

852-- [Unified: Table 11]

849ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()

853ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()

@@ -851,7 +855,7 @@ ncEffectUnmonitor from ref = do

851 node <- ask

855 node <- ask

852 when (isLocal node (ProcessIdentifier from)) $

856 when (isLocal node (ProcessIdentifier from)) $

853 postAsMessage from $ DidUnmonitor ref

857 postAsMessage from $ DidUnmonitor ref

854− modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)

858+ modify' $ monitors ^: BiMultiMap.delete (monitorRefIdent ref) from ref

855

859

856-- [Unified: Table 12]

860-- [Unified: Table 12]

857ncEffectDied :: Identifier -> DiedReason -> NC ()

861ncEffectDied :: Identifier -> DiedReason -> NC ()

@@ -866,7 +870,7 @@ ncEffectDied ident reason = do

866 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False

870 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False

867

871

868 forM_ (Map.toList affectedLinks) $ \(them, uss) ->

872 forM_ (Map.toList affectedLinks) $ \(them, uss) ->

869− forM_ uss $ \us ->

873+ forM_ uss $ \(us, _) ->

870 when (localOnly <= isLocal node (ProcessIdentifier us)) $

874 when (localOnly <= isLocal node (ProcessIdentifier us)) $

871 notifyDied us them reason Nothing

875 notifyDied us them reason Nothing

872

876

@@ -875,7 +879,15 @@ ncEffectDied ident reason = do

875 when (localOnly <= isLocal node (ProcessIdentifier us)) $

879 when (localOnly <= isLocal node (ProcessIdentifier us)) $

876 notifyDied us them reason (Just ref)

880 notifyDied us them reason (Just ref)

877

881

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+

878− modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)

890+ modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')

879

891

880 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)

892 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)

881

893

@@ -1045,8 +1057,9 @@ ncEffectGetInfo from pid =

1045 Nothing -> dispatch (isLocal node (ProcessIdentifier from))

1057 Nothing -> dispatch (isLocal node (ProcessIdentifier from))

1046 from (ProcessInfoNone DiedUnknownId)

1058 from (ProcessInfoNone DiedUnknownId)

1047 Just proc -> do

1059 Just proc -> do

1048− itsLinks <- gets (^. linksFor them)

1060+ itsLinks <- Set.map fst . BiMultiMap.lookup them <$>

1049− itsMons <- gets (^. monitorsFor them)

1061+ gets (^. links)

1062+ itsMons <- BiMultiMap.lookup them <$> gets (^. monitors)

1050 registered <- gets (^. registeredHere)

1063 registered <- gets (^. registeredHere)

1051 size <- liftIO $ queueSize $ processQueue $ proc

1064 size <- liftIO $ queueSize $ processQueue $ proc

1052

1065

@@ -1082,8 +1095,8 @@ ncEffectGetNodeStats from _nid = do

1082 NodeStats {

1095 NodeStats {

1083 nodeStatsNode = localNodeId node

1096 nodeStatsNode = localNodeId node

1084 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere

1097 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere

1085− , nodeStatsMonitors = Map.size $ ncState ^. monitors

1098+ , nodeStatsMonitors = BiMultiMap.size $ ncState ^. monitors

1086− , nodeStatsLinks = Map.size $ ncState ^. links

1099+ , nodeStatsLinks = BiMultiMap.size $ ncState ^. links

1087 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)

1100 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)

1088 }

1101 }

1089 postAsMessage from stats

1102 postAsMessage from stats

@@ -1204,10 +1217,10 @@ withLocalProc node pid p =

1204-- Accessors --

1217-- Accessors --

1205--------------------------------------------------------------------------------

1218--------------------------------------------------------------------------------

1206

1219

1207−links :: Accessor NCState (Map Identifier (Set ProcessId))

1220+links :: Accessor NCState (BiMultiMap Identifier ProcessId ())

1208links = accessor _links (\ls st -> st { _links = ls })

1221links = accessor _links (\ls st -> st { _links = ls })

1209

1222

1210−monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))

1223+monitors :: Accessor NCState (BiMultiMap Identifier ProcessId MonitorRef)

1211monitors = accessor _monitors (\ms st -> st { _monitors = ms })

1224monitors = accessor _monitors (\ms st -> st { _monitors = ms })

1212

1225

1213registeredHere :: Accessor NCState (Map String ProcessId)

1226registeredHere :: Accessor NCState (Map String ProcessId)

@@ -1216,12 +1229,6 @@ registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry }

1216registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])

1229registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])

1217registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })

1230registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })

1218

1231

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−

1225registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)

1232registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)

1226registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident

1233registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident

1227

1234

@@ -1248,10 +1255,12 @@ registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident

1248-- * the notifications for typed channels to that process.

1255-- * the notifications for typed channels to that process.

1249--

1256--

1250-- See https://github.com/haskell/containers/issues/14 for the bang on _v.

1257-- See https://github.com/haskell/containers/issues/14 for the bang on _v.

1258+splitNotif :: (Ord a, Ord v)

1251−splitNotif :: Identifier

1259+ => Identifier

1252− -> Map Identifier a

1260+ -> BiMultiMap Identifier a v

1253− -> (Map Identifier a, Map Identifier a)

1261+ -> (Map Identifier (Set (a,v)), BiMultiMap Identifier a v)

1262+splitNotif ident =

1254−splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)

1263+ BiMultiMap.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)

1255

1264

1256--------------------------------------------------------------------------------

1265--------------------------------------------------------------------------------

1257-- Auxiliary --

1266-- Auxiliary --