git.iko.soy/distributed-process
on 07c5422 help sign in
distributed-process/src/Control/Distributed/Process/Node.hs
1223 lines 48 kB history raw
1{-# LANGUAGE CPP #-}
2{-# LANGUAGE DeriveDataTypeable #-}
3{-# LANGUAGE ScopedTypeVariables #-}
4{-# LANGUAGE RankNTypes #-}
5{-# LANGUAGE BangPatterns #-}
6{-# LANGUAGE GeneralizedNewtypeDeriving #-}
7{-# LANGUAGE RecordWildCards #-}
9-- | Local nodes
10--
11module Control.Distributed.Process.Node
12 ( LocalNode
13 , newLocalNode
14 , closeLocalNode
15 , forkProcess
16 , runProcess
17 , initRemoteTable
18 , localNodeId
19 ) where
21-- TODO: Calls to 'sendBinary' and co (by the NC) may stall the node controller.
23#if ! MIN_VERSION_base(4,6,0)
24import Prelude hiding (catch)
25#endif
27import System.IO (fixIO, hPutStrLn, stderr)
28import System.Mem.Weak (Weak, deRefWeak)
29import qualified Data.ByteString.Lazy as BSL (fromChunks)
30import Data.Binary (decode)
31import Data.Map (Map)
32import qualified Data.Map as Map
33 ( empty
34 , toList
35 , fromList
36 , filter
37 , partitionWithKey
38 , elems
39 , size
40 , filterWithKey
41 , foldlWithKey
42 )
43import Data.Set (Set)
44import qualified Data.Set as Set
45 ( empty
46 , insert
47 , delete
48 , member
49 , toList
50 )
51import Data.Foldable (forM_)
52import Data.Maybe (isJust, fromJust, isNothing, catMaybes)
53import Data.Typeable (Typeable)
54import Control.Category ((>>>))
55import Control.Applicative (Applicative, (<$>))
56import Control.Monad (void, when, join)
57import Control.Monad.IO.Class (MonadIO, liftIO)
58import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets)
59import qualified Control.Monad.State.Strict as StateT (get, put)
60import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)
61import Control.Exception
62 ( throwIO
63 , AsyncException(ThreadKilled)
64 , SomeException
65 , Exception
66 , throwTo
67 , uninterruptibleMask_
68 )
69import qualified Control.Exception as Exception
70 ( Handler(..)
71 , catch
72 , catches
73 , finally
74 )
75import Control.Concurrent (forkIO, forkIOWithUnmask, killThread, myThreadId)
76import Control.Distributed.Process.Internal.StrictMVar
77 ( newMVar
78 , withMVar
79 , modifyMVarMasked
80 , modifyMVar_
81 , modifyMVar
82 , newEmptyMVar
83 , putMVar
84 , takeMVar
85 )
86import Control.Concurrent.Chan (newChan, writeChan, readChan)
87import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)
88import Control.Concurrent.STM
89 ( atomically
90 )
91import Control.Distributed.Process.Internal.CQueue
92 ( CQueue
93 , enqueue
94 , newCQueue
95 , mkWeakCQueue
96 , queueSize
97 )
98import qualified Network.Transport as NT
99 ( Transport
100 , EndPoint
101 , newEndPoint
102 , receive
103 , Event(..)
104 , EventErrorCode(..)
105 , TransportError(..)
106 , address
107 , closeEndPoint
108 , Connection
109 , ConnectionId
110 , close
111 , EndPointAddress
112 , Reliability(ReliableOrdered)
113 )
114import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
115import System.Random (randomIO)
116import Control.Distributed.Static (RemoteTable, Closure)
117import qualified Control.Distributed.Static as Static
118 ( unclosure
119 , initRemoteTable
120 )
121import Control.Distributed.Process.Internal.Types
122 ( NodeId(..)
123 , LocalProcessId(..)
124 , ProcessId(..)
125 , LocalNode(..)
126 , MxEventBus(..)
127 , LocalNodeState(..)
128 , ValidLocalNodeState(..)
129 , withValidLocalState
130 , modifyValidLocalState
131 , LocalProcess(..)
132 , LocalProcessState(..)
133 , Process(..)
134 , DiedReason(..)
135 , NCMsg(..)
136 , ProcessSignal(..)
137 , localPidCounter
138 , localPidUnique
139 , localProcessWithId
140 , localProcesses
141 , localConnections
142 , forever'
143 , MonitorRef(..)
144 , NodeClosedException(..)
145 , ProcessMonitorNotification(..)
146 , NodeMonitorNotification(..)
147 , PortMonitorNotification(..)
148 , ProcessExitException(..)
149 , ProcessLinkException(..)
150 , NodeLinkException(..)
151 , PortLinkException(..)
152 , DidUnmonitor(..)
153 , DidUnlinkProcess(..)
154 , DidUnlinkNode(..)
155 , DidUnlinkPort(..)
156 , SpawnRef
157 , DidSpawn(..)
158 , Message(..)
159 , TypedChannel(..)
160 , Identifier(..)
161 , nodeOf
162 , ProcessInfo(..)
163 , ProcessInfoNone(..)
164 , NodeStats(..)
165 , SendPortId(..)
166 , typedChannelWithId
167 , RegisterReply(..)
168 , WhereIsReply(..)
169 , payloadToMessage
170 , messageToPayload
171 , createUnencodedMessage
172 , unsafeCreateUnencodedMessage
173 , runLocalProcess
174 , firstNonReservedProcessId
175 , ImplicitReconnect(WithImplicitReconnect,NoImplicitReconnect)
176 )
177import Control.Distributed.Process.Management.Internal.Agent
178 ( mxAgentController
179 )
180import qualified Control.Distributed.Process.Management.Internal.Table as Table
181 ( mxTableCoordinator
182 , startTableCoordinator
183 )
184import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace
185 ( remoteTable
186 )
187import Control.Distributed.Process.Management.Internal.Trace.Tracer
188 ( defaultTracer
189 )
190import Control.Distributed.Process.Management.Internal.Trace.Types
191 ( TraceArg(..)
192 , traceEvent
193 , traceLogFmt
194 , enableTrace
195 )
196import Control.Distributed.Process.Management.Internal.Types
197 ( MxEvent(..)
198 )
199import Control.Distributed.Process.Serializable (Serializable)
200import Control.Distributed.Process.Internal.Messaging
201 ( sendBinary
202 , sendPayload
203 , closeImplicitReconnections
204 , impliesDeathOf
205 )
206import Control.Distributed.Process.Internal.Primitives
207 ( register
208 , receiveWait
209 , match
210 , sendChan
211 , try
212 , unwrapMessage
213 )
214import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))
215import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
216import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)
217import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC
218 ( mapMaybe
219 , mapDefault
220 )
221import Unsafe.Coerce
223--------------------------------------------------------------------------------
224-- Initialization --
225--------------------------------------------------------------------------------
227initRemoteTable :: RemoteTable
228initRemoteTable = Trace.remoteTable $ BuiltIn.remoteTable Static.initRemoteTable
230-- | Initialize a new local node.
231newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
232newLocalNode transport rtable = do
233 mEndPoint <- NT.newEndPoint transport
234 case mEndPoint of
235 Left ex -> throwIO ex
236 Right endPoint -> do
237 localNode <- createBareLocalNode endPoint rtable
238 startServiceProcesses localNode
239 return localNode
241-- | Create a new local node (without any service processes running)
242createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
243createBareLocalNode endPoint rtable = do
244 unq <- randomIO
245 state <- newMVar $ LocalNodeValid $ ValidLocalNodeState
246 { _localProcesses = Map.empty
247 , _localPidCounter = firstNonReservedProcessId
248 , _localPidUnique = unq
249 , _localConnections = Map.empty
250 }
251 ctrlChan <- newChan
252 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
253 , localEndPoint = endPoint
254 , localState = state
255 , localCtrlChan = ctrlChan
256 , localEventBus = MxEventBusInitialising
257 , remoteTable = rtable
258 }
259 tracedNode <- startMxAgent node
261 -- Once the NC terminates, the endpoint isn't much use,
262 void $ forkIO $ Exception.finally (runNodeController tracedNode)
263 (NT.closeEndPoint (localEndPoint node))
265 -- whilst a closed/failing endpoint will terminate the NC
266 void $ forkIO $ Exception.finally (handleIncomingMessages tracedNode)
267 (stopNC node)
269 return tracedNode
270 where
271 stopNC node =
272 writeChan (localCtrlChan node) NCMsg
273 { ctrlMsgSender = NodeIdentifier (localNodeId node)
274 , ctrlMsgSignal = SigShutdown
275 }
277startMxAgent :: LocalNode -> IO LocalNode
278startMxAgent node = do
279 -- see note [tracer/forkProcess races]
280 let fork = forkProcess node
281 mv <- MVar.newEmptyMVar
282 pid <- fork $ mxAgentController fork mv
283 (tracer', wqRef, mxNew') <- MVar.takeMVar mv
284 return node { localEventBus = (MxEventBus pid tracer' wqRef mxNew') }
286startDefaultTracer :: LocalNode -> IO ()
287startDefaultTracer node' = do
288 let t = localEventBus node'
289 case t of
290 MxEventBus _ (Tracer pid _) _ _ -> do
291 runProcess node' $ register "trace.controller" pid
292 pid' <- forkProcess node' defaultTracer
293 enableTrace (localEventBus node') pid'
294 runProcess node' $ register "tracer.initial" pid'
295 _ -> return ()
297-- TODO: we need a better mechanism for defining and registering services
299-- | Start and register the service processes on a node
300startServiceProcesses :: LocalNode -> IO ()
301startServiceProcesses node = do
302 -- tracing /spawns/ relies on the tracer being enabled, but we start
303 -- the default tracer first, even though it might @nsend@ to the logger
304 -- before /that/ process has started - this is a totally harmless race
305 -- however, so we deliberably ignore it
306 startDefaultTracer node
307 tableCoordinatorPid <- fork $ Table.startTableCoordinator fork
308 runProcess node $ register Table.mxTableCoordinator tableCoordinatorPid
309 logger <- forkProcess node loop
310 runProcess node $ do
311 register "logger" logger
312 -- The trace.logger is used for tracing to the console to avoid feedback
313 -- loops during tracing if the user reregisters the "logger" with a custom
314 -- process which uses 'send' or other primitives which are traced.
315 register "trace.logger" logger
316 where
317 fork = forkProcess node
319 loop = do
320 receiveWait
321 [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do
322 liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string
323 loop
324 , match $ \((time, string) :: (String, String)) -> do
325 -- this is a 'trace' message from the local node tracer
326 liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string
327 loop
328 , match $ \(ch :: SendPort ()) -> -- a shutdown request
329 sendChan ch ()
330 ]
332-- | Force-close a local node, killing all processes on that node.
333closeLocalNode :: LocalNode -> IO ()
334closeLocalNode node = do
335 -- Kill processes after refilling the mvar. Otherwise, there is potential for
336 -- deadlock as a dying process tries to get the mvar while masking exceptions
337 -- uninterruptibly.
338 join $ modifyMVar (localState node) $ \st -> case st of
339 LocalNodeValid vst -> do
340 return ( LocalNodeClosed
341 , forM_ (vst ^. localProcesses) $ \lproc ->
342 -- Semantics of 'throwTo' guarantee that target thread will get
343 -- delivered an exception. Therefore, target thread will be
344 -- killed eventually and that's as good as we can do. No need
345 -- to wait for thread to actually finish dying.
346 killThread (processThread lproc)
347 )
348 LocalNodeClosed -> return (LocalNodeClosed, return ())
349 -- This call will have the effect of shutting down the NC as well (see
350 -- 'createBareLocalNode').
351 NT.closeEndPoint (localEndPoint node)
353-- | Run a process on a local node and wait for it to finish
354runProcess :: LocalNode -> Process () -> IO ()
355runProcess node proc = do
356 done <- newEmptyMVar
357 -- TODO; When forkProcess inherits the masking state, protect the forked
358 -- thread against async exceptions that could occur before 'try' is evaluated.
359 void $ forkProcess node $ try proc >>= liftIO . putMVar done
360 takeMVar done >>= either (throwIO :: SomeException -> IO a) return
362-- | Spawn a new process on a local node
363forkProcess :: LocalNode -> Process () -> IO ProcessId
364forkProcess node proc =
365 modifyMVarMasked (localState node) startProcess
366 where
367 startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)
368 startProcess (LocalNodeValid vst) = do
369 let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter
370 , lpidUnique = vst ^. localPidUnique
371 }
372 let pid = ProcessId { processNodeId = localNodeId node
373 , processLocalId = lpid
374 }
375 pst <- newMVar LocalProcessState { _monitorCounter = 0
376 , _spawnCounter = 0
377 , _channelCounter = 0
378 , _typedChannels = Map.empty
379 }
380 queue <- newCQueue
381 weakQueue <- mkWeakCQueue queue (return ())
382 (_, lproc) <- fixIO $ \ ~(tid, _) -> do
383 let lproc = LocalProcess { processQueue = queue
384 , processWeakQ = weakQueue
385 , processId = pid
386 , processState = pst
387 , processThread = tid
388 , processNode = node
389 }
390 tid' <- uninterruptibleMask_ $ forkIOWithUnmask $ \unmask -> do
391 reason <- Exception.catches
392 (unmask $ runLocalProcess lproc proc >> return DiedNormal)
393 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
394 mMsg <- unwrapMessage msg :: IO (Maybe String)
395 case mMsg of
396 Nothing -> return $ DiedException $ show ex
397 Just m -> return $ DiedException ("exit-from=" ++ (show from) ++ ",reason=" ++ m)))
398 , (Exception.Handler
399 (return . DiedException . (show :: SomeException -> String)))]
401 -- [Unified: Table 4, rules termination and exiting]
402 mconns <- modifyValidLocalState node (cleanupProcess pid)
403 -- XXX: Revisit after agreeing on the bigger picture for the semantics
404 -- of transport operations.
405 -- https://github.com/haskell-distributed/distributed-process/issues/204
406 forM_ mconns $ forkIO . mapM_ NT.close
408 writeChan (localCtrlChan node) NCMsg
409 { ctrlMsgSender = ProcessIdentifier pid
410 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
411 }
412 return (tid', lproc)
414 -- see note [tracer/forkProcess races]
415 trace node (MxSpawned pid)
417 if lpidCounter lpid == maxBound
418 then do
419 -- TODO: this doesn't look right at all - how do we know
420 -- that newUnique represents a process id that is available!?
421 newUnique <- randomIO
422 return ( LocalNodeValid
423 $ (localProcessWithId lpid ^= Just lproc)
424 . (localPidCounter ^= firstNonReservedProcessId)
425 . (localPidUnique ^= newUnique)
426 $ vst
427 , pid
428 )
429 else
430 return ( LocalNodeValid
431 $ (localProcessWithId lpid ^= Just lproc)
432 . (localPidCounter ^: (+ 1))
433 $ vst
434 , pid
435 )
436 startProcess LocalNodeClosed =
437 throwIO $ NodeClosedException $ localNodeId node
439 cleanupProcess :: ProcessId
440 -> ValidLocalNodeState
441 -> IO (ValidLocalNodeState, [NT.Connection])
442 cleanupProcess pid vst = do
443 let pid' = ProcessIdentifier pid
444 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)
445 return ( (localProcessWithId (processLocalId pid) ^= Nothing)
446 . (localConnections ^= unaffected)
447 $ vst
448 , map fst $ Map.elems affected
449 )
451-- note [tracer/forkProcess races]
452--
453-- Our startTracing function uses forkProcess to start the trace controller
454-- process, and of course forkProcess attempts to call traceEvent once the
455-- process has started. This is harmless, as the localEventBus is not updated
456-- until /after/ the initial forkProcess completes, so the first call to
457-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
458--
460--------------------------------------------------------------------------------
461-- Handle incoming messages --
462--------------------------------------------------------------------------------
464type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
466data IncomingTarget =
467 Uninit
468 | ToProc ProcessId (Weak (CQueue Message))
469 | ToChan TypedChannel
470 | ToNode
472data ConnectionState = ConnectionState {
473 _incoming :: !(Map NT.ConnectionId IncomingConnection)
474 , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId))
475 }
477initConnectionState :: ConnectionState
478initConnectionState = ConnectionState {
479 _incoming = Map.empty
480 , _incomingFrom = Map.empty
481 }
483incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection)
484incoming = accessor _incoming (\conns st -> st { _incoming = conns })
486incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection)
487incomingAt cid = incoming >>> DAC.mapMaybe cid
489incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId)
490incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
491 where
492 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
494handleIncomingMessages :: LocalNode -> IO ()
495handleIncomingMessages node = go initConnectionState
496 `Exception.catch` \(NodeClosedException _) -> return ()
497 where
498 go :: ConnectionState -> IO ()
499 go !st = do
500 event <- NT.receive endpoint
501 case event of
502 NT.ConnectionOpened cid rel theirAddr ->
503 if rel == NT.ReliableOrdered
504 then
505 trace node (MxConnected cid theirAddr)
506 >> go (
507 (incomingAt cid ^= Just (theirAddr, Uninit))
508 . (incomingFrom theirAddr ^: Set.insert cid)
509 $ st
510 )
511 else invalidRequest cid st
512 NT.Received cid payload ->
513 case st ^. incomingAt cid of
514 Just (_, ToProc pid weakQueue) -> do
515 mQueue <- deRefWeak weakQueue
516 forM_ mQueue $ \queue -> do
517 -- TODO: if we find that the queue is Nothing, should we remove
518 -- it from the NC state? (and same for channels, below)
519 let msg = payloadToMessage payload
520 enqueue queue msg -- 'enqueue' is strict
521 trace node (MxReceived pid msg)
522 go st
523 Just (_, ToChan (TypedChannel chan')) -> do
524 mChan <- deRefWeak chan'
525 -- If mChan is Nothing, the process has given up the read end of
526 -- the channel and we simply ignore the incoming message
527 forM_ mChan $ \chan -> atomically $
528 -- We make sure the message is fully decoded when it is enqueued
529 writeTQueue chan $! decode (BSL.fromChunks payload)
530 go st
531 Just (_, ToNode) -> do
532 let ctrlMsg = decode . BSL.fromChunks $ payload
533 writeChan ctrlChan $! ctrlMsg
534 go st
535 Just (src, Uninit) ->
536 case decode (BSL.fromChunks payload) of
537 ProcessIdentifier pid -> do
538 let lpid = processLocalId pid
539 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
540 case mProc of
541 Just proc ->
542 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
543 Nothing ->
544 invalidRequest cid st
545 SendPortIdentifier chId -> do
546 let lcid = sendPortLocalId chId
547 lpid = processLocalId (sendPortProcessId chId)
548 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
549 case mProc of
550 Just proc -> do
551 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
552 case mChannel of
553 Just channel ->
554 go (incomingAt cid ^= Just (src, ToChan channel) $ st)
555 Nothing ->
556 invalidRequest cid st
557 Nothing ->
558 invalidRequest cid st
559 NodeIdentifier nid ->
560 if nid == localNodeId node
561 then go (incomingAt cid ^= Just (src, ToNode) $ st)
562 else invalidRequest cid st
563 Nothing ->
564 invalidRequest cid st
565 NT.ConnectionClosed cid ->
566 case st ^. incomingAt cid of
567 Nothing ->
568 invalidRequest cid st
569 Just (src, _) -> do
570 trace node (MxDisconnected cid src)
571 go ( (incomingAt cid ^= Nothing)
572 . (incomingFrom src ^: Set.delete cid)
573 $ st
574 )
575 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
576 -- [Unified table 9, rule node_disconnect]
577 let nid = NodeIdentifier $ NodeId theirAddr
578 writeChan ctrlChan NCMsg
579 { ctrlMsgSender = nid
580 , ctrlMsgSignal = Died nid DiedDisconnect
581 }
582 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
583 closeImplicitReconnections node nid
584 go ( (incomingFrom theirAddr ^= Set.empty)
585 . (incoming ^: Map.filterWithKey (const . notLost))
586 $ st
587 )
588 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
589 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
590 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
591 fail $ "Cloud Haskell fatal error: transport failed: " ++ str
592 NT.EndPointClosed ->
593 return ()
594 NT.ReceivedMulticast _ _ ->
595 -- If we received a multicast message, something went horribly wrong
596 -- and we just give up
597 fail "Cloud Haskell fatal error: received unexpected multicast"
599 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
600 invalidRequest cid st = do
601 -- TODO: We should treat this as a fatal error on the part of the remote
602 -- node. That is, we should report the remote node as having died, and we
603 -- should close incoming connections (this requires a Transport layer
604 -- extension).
605 traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),
606 (Trace cid)]
607 go ( incomingAt cid ^= Nothing
608 $ st
609 )
611 endpoint = localEndPoint node
612 ctrlChan = localCtrlChan node
614--------------------------------------------------------------------------------
615-- Top-level access to the node controller --
616--------------------------------------------------------------------------------
618runNodeController :: LocalNode -> IO ()
619runNodeController node =
620 runReaderT (evalStateT (unNC nodeController) initNCState) node
621 `Exception.catch` \(NodeClosedException _) -> return ()
623--------------------------------------------------------------------------------
624-- Internal data types --
625--------------------------------------------------------------------------------
627data NCState = NCState
628 { -- Mapping from remote processes to linked local processes
629 _links :: !(Map Identifier (Set ProcessId))
630 -- Mapping from remote processes to monitoring local processes
631 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
632 -- Process registry: names and where they live, mapped to the PIDs
633 , _registeredHere :: !(Map String ProcessId)
634 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
635 }
637newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
638 deriving ( Applicative
639 , Functor
640 , Monad
641 , MonadIO
642 , MonadState NCState
643 , MonadReader LocalNode
644 )
646initNCState :: NCState
647initNCState = NCState { _links = Map.empty
648 , _monitors = Map.empty
649 , _registeredHere = Map.empty
650 , _registeredOnNodes = Map.empty
651 }
653-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This
654-- type is deliberately not exported so it cannot be caught explicitly.
655data ProcessKillException =
656 ProcessKillException !ProcessId !String
657 deriving (Typeable)
659instance Exception ProcessKillException
660instance Show ProcessKillException where
661 show (ProcessKillException pid reason) =
662 "killed-by=" ++ show pid ++ ",reason=" ++ reason
664ncSendToProcess :: ProcessId -> Message -> NC ()
665ncSendToProcess pid msg = do
666 node <- ask
667 if processNodeId pid == localNodeId node
668 then ncEffectLocalSend node pid msg
669 else liftIO $ sendBinary node
670 (NodeIdentifier $ localNodeId node)
671 (NodeIdentifier $ processNodeId pid)
672 WithImplicitReconnect
673 NCMsg { ctrlMsgSender = NodeIdentifier $ localNodeId node
674 , ctrlMsgSignal = UnreliableSend (processLocalId pid) msg
675 }
677ncSendToNode :: NodeId -> NCMsg -> NC ()
678ncSendToNode to msg = do
679 node <- ask
680 liftIO $ if to == localNodeId node
681 then writeChan (localCtrlChan node) $! msg
682 else sendBinary node
683 (NodeIdentifier $ localNodeId node)
684 (NodeIdentifier to)
685 WithImplicitReconnect
686 msg
688--------------------------------------------------------------------------------
689-- Tracing/Debugging --
690--------------------------------------------------------------------------------
692-- [Issue #104 / DP-13]
694traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
695traceNotifyDied node ident reason =
696 -- TODO: sendPortDied notifications
697 liftIO $ withLocalTracer node $ \t ->
698 case ident of
699 (NodeIdentifier nid) -> traceEvent t (MxNodeDied nid reason)
700 (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)
701 _ -> return ()
703traceEventFmtIO :: LocalNode
704 -> String
705 -> [TraceArg]
706 -> IO ()
707traceEventFmtIO node fmt args =
708 withLocalTracer node $ \t -> traceLogFmt t fmt args
710trace :: LocalNode -> MxEvent -> IO ()
711trace node ev = withLocalTracer node $ \t -> traceEvent t ev
713withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()
714withLocalTracer node act = act (localEventBus node)
716--------------------------------------------------------------------------------
717-- Core functionality --
718--------------------------------------------------------------------------------
720-- [Unified: Table 7]
721nodeController :: NC ()
722nodeController = do
723 node <- ask
724 forever' $ do
725 msg <- liftIO $ readChan (localCtrlChan node)
727 -- [Unified: Table 7, rule nc_forward]
728 case destNid (ctrlMsgSignal msg) of
729 Just nid' | nid' /= localNodeId node ->
730 ncSendToNode nid' msg
731 _ ->
732 return ()
734 case msg of
735 NCMsg (ProcessIdentifier from) (Link them) ->
736 ncEffectMonitor from them Nothing
737 NCMsg (ProcessIdentifier from) (Monitor ref) ->
738 ncEffectMonitor from (monitorRefIdent ref) (Just ref)
739 NCMsg (ProcessIdentifier from) (Unlink them) ->
740 ncEffectUnlink from them
741 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->
742 ncEffectUnmonitor from ref
743 NCMsg _from (Died ident reason) ->
744 ncEffectDied ident reason
745 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
746 ncEffectSpawn from proc ref
747 NCMsg (ProcessIdentifier from) (Register label atnode pid force) ->
748 ncEffectRegister from label atnode pid force
749 NCMsg (ProcessIdentifier from) (WhereIs label) ->
750 ncEffectWhereIs from label
751 NCMsg _ (NamedSend label msg') ->
752 ncEffectNamedSend label msg'
753 NCMsg _ (UnreliableSend lpid msg') ->
754 ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg'
755 NCMsg _ (LocalSend to msg') ->
756 ncEffectLocalSend node to msg'
757 NCMsg _ (LocalPortSend to msg') ->
758 ncEffectLocalPortSend to msg'
759 NCMsg (ProcessIdentifier from) (Kill to reason) ->
760 ncEffectKill from to reason
761 NCMsg (ProcessIdentifier from) (Exit to reason) ->
762 ncEffectExit from to reason
763 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
764 ncEffectGetInfo from pid
765 NCMsg _ SigShutdown ->
766 liftIO $ do
767 NT.closeEndPoint (localEndPoint node)
768 `Exception.finally` throwIO ThreadKilled
769 -- ThreadKilled seems to make more sense than fail/error here
770 NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->
771 ncEffectGetNodeStats from nid
772 unexpected ->
773 error $ "nodeController: unexpected message " ++ show unexpected
775-- [Unified: Table 10]
776ncEffectMonitor :: ProcessId -- ^ Who's watching?
777 -> Identifier -- ^ Who's being watched?
778 -> Maybe MonitorRef -- ^ 'Nothing' to link
779 -> NC ()
780ncEffectMonitor from them mRef = do
781 node <- ask
782 shouldLink <-
783 if not (isLocal node them)
784 then return True
785 else isValidLocalIdentifier them
786 case (shouldLink, isLocal node (ProcessIdentifier from)) of
787 (True, _) -> -- [Unified: first rule]
788 case mRef of
789 Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)
790 Nothing -> modify' $ linksFor them ^: Set.insert from
791 (False, True) -> -- [Unified: second rule]
792 notifyDied from them DiedUnknownId mRef
793 (False, False) -> -- [Unified: third rule]
794 -- TODO: this is the right sender according to the Unified semantics,
795 -- but perhaps having 'them' as the sender would make more sense
796 -- (see also: notifyDied)
797 ncSendToNode (processNodeId from) $ NCMsg
798 { ctrlMsgSender = NodeIdentifier (localNodeId node)
799 , ctrlMsgSignal = Died them DiedUnknownId
800 }
802-- [Unified: Table 11]
803ncEffectUnlink :: ProcessId -> Identifier -> NC ()
804ncEffectUnlink from them = do
805 node <- ask
806 when (isLocal node (ProcessIdentifier from)) $
807 case them of
808 ProcessIdentifier pid ->
809 postAsMessage from $ DidUnlinkProcess pid
810 NodeIdentifier nid ->
811 postAsMessage from $ DidUnlinkNode nid
812 SendPortIdentifier cid ->
813 postAsMessage from $ DidUnlinkPort cid
814 modify' $ linksFor them ^: Set.delete from
816-- [Unified: Table 11]
817ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
818ncEffectUnmonitor from ref = do
819 node <- ask
820 when (isLocal node (ProcessIdentifier from)) $
821 postAsMessage from $ DidUnmonitor ref
822 modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
824-- [Unified: Table 12]
825ncEffectDied :: Identifier -> DiedReason -> NC ()
826ncEffectDied ident reason = do
827 node <- ask
828 traceNotifyDied node ident reason
829 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
830 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
832-- _registry :: !(Map (String,NodeId) ProcessId)
834 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
836 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
837 forM_ uss $ \us ->
838 when (localOnly <= isLocal node (ProcessIdentifier us)) $
839 notifyDied us them reason Nothing
841 forM_ (Map.toList affectedMons) $ \(them, refs) ->
842 forM_ refs $ \(us, ref) ->
843 when (localOnly <= isLocal node (ProcessIdentifier us)) $
844 notifyDied us them reason (Just ref)
846 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
848 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
850 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
851 mapM (\(pid,nidlist) ->
852 case ident `impliesDeathOf` ProcessIdentifier pid of
853 True ->
854 do forM_ nidlist $ \(nid,_) ->
855 when (not $ isLocal node (NodeIdentifier nid))
856 (forwardNameDeath node nid)
857 return Nothing
858 False -> return $ Just (pid,nidlist) )
859 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
860 where
861 forwardNameDeath node nid = ncSendToNode nid
862 NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node)
863 , ctrlMsgSignal = Died ident reason
864 }
866-- [Unified: Table 13]
867ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
868ncEffectSpawn pid cProc ref = do
869 mProc <- unClosure cProc
870 -- If the closure does not exist, we spawn a process that throws an exception
871 -- This allows the remote node to find out what's happening
872 -- TODO:
873 let proc = case mProc of
874 Left err -> fail $ "Error: Could not resolve closure: " ++ err
875 Right p -> p
876 node <- ask
877 pid' <- liftIO $ forkProcess node proc
878 ncSendToProcess pid $ unsafeCreateUnencodedMessage $ DidSpawn ref pid'
880-- Unified semantics does not explicitly describe how to implement 'register',
881-- but mentions it's "very similar to nsend" (Table 14)
882-- We send a response indicated if the operation is invalid
883ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
884ncEffectRegister from label atnode mPid reregistration = do
885 node <- ask
886 currentVal <- gets (^. registeredHereFor label)
887 isOk <-
888 case mPid of
889 Nothing -> -- unregister request
890 return $ isJust currentVal
891 Just thepid -> -- register request
892 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
893 return $ (isNothing currentVal /= reregistration) &&
894 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
895 if isLocal node (NodeIdentifier atnode)
896 then do when isOk $
897 do modify' $ registeredHereFor label ^= mPid
898 updateRemote node currentVal mPid
899 case mPid of
900 (Just p) -> liftIO $ trace node (MxRegistered p label)
901 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
902 newVal <- gets (^. registeredHereFor label)
903 ncSendToProcess from $ unsafeCreateUnencodedMessage $
904 RegisterReply label isOk newVal
905 else let operation =
906 case reregistration of
907 True -> flip decList
908 False -> flip incList
909 in case mPid of
910 Nothing -> return ()
911 Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
912 where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
913 do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
914 forward node (processNodeId newval) (Register label atnode (Just newval) False)
915 updateRemote node Nothing (Just newval) =
916 forward node (processNodeId newval) (Register label atnode (Just newval) False)
917 updateRemote node (Just oldval) Nothing =
918 forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
919 updateRemote _ _ _ = return ()
920 maybeify f Nothing = unmaybeify $ f []
921 maybeify f (Just x) = unmaybeify $ f x
923 unmaybeify [] = Nothing
924 unmaybeify x = Just x
925 incList [] tag = [(tag,1)]
926 incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
927 incList (x:xs) tag = x : incList xs tag
928 decList [] _ = []
929 decList ((atag,1):xs) tag | atag == tag = xs
930 decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
931 decList (x:xs) tag = x:decList xs tag
932 forward node to reg =
933 when (not $ isLocal node (NodeIdentifier to)) $
934 ncSendToNode to $ NCMsg { ctrlMsgSender = ProcessIdentifier from
935 , ctrlMsgSignal = reg
936 }
939-- Unified semantics does not explicitly describe 'whereis'
940ncEffectWhereIs :: ProcessId -> String -> NC ()
941ncEffectWhereIs from label = do
942 mPid <- gets (^. registeredHereFor label)
943 ncSendToProcess from $ unsafeCreateUnencodedMessage $ WhereIsReply label mPid
945-- [Unified: Table 14]
946ncEffectNamedSend :: String -> Message -> NC ()
947ncEffectNamedSend label msg = do
948 mPid <- gets (^. registeredHereFor label)
949 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
950 forM_ mPid (`ncSendToProcess` msg)
952-- [Issue #DP-20]
953ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
954ncEffectLocalSend node to msg =
955 liftIO $ withLocalProc node to $ \p -> do
956 enqueue (processQueue p) msg
957 trace node (MxReceived to msg)
959-- [Issue #DP-20]
960ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
961ncEffectLocalPortSend from msg = do
962 node <- ask
963 let pid = sendPortProcessId from
964 cid = sendPortLocalId from
965 liftIO $ withLocalProc node pid $ \proc -> do
966 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
967 case mChan of
968 -- in the unlikely event we know nothing about this channel id,
969 -- there's little to be done - perhaps some logging/tracing though...
970 Nothing -> return ()
971 Just (TypedChannel chan') -> do
972 -- If ch is Nothing, the process has given up the read end of
973 -- the channel and we simply ignore the incoming message - this
974 ch <- deRefWeak chan'
975 forM_ ch $ \chan -> deliverChan msg chan
976 where deliverChan :: forall a . Message -> TQueue a -> IO ()
977 deliverChan (UnencodedMessage _ raw) chan' =
978 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
979 deliverChan (EncodedMessage _ _) _ =
980 -- this will not happen unless someone screws with Primitives.hs
981 error "invalid local channel delivery"
983-- [Issue #69]
984ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()
985ncEffectKill from to reason = do
986 node <- ask
987 when (isLocal node (ProcessIdentifier to)) $
988 throwException to $ ProcessKillException from reason
990-- [Issue #69]
991ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()
992ncEffectExit from to reason = do
993 node <- ask
994 when (isLocal node (ProcessIdentifier to)) $
995 throwException to $ ProcessExitException from reason
997-- [Issue #89]
998ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()
999ncEffectGetInfo from pid =
1000 let lpid = processLocalId pid
1001 them = (ProcessIdentifier pid)
1002 in do
1003 node <- ask
1004 mProc <- liftIO $ withValidLocalState node
1005 $ return . (^. localProcessWithId lpid)
1006 case mProc of
1007 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
1008 from (ProcessInfoNone DiedUnknownId)
1009 Just proc -> do
1010 itsLinks <- gets (^. linksFor them)
1011 itsMons <- gets (^. monitorsFor them)
1012 registered <- gets (^. registeredHere)
1013 size <- liftIO $ queueSize $ processQueue $ proc
1015 let reg = registeredNames registered
1016 dispatch (isLocal node (ProcessIdentifier from))
1017 from
1018 ProcessInfo {
1019 infoNode = (processNodeId pid)
1020 , infoRegisteredNames = reg
1021 , infoMessageQueueLength = size
1022 , infoMonitors = Set.toList itsMons
1023 , infoLinks = Set.toList itsLinks
1025 where dispatch :: (Serializable a, Show a)
1026 => Bool
1027 -> ProcessId
1028 -> a
1029 -> NC ()
1030 dispatch True dest pInfo = postAsMessage dest $ pInfo
1031 dispatch False dest pInfo =
1032 ncSendToProcess dest $ unsafeCreateUnencodedMessage pInfo
1034 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
1035 then (k:ks)
1036 else ks) []
1038ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()
1039ncEffectGetNodeStats from _nid = do
1040 node <- ask
1041 ncState <- StateT.get
1042 nodeState <- liftIO $ withValidLocalState node return
1043 let stats =
1044 NodeStats {
1045 nodeStatsNode = localNodeId node
1046 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
1047 , nodeStatsMonitors = Map.size $ ncState ^. monitors
1048 , nodeStatsLinks = Map.size $ ncState ^. links
1049 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
1051 postAsMessage from stats
1053--------------------------------------------------------------------------------
1054-- Auxiliary --
1055--------------------------------------------------------------------------------
1057notifyDied :: ProcessId -- ^ Who to notify?
1058 -> Identifier -- ^ Who died?
1059 -> DiedReason -- ^ How did they die?
1060 -> Maybe MonitorRef -- ^ 'Nothing' for linking
1061 -> NC ()
1062notifyDied dest src reason mRef = do
1063 node <- ask
1064 case (isLocal node (ProcessIdentifier dest), mRef, src) of
1065 (True, Just ref, ProcessIdentifier pid) ->
1066 postAsMessage dest $ ProcessMonitorNotification ref pid reason
1067 (True, Just ref, NodeIdentifier nid) ->
1068 postAsMessage dest $ NodeMonitorNotification ref nid reason
1069 (True, Just ref, SendPortIdentifier cid) ->
1070 postAsMessage dest $ PortMonitorNotification ref cid reason
1071 (True, Nothing, ProcessIdentifier pid) ->
1072 throwException dest $ ProcessLinkException pid reason
1073 (True, Nothing, NodeIdentifier pid) ->
1074 throwException dest $ NodeLinkException pid reason
1075 (True, Nothing, SendPortIdentifier pid) ->
1076 throwException dest $ PortLinkException pid reason
1077 (False, _, _) ->
1078 -- The change in sender comes from [Unified: Table 10]
1079 ncSendToNode (processNodeId dest) $ NCMsg
1080 { ctrlMsgSender = NodeIdentifier (localNodeId node)
1081 , ctrlMsgSignal = Died src reason
1084-- | [Unified: Table 8]
1085destNid :: ProcessSignal -> Maybe NodeId
1086destNid (Link ident) = Just $ nodeOf ident
1087destNid (Unlink ident) = Just $ nodeOf ident
1088destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)
1089destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
1090destNid (Spawn _ _) = Nothing
1091destNid (Register _ _ _ _) = Nothing
1092destNid (WhereIs _) = Nothing
1093destNid (NamedSend _ _) = Nothing
1094destNid (UnreliableSend _ _) = Nothing
1095-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
1096-- then when a local process dies the monitoring/linking machinery will take
1097-- care of notifying remote nodes
1098destNid (Died _ _) = Nothing
1099destNid (Kill pid _) = Just $ processNodeId pid
1100destNid (Exit pid _) = Just $ processNodeId pid
1101destNid (GetInfo pid) = Just $ processNodeId pid
1102destNid (GetNodeStats nid) = Just nid
1103destNid (LocalSend pid _) = Just $ processNodeId pid
1104destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1105destNid (SigShutdown) = Nothing
1107-- | Check if a process is local to our own node
1108isLocal :: LocalNode -> Identifier -> Bool
1109isLocal nid ident = nodeOf ident == localNodeId nid
1111-- | Lookup a local closure
1112unClosure :: Typeable a => Closure a -> NC (Either String a)
1113unClosure closure = do
1114 rtable <- remoteTable <$> ask
1115 return (Static.unclosure rtable closure)
1117-- | Check if an identifier refers to a valid local object
1118isValidLocalIdentifier :: Identifier -> NC Bool
1119isValidLocalIdentifier ident = do
1120 node <- ask
1121 liftIO . withValidLocalState node $ \nSt ->
1122 case ident of
1123 NodeIdentifier nid ->
1124 return $ nid == localNodeId node
1125 ProcessIdentifier pid -> do
1126 let mProc = nSt ^. localProcessWithId (processLocalId pid)
1127 return $ isJust mProc
1128 SendPortIdentifier cid -> do
1129 let pid = sendPortProcessId cid
1130 mProc = nSt ^. localProcessWithId (processLocalId pid)
1131 case mProc of
1132 Nothing -> return False
1133 Just proc -> withMVar (processState proc) $ \pSt -> do
1134 let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)
1135 return $ isJust mCh
1137--------------------------------------------------------------------------------
1138-- Messages to local processes --
1139--------------------------------------------------------------------------------
1141postAsMessage :: Serializable a => ProcessId -> a -> NC ()
1142postAsMessage pid = postMessage pid . createUnencodedMessage
1144postMessage :: ProcessId -> Message -> NC ()
1145postMessage pid msg = do
1146 node <- ask
1147 liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
1149throwException :: Exception e => ProcessId -> e -> NC ()
1150throwException pid e = do
1151 node <- ask
1152 -- throwTo blocks until the exception is received by the target thread.
1153 -- We fork a helper thread to avoid blocking the node controller.
1154 liftIO $ withLocalProc node pid $ \p -> void $ forkIO $ throwTo (processThread p) e
1156withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1157withLocalProc node pid p =
1158 -- By [Unified: table 6, rule missing_process] messages to dead processes
1159 -- can silently be dropped
1160 let lpid = processLocalId pid in do
1161 withValidLocalState node $ \vst ->
1162 forM_ (vst ^. localProcessWithId lpid) p
1164--------------------------------------------------------------------------------
1165-- Accessors --
1166--------------------------------------------------------------------------------
1168links :: Accessor NCState (Map Identifier (Set ProcessId))
1169links = accessor _links (\ls st -> st { _links = ls })
1171monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
1172monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1174registeredHere :: Accessor NCState (Map String ProcessId)
1175registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
1177registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1178registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1180linksFor :: Identifier -> Accessor NCState (Set ProcessId)
1181linksFor ident = links >>> DAC.mapDefault Set.empty ident
1183monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
1184monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
1186registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1187registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1189registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
1190registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1192-- | @splitNotif ident@ splits a notifications map into those
1193-- notifications that should trigger when 'ident' fails and those links that
1194-- should not.
1196-- There is a hierarchy between identifiers: failure of a node implies failure
1197-- of all processes on that node, and failure of a process implies failure of
1198-- all typed channels to that process. In other words, if 'ident' refers to a
1199-- node, then the /should trigger/ set will include
1201-- * the notifications for the node specifically
1202-- * the notifications for processes on that node, and
1203-- * the notifications for typed channels to processes on that node.
1205-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
1206-- include
1208-- * the notifications for that process specifically and
1209-- * the notifications for typed channels to that process.
1211-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1212splitNotif :: Identifier
1213 -> Map Identifier a
1214 -> (Map Identifier a, Map Identifier a)
1215splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
1217--------------------------------------------------------------------------------
1218-- Auxiliary --
1219--------------------------------------------------------------------------------
1221-- | Modify and evaluate the state
1222modify' :: MonadState s m => (s -> s) -> m ()
1223modify' f = StateT.get >>= \s -> StateT.put $! f s