diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,13 @@ +2016-10-12 Facundo Domínguez 0.6.5 + +* Use only one connection to communicate between NCs. (#296, #297) +* Improve documentation of CQueue. +* Implement bidirectional multimaps for links and monitors. (#293, #294, #295) +* Fix various warnings. (#292) +* Fix some of the intermittent failures in tests. +* Improve error messages when node controllers receive invalid requests. +(#290) + 2016-06-09 Facundo Domínguez 0.6.4 * Fixup build errors. diff --git a/distributed-process.cabal b/distributed-process.cabal --- a/distributed-process.cabal +++ b/distributed-process.cabal @@ -1,5 +1,5 @@ Name: distributed-process -Version: 0.6.4 +Version: 0.6.5 Cabal-Version: >=1.8 Build-Type: Simple License: BSD3 @@ -58,6 +58,7 @@ Exposed-modules: Control.Distributed.Process, Control.Distributed.Process.Closure, Control.Distributed.Process.Debug, + Control.Distributed.Process.Internal.BiMultiMap, Control.Distributed.Process.Internal.Closure.BuiltIn, Control.Distributed.Process.Internal.Closure.Explicit, Control.Distributed.Process.Internal.CQueue, diff --git a/src/Control/Distributed/Process.hs b/src/Control/Distributed/Process.hs --- a/src/Control/Distributed/Process.hs +++ b/src/Control/Distributed/Process.hs @@ -325,7 +325,6 @@ #else import Prelude hiding (catch) #endif -import Control.Distributed.Process.Internal.StrictMVar (readMVar) import qualified Control.Exception as Exception (onException) import Data.Accessor ((^.)) import Data.Foldable (forM_) diff --git a/src/Control/Distributed/Process/Internal/BiMultiMap.hs b/src/Control/Distributed/Process/Internal/BiMultiMap.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Internal/BiMultiMap.hs @@ -0,0 +1,132 @@ +-- | This is an implementation of bidirectional multimaps. +module Control.Distributed.Process.Internal.BiMultiMap + ( BiMultiMap + , empty + , singleton + , size + , insert + , lookup + , delete + , deleteAllBy1st + , deleteAllBy2nd + , partitionWithKeyBy1st + , partitionWithKeyBy2nd + , flip + ) where + +import Data.List (foldl') +import Data.Map.Strict (Map) +import qualified Data.Map.Strict as Map +import Data.Set (Set) +import qualified Data.Set as Set +import Prelude hiding (flip, lookup) + +-- | A bidirectional multimaps @BiMultiMap a b v@ is a set of triplets of type +-- @(a, b, v)@. +-- +-- It is possible to lookup values by using either @a@ or @b@ as keys. +-- +data BiMultiMap a b v = BiMultiMap !(Map a (Set (b, v))) !(Map b (Set (a, v))) + +-- The bidirectional multimap is implemented with a pair of multimaps. +-- +-- Each multimap represents a set of triples, and one invariant is that both +-- multimaps should represent exactly the same set of triples. +-- +-- Each of the multimaps, however, uses a different component of the triplets +-- as key. This allows to do efficient deletions by any of the two components. + +-- | The empty bidirectional multimap. +empty :: BiMultiMap a b v +empty = BiMultiMap Map.empty Map.empty + +-- | A bidirectional multimap containing a single triplet. +singleton :: (Ord a, Ord b, Ord v) => a -> b -> v -> BiMultiMap a b v +singleton a b v = insert a b v empty + +-- | Yields the amount of triplets in the multimap. +size :: BiMultiMap a b v -> Int +size (BiMultiMap m _) = foldl' (+) 0 $ map Set.size $ Map.elems m + +-- | Inserts a triplet in the multimap. +insert :: (Ord a, Ord b, Ord v) + => a -> b -> v -> BiMultiMap a b v -> BiMultiMap a b v +insert a b v (BiMultiMap m r) = + BiMultiMap (Map.insertWith (\_new old -> Set.insert (b, v) old) + a + (Set.singleton (b, v)) + m) + (Map.insertWith (\_new old -> Set.insert (a, v) old) + b + (Set.singleton (a, v)) + r) + +-- | Looks up all the triplets whose first component is the given value. +-- +-- See 'flip' in order to look up by the second component. +lookup :: Ord a => a -> BiMultiMap a b v -> Set (b, v) +lookup a (BiMultiMap m _) = maybe Set.empty id $ Map.lookup a m + +-- | Deletes a triplet. It yields the original multimap if the triplet is +-- not present. +delete :: (Ord a, Ord b, Ord v) + => a -> b -> v -> BiMultiMap a b v -> BiMultiMap a b v +delete a b v (BiMultiMap m r) = + let m' = Map.update (nothingWhen Set.null . Set.delete (b, v)) a m + r' = Map.update (nothingWhen Set.null . Set.delete (a, v)) b r + in BiMultiMap m' r' + +-- | Deletes all triplets whose first component is the given value. +deleteAllBy1st :: (Ord a, Ord b, Ord v) => a -> BiMultiMap a b v -> BiMultiMap a b v +deleteAllBy1st a (BiMultiMap m r) = + let (mm, m') = Map.updateLookupWithKey (\_ _ -> Nothing) a m + r' = case mm of + Nothing -> r + Just mb -> reverseDelete a (Set.toList mb) r + in BiMultiMap m' r' + +-- | Like 'deleteAllBy1st' but deletes by the second component of the triplets. +deleteAllBy2nd :: (Ord a, Ord b, Ord v) + => b -> BiMultiMap a b v -> BiMultiMap a b v +deleteAllBy2nd b = flip . deleteAllBy1st b . flip + +-- | Yields the triplets satisfying the given predicate, and a multimap +-- with all this triplets removed. +partitionWithKeyBy1st :: (Ord a, Ord b, Ord v) + => (a -> Set (b, v) -> Bool) -> BiMultiMap a b 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) + +-- | Like 'partitionWithKeyBy1st' but the predicates takes the second component +-- of the triplets as first argument. +partitionWithKeyBy2nd :: (Ord a, Ord b, Ord v) + => (b -> Set (a, v) -> Bool) -> BiMultiMap a b v + -> (Map b (Set (a, v)), BiMultiMap a b v) +partitionWithKeyBy2nd p b = let (m, b') = partitionWithKeyBy1st p $ flip b + in (m, flip b') + +-- | 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 + +-- Internal functions + +-- | @reverseDelete a bs m@ removes from @m@ all the triplets wich have @a@ as +-- first component and second and third components in @bs@. +-- +-- The @m@ map is in reversed form, meaning that the second component of the +-- triplets is used as key. +reverseDelete :: (Ord a, Ord b, Ord v) + => a -> [(b, v)] -> Map b (Set (a, v)) -> Map b (Set (a, v)) +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) + +-- | @nothingWhen p a@ is @Just a@ when @a@ satisfies predicate @p@. +-- Yields @Nothing@ otherwise. +nothingWhen :: (a -> Bool) -> a -> Maybe a +nothingWhen p a = if p a then Nothing else Just a diff --git a/src/Control/Distributed/Process/Internal/CQueue.hs b/src/Control/Distributed/Process/Internal/CQueue.hs --- a/src/Control/Distributed/Process/Internal/CQueue.hs +++ b/src/Control/Distributed/Process/Internal/CQueue.hs @@ -76,13 +76,25 @@ | Blocking | Timeout Int +-- Match operations +-- +-- They can be either a message match or a channel match. data MatchOn m a = MatchMsg (m -> Maybe a) | MatchChan (STM a) deriving (Functor) +-- Lists of chunks of matches +-- +-- Two consecutive chunks never have the same kind of matches. i.e. if one chunk +-- contains message matches then the next one must contain channel matches and +-- viceversa. type MatchChunks m a = [Either [m -> Maybe a] [STM a]] +-- Splits a list of matches into chunks. +-- +-- > concatMap (either (map MatchMsg) (map MatchChan)) . chunkMatches == id +-- chunkMatches :: [MatchOn m a] -> MatchChunks m a chunkMatches [] = [] chunkMatches (MatchMsg m : ms) = Left (m : chk) : chunkMatches rest @@ -90,6 +102,7 @@ chunkMatches (MatchChan r : ms) = Right (r : chk) : chunkMatches rest where (chk, rest) = spanMatchChan ms +-- | @spanMatchMsg = first (map (\(MatchMsg x) -> x)) . span isMatchMsg@ spanMatchMsg :: [MatchOn m a] -> ([m -> Maybe a], [MatchOn m a]) spanMatchMsg [] = ([],[]) spanMatchMsg (m : ms) @@ -97,6 +110,7 @@ | otherwise = ([], m:ms) where !(msgs,rest) = spanMatchMsg ms +-- | @spanMatchMsg = first (map (\(MatchChan x) -> x)) . span isMatchChan@ spanMatchChan :: [MatchOn m a] -> ([STM a], [MatchOn m a]) spanMatchChan [] = ([],[]) spanMatchChan (m : ms) @@ -128,6 +142,7 @@ -- Decrement counter is smth is returned from the queue, -- this is safe to use as method is called under a mask -- and there is no 'unmasked' operation inside + decrementJust :: IO (Maybe (Either a a)) -> IO (Maybe a) decrementJust f = traverse (either return (\x -> decrement >> return x)) =<< f decrement = atomically $ modifyTVar' size pred @@ -144,6 +159,10 @@ arr' <- grabNew arr goCheck chunks arr' + -- Yields the value of the first succesful STM transaction as + -- @Just (Left v)@. If all transactions fail, yields the value of the second + -- argument. + waitChans :: [STM a] -> STM (Maybe (Either a a)) -> STM (Maybe (Either a a)) waitChans ports on_block = foldr orElse on_block (map (fmap (Just . Left)) ports) @@ -152,6 +171,10 @@ -- mailbox. For channel matches, we do a non-blocking check at -- this point. -- + -- Yields @Just (Left a)@ when a channel is matched, @Just (Right a)@ + -- when a message is matched and @Nothing@ when there are no messages and we + -- aren't blocking. + -- goCheck :: MatchChunks m a -> StrictList m -- messages to check, in this order -> IO (Maybe (Either a a)) @@ -250,6 +273,7 @@ checkArrived :: [m -> Maybe a] -> StrictList m -> (StrictList m, Maybe a) checkArrived matches list = go list Nil where + -- @go xs ys@ searches for a message match in @append xs ys@ go Nil Nil = (Nil, Nothing) go Nil r = go r Nil go (Append xs ys) tl = go xs (append ys tl) diff --git a/src/Control/Distributed/Process/Internal/Closure/Explicit.hs b/src/Control/Distributed/Process/Internal/Closure/Explicit.hs --- a/src/Control/Distributed/Process/Internal/Closure/Explicit.hs +++ b/src/Control/Distributed/Process/Internal/Closure/Explicit.hs @@ -6,7 +6,6 @@ , UndecidableInstances , KindSignatures , GADTs - , OverlappingInstances , EmptyDataDecls , DeriveDataTypeable #-} module Control.Distributed.Process.Internal.Closure.Explicit @@ -109,7 +108,7 @@ class Curry a b | a -> b where curryFun :: a -> b -instance Curry ((a,EndOfTuple) -> b) (a -> b) where +instance Curry ((a, EndOfTuple) -> b) (a -> b) where curryFun f = \x -> f (x,undefined) instance Curry (b -> c) r => Curry ((a,b) -> c) (a -> r) where diff --git a/src/Control/Distributed/Process/Internal/Closure/TH.hs b/src/Control/Distributed/Process/Internal/Closure/TH.hs --- a/src/Control/Distributed/Process/Internal/Closure/TH.hs +++ b/src/Control/Distributed/Process/Internal/Closure/TH.hs @@ -29,11 +29,14 @@ , Type(AppT, ForallT, VarT, ArrowT) , Info(VarI) , TyVarBndr(PlainTV, KindedTV) -#if ! MIN_VERSION_template_haskell(2,10,0) , Pred +#if MIN_VERSION_template_haskell(2,10,0) + , conT + , appT +#else + , classP #endif , varT - , classP -- Lifted constructors -- .. Literals , stringL @@ -71,10 +74,6 @@ ) import Control.Distributed.Process.Internal.Closure.BuiltIn (staticDecode) -#if MIN_VERSION_template_haskell(2,10,0) -type Pred = Type -#endif - -------------------------------------------------------------------------------- -- User-level API -- -------------------------------------------------------------------------------- @@ -261,7 +260,12 @@ ] where typeable :: TyVarBndr -> Q Pred - typeable tv = classP (mkName "Typeable") [varT (tyVarBndrName tv)] + typeable tv = +#if MIN_VERSION_template_haskell(2,10,0) + conT (mkName "Typeable") `appT` varT (tyVarBndrName tv) +#else + classP (mkName "Typeable") [varT (tyVarBndrName tv)] +#endif -- | Generate a serialization dictionary with name 'n' for type 'typ' generateDict :: Name -> Type -> Q [Dec] diff --git a/src/Control/Distributed/Process/Internal/Messaging.hs b/src/Control/Distributed/Process/Internal/Messaging.hs --- a/src/Control/Distributed/Process/Internal/Messaging.hs +++ b/src/Control/Distributed/Process/Internal/Messaging.hs @@ -30,7 +30,7 @@ , close ) import Control.Distributed.Process.Internal.Types - ( LocalNode(localState, localEndPoint, localCtrlChan) + ( LocalNode(localEndPoint, localCtrlChan) , withValidLocalState , modifyValidLocalState , modifyValidLocalState_ @@ -201,7 +201,7 @@ liftIO $ writeChan (localCtrlChan (processNode proc)) $! msg Just nid -> liftIO $ sendBinary (processNode proc) - (ProcessIdentifier (processId proc)) + (NodeIdentifier (processNodeId $ processId proc)) (NodeIdentifier nid) WithImplicitReconnect msg diff --git a/src/Control/Distributed/Process/Internal/Spawn.hs b/src/Control/Distributed/Process/Internal/Spawn.hs --- a/src/Control/Distributed/Process/Internal/Spawn.hs +++ b/src/Control/Distributed/Process/Internal/Spawn.hs @@ -43,7 +43,7 @@ ) import Control.Distributed.Process.Internal.Primitives ( -- Basic messaging - send + usend , expect , receiveWait , match @@ -71,7 +71,7 @@ receiveWait [ matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do unmonitor mRef - send pid () + usend pid () return pid , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ -> return (nullProcessId nid) @@ -118,7 +118,7 @@ cpDelayed us (returnCP sdictUnit ()) ) mResult <- receiveWait - [ match $ \a -> send pid () >> return (Right a) + [ match $ \a -> usend pid () >> return (Right a) , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef) (\(ProcessMonitorNotification _ _ reason) -> return (Left reason)) ] diff --git a/src/Control/Distributed/Process/Management/Internal/Types.hs b/src/Control/Distributed/Process/Management/Internal/Types.hs --- a/src/Control/Distributed/Process/Management/Internal/Types.hs +++ b/src/Control/Distributed/Process/Management/Internal/Types.hs @@ -1,5 +1,4 @@ {-# LANGUAGE DeriveDataTypeable #-} -{-# LANGUAGE StandaloneDeriving #-} {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE DeriveGeneric #-} diff --git a/src/Control/Distributed/Process/Node.hs b/src/Control/Distributed/Process/Node.hs --- a/src/Control/Distributed/Process/Node.hs +++ b/src/Control/Distributed/Process/Node.hs @@ -42,10 +42,12 @@ ( empty , insert , delete + , map , member , toList ) import Data.Foldable (forM_) +import Data.List (foldl') import Data.Maybe (isJust, fromJust, isNothing, catMaybes) import Data.Typeable (Typeable) import Control.Category ((>>>)) @@ -70,7 +72,9 @@ , catches , finally ) -import Control.Concurrent (forkIO, forkIOWithUnmask, killThread) +import Control.Concurrent (forkIO, killThread) +import Control.Distributed.Process.Internal.BiMultiMap (BiMultiMap) +import qualified Control.Distributed.Process.Internal.BiMultiMap as BiMultiMap import Control.Distributed.Process.Internal.StrictMVar ( newMVar , withMVar @@ -524,7 +528,8 @@ . (incomingFrom theirAddr ^: Set.insert cid) $ st ) - else invalidRequest cid st + else invalidRequest cid st $ + "attempt to connect with unsupported reliability " ++ show rel NT.Received cid payload -> case st ^. incomingAt cid of Just (_, ToProc pid weakQueue) -> do @@ -557,7 +562,9 @@ Just proc -> go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st) Nothing -> - invalidRequest cid st + -- incoming attempt to connect to unknown process - might + -- be dead already + go (incomingAt cid ^= Nothing $ st) SendPortIdentifier chId -> do let lcid = sendPortLocalId chId lpid = processLocalId (sendPortProcessId chId) @@ -569,19 +576,28 @@ Just channel -> go (incomingAt cid ^= Just (src, ToChan channel) $ st) Nothing -> - invalidRequest cid st + invalidRequest cid st $ + "incoming attempt to connect to unknown channel of" + ++ " process " ++ show (sendPortProcessId chId) Nothing -> - invalidRequest cid st + -- incoming attempt to connect to channel of unknown + -- process - might be dead already + go (incomingAt cid ^= Nothing $ st) NodeIdentifier nid -> if nid == localNodeId node then go (incomingAt cid ^= Just (src, ToNode) $ st) - else invalidRequest cid st + else invalidRequest cid st $ + "incoming attempt to connect to a different node -" + ++ " I'm " ++ show (localNodeId node) + ++ " but the remote peer wants to connect to " + ++ show nid Nothing -> invalidRequest cid st + "message received from an unknown connection" NT.ConnectionClosed cid -> case st ^. incomingAt cid of Nothing -> - invalidRequest cid st + invalidRequest cid st "closed unknown connection" Just (src, _) -> do trace node (MxDisconnected cid src) go ( (incomingAt cid ^= Nothing) @@ -612,14 +628,16 @@ -- and we just give up fail "Cloud Haskell fatal error: received unexpected multicast" - invalidRequest :: NT.ConnectionId -> ConnectionState -> IO () - invalidRequest cid st = do + invalidRequest :: NT.ConnectionId -> ConnectionState -> String -> IO () + invalidRequest cid st msg = do -- TODO: We should treat this as a fatal error on the part of the remote -- node. That is, we should report the remote node as having died, and we -- should close incoming connections (this requires a Transport layer -- extension). - traceEventFmtIO node "" [(TraceStr " [network] invalid request: "), - (Trace cid)] + traceEventFmtIO node "" [ TraceStr $ " [network] invalid request" + ++ " (" ++ msg ++ "): " + , (Trace cid) + ] go ( incomingAt cid ^= Nothing $ st ) @@ -642,9 +660,9 @@ data NCState = NCState { -- Mapping from remote processes to linked local processes - _links :: !(Map Identifier (Set ProcessId)) + _links :: !(BiMultiMap Identifier ProcessId ()) -- Mapping from remote processes to monitoring local processes - , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef))) + , _monitors :: !(BiMultiMap Identifier ProcessId MonitorRef) -- Process registry: names and where they live, mapped to the PIDs , _registeredHere :: !(Map String ProcessId) , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)]) @@ -660,8 +678,8 @@ ) initNCState :: NCState -initNCState = NCState { _links = Map.empty - , _monitors = Map.empty +initNCState = NCState { _links = BiMultiMap.empty + , _monitors = BiMultiMap.empty , _registeredHere = Map.empty , _registeredOnNodes = Map.empty } @@ -804,8 +822,8 @@ case (shouldLink, isLocal node (ProcessIdentifier from)) of (True, _) -> -- [Unified: first rule] case mRef of - Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref) - Nothing -> modify' $ linksFor them ^: Set.insert from + Just ref -> modify' $ monitors ^: BiMultiMap.insert them from ref + Nothing -> modify' $ links ^: BiMultiMap.insert them from () (False, True) -> -- [Unified: second rule] notifyDied from them DiedUnknownId mRef (False, False) -> -- [Unified: third rule] @@ -829,7 +847,7 @@ postAsMessage from $ DidUnlinkNode nid SendPortIdentifier cid -> postAsMessage from $ DidUnlinkPort cid - modify' $ linksFor them ^: Set.delete from + modify' $ links ^: BiMultiMap.delete them from () -- [Unified: Table 11] ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC () @@ -837,7 +855,7 @@ node <- ask when (isLocal node (ProcessIdentifier from)) $ postAsMessage from $ DidUnmonitor ref - modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref) + modify' $ monitors ^: BiMultiMap.delete (monitorRefIdent ref) from ref -- [Unified: Table 12] ncEffectDied :: Identifier -> DiedReason -> NC () @@ -852,7 +870,7 @@ let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False forM_ (Map.toList affectedLinks) $ \(them, uss) -> - forM_ uss $ \us -> + forM_ uss $ \(us, _) -> when (localOnly <= isLocal node (ProcessIdentifier us)) $ notifyDied us them reason Nothing @@ -861,8 +879,18 @@ when (localOnly <= isLocal node (ProcessIdentifier us)) $ notifyDied us them reason (Just ref) - modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons) + 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) + unaffectedLinks' = deleteDeads unaffectedLinks + unaffectedMons' = deleteDeads unaffectedMons + modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons') + modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid) remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>= @@ -1031,8 +1059,9 @@ Nothing -> dispatch (isLocal node (ProcessIdentifier from)) from (ProcessInfoNone DiedUnknownId) Just proc -> do - itsLinks <- gets (^. linksFor them) - itsMons <- gets (^. monitorsFor them) + itsLinks <- Set.map fst . BiMultiMap.lookup them <$> + gets (^. links) + itsMons <- BiMultiMap.lookup them <$> gets (^. monitors) registered <- gets (^. registeredHere) size <- liftIO $ queueSize $ processQueue $ proc @@ -1046,7 +1075,7 @@ , infoMonitors = Set.toList itsMons , infoLinks = Set.toList itsLinks } - where dispatch :: (Serializable a, Show a) + where dispatch :: (Serializable a) => Bool -> ProcessId -> a @@ -1068,8 +1097,8 @@ NodeStats { nodeStatsNode = localNodeId node , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere - , nodeStatsMonitors = Map.size $ ncState ^. monitors - , nodeStatsLinks = Map.size $ ncState ^. links + , nodeStatsMonitors = BiMultiMap.size $ ncState ^. monitors + , nodeStatsLinks = BiMultiMap.size $ ncState ^. links , nodeStatsProcesses = Map.size (nodeState ^. localProcesses) } postAsMessage from stats @@ -1190,10 +1219,10 @@ -- Accessors -- -------------------------------------------------------------------------------- -links :: Accessor NCState (Map Identifier (Set ProcessId)) +links :: Accessor NCState (BiMultiMap Identifier ProcessId ()) links = accessor _links (\ls st -> st { _links = ls }) -monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef))) +monitors :: Accessor NCState (BiMultiMap Identifier ProcessId MonitorRef) monitors = accessor _monitors (\ms st -> st { _monitors = ms }) registeredHere :: Accessor NCState (Map String ProcessId) @@ -1202,12 +1231,6 @@ registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)]) registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry }) -linksFor :: Identifier -> Accessor NCState (Set ProcessId) -linksFor ident = links >>> DAC.mapDefault Set.empty ident - -monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef)) -monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident - registeredHereFor :: String -> Accessor NCState (Maybe ProcessId) registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident @@ -1234,10 +1257,12 @@ -- * the notifications for typed channels to that process. -- -- See https://github.com/haskell/containers/issues/14 for the bang on _v. -splitNotif :: Identifier - -> Map Identifier a - -> (Map Identifier a, Map Identifier a) -splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k) +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) -------------------------------------------------------------------------------- -- Auxiliary -- diff --git a/src/Control/Distributed/Process/Serializable.hs b/src/Control/Distributed/Process/Serializable.hs --- a/src/Control/Distributed/Process/Serializable.hs +++ b/src/Control/Distributed/Process/Serializable.hs @@ -30,12 +30,10 @@ import GHC.Fingerprint.Type (Fingerprint(..)) import Data.ByteString (ByteString) import qualified Data.ByteString as BS -import qualified Data.ByteString.Internal as BSI ( unsafeCreate - , inlinePerformIO - , toForeignPtr - ) +import qualified Data.ByteString.Internal as BSI ( unsafeCreate, toForeignPtr ) import Foreign.Storable (pokeByteOff, peekByteOff, sizeOf) import Foreign.ForeignPtr (withForeignPtr) +import System.IO.Unsafe (unsafePerformIO) -- | Reification of 'Serializable' (see "Control.Distributed.Process.Closure") data SerializableDict a where @@ -63,7 +61,7 @@ decodeFingerprint bs | BS.length bs /= sizeOfFingerprint = throw $ userError "decodeFingerprint: Invalid length" - | otherwise = BSI.inlinePerformIO $ do + | otherwise = unsafePerformIO $ do let (fp, offset, _) = BSI.toForeignPtr bs withForeignPtr fp $ \p -> peekByteOff p offset