distributed-process 0.6.4 → 0.6.5

raw patch · 12 files changed

+251/−60 lines, 12 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• ChangeLog +10/−0
• distributed-process.cabal +2/−1
• src/Control/Distributed/Process.hs +0/−1
• src/Control/Distributed/Process/Internal/BiMultiMap.hs +132/−0
• src/Control/Distributed/Process/Internal/CQueue.hs +24/−0
• src/Control/Distributed/Process/Internal/Closure/Explicit.hs +1/−2
• src/Control/Distributed/Process/Internal/Closure/TH.hs +11/−7
• src/Control/Distributed/Process/Internal/Messaging.hs +2/−2
• src/Control/Distributed/Process/Internal/Spawn.hs +3/−3
• src/Control/Distributed/Process/Management/Internal/Types.hs +0/−1
• src/Control/Distributed/Process/Node.hs +63/−38
• src/Control/Distributed/Process/Serializable.hs +3/−5

@@ -1,3 +1,13 @@+2016-10-12 Facundo Domínguez <facundo.dominguez@tweag.io> 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 <facundo.dominguez@tweag.io> 0.6.4  * Fixup build errors.

@@ -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,

@@ -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_)

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

@@ -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)

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

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

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

@@ -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))     ]

@@ -1,5 +1,4 @@ {-# LANGUAGE DeriveDataTypeable  #-}-{-# LANGUAGE StandaloneDeriving  #-} {-# LANGUAGE ExistentialQuantification  #-} {-# LANGUAGE GeneralizedNewtypeDeriving  #-} {-# LANGUAGE DeriveGeneric   #-}

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

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