git.iko.soy/distributed-process
on e2bfbeb help sign in
distributed-process/src/Control/Distributed/Process/Node.hs
1222 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 (NodeClosedException $ localNodeId node)
769 NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->
770 ncEffectGetNodeStats from nid
771 unexpected ->
772 error $ "nodeController: unexpected message " ++ show unexpected
774-- [Unified: Table 10]
775ncEffectMonitor :: ProcessId -- ^ Who's watching?
776 -> Identifier -- ^ Who's being watched?
777 -> Maybe MonitorRef -- ^ 'Nothing' to link
778 -> NC ()
779ncEffectMonitor from them mRef = do
780 node <- ask
781 shouldLink <-
782 if not (isLocal node them)
783 then return True
784 else isValidLocalIdentifier them
785 case (shouldLink, isLocal node (ProcessIdentifier from)) of
786 (True, _) -> -- [Unified: first rule]
787 case mRef of
788 Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)
789 Nothing -> modify' $ linksFor them ^: Set.insert from
790 (False, True) -> -- [Unified: second rule]
791 notifyDied from them DiedUnknownId mRef
792 (False, False) -> -- [Unified: third rule]
793 -- TODO: this is the right sender according to the Unified semantics,
794 -- but perhaps having 'them' as the sender would make more sense
795 -- (see also: notifyDied)
796 ncSendToNode (processNodeId from) $ NCMsg
797 { ctrlMsgSender = NodeIdentifier (localNodeId node)
798 , ctrlMsgSignal = Died them DiedUnknownId
799 }
801-- [Unified: Table 11]
802ncEffectUnlink :: ProcessId -> Identifier -> NC ()
803ncEffectUnlink from them = do
804 node <- ask
805 when (isLocal node (ProcessIdentifier from)) $
806 case them of
807 ProcessIdentifier pid ->
808 postAsMessage from $ DidUnlinkProcess pid
809 NodeIdentifier nid ->
810 postAsMessage from $ DidUnlinkNode nid
811 SendPortIdentifier cid ->
812 postAsMessage from $ DidUnlinkPort cid
813 modify' $ linksFor them ^: Set.delete from
815-- [Unified: Table 11]
816ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
817ncEffectUnmonitor from ref = do
818 node <- ask
819 when (isLocal node (ProcessIdentifier from)) $
820 postAsMessage from $ DidUnmonitor ref
821 modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
823-- [Unified: Table 12]
824ncEffectDied :: Identifier -> DiedReason -> NC ()
825ncEffectDied ident reason = do
826 node <- ask
827 traceNotifyDied node ident reason
828 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
829 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
831-- _registry :: !(Map (String,NodeId) ProcessId)
833 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
835 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
836 forM_ uss $ \us ->
837 when (localOnly <= isLocal node (ProcessIdentifier us)) $
838 notifyDied us them reason Nothing
840 forM_ (Map.toList affectedMons) $ \(them, refs) ->
841 forM_ refs $ \(us, ref) ->
842 when (localOnly <= isLocal node (ProcessIdentifier us)) $
843 notifyDied us them reason (Just ref)
845 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
847 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
849 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
850 mapM (\(pid,nidlist) ->
851 case ident `impliesDeathOf` ProcessIdentifier pid of
852 True ->
853 do forM_ nidlist $ \(nid,_) ->
854 when (not $ isLocal node (NodeIdentifier nid))
855 (forwardNameDeath node nid)
856 return Nothing
857 False -> return $ Just (pid,nidlist) )
858 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
859 where
860 forwardNameDeath node nid = ncSendToNode nid
861 NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node)
862 , ctrlMsgSignal = Died ident reason
863 }
865-- [Unified: Table 13]
866ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
867ncEffectSpawn pid cProc ref = do
868 mProc <- unClosure cProc
869 -- If the closure does not exist, we spawn a process that throws an exception
870 -- This allows the remote node to find out what's happening
871 -- TODO:
872 let proc = case mProc of
873 Left err -> fail $ "Error: Could not resolve closure: " ++ err
874 Right p -> p
875 node <- ask
876 pid' <- liftIO $ forkProcess node proc
877 ncSendToProcess pid $ unsafeCreateUnencodedMessage $ DidSpawn ref pid'
879-- Unified semantics does not explicitly describe how to implement 'register',
880-- but mentions it's "very similar to nsend" (Table 14)
881-- We send a response indicated if the operation is invalid
882ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
883ncEffectRegister from label atnode mPid reregistration = do
884 node <- ask
885 currentVal <- gets (^. registeredHereFor label)
886 isOk <-
887 case mPid of
888 Nothing -> -- unregister request
889 return $ isJust currentVal
890 Just thepid -> -- register request
891 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
892 return $ (isNothing currentVal /= reregistration) &&
893 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
894 if isLocal node (NodeIdentifier atnode)
895 then do when isOk $
896 do modify' $ registeredHereFor label ^= mPid
897 updateRemote node currentVal mPid
898 case mPid of
899 (Just p) -> liftIO $ trace node (MxRegistered p label)
900 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
901 newVal <- gets (^. registeredHereFor label)
902 ncSendToProcess from $ unsafeCreateUnencodedMessage $
903 RegisterReply label isOk newVal
904 else let operation =
905 case reregistration of
906 True -> flip decList
907 False -> flip incList
908 in case mPid of
909 Nothing -> return ()
910 Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
911 where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
912 do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
913 forward node (processNodeId newval) (Register label atnode (Just newval) False)
914 updateRemote node Nothing (Just newval) =
915 forward node (processNodeId newval) (Register label atnode (Just newval) False)
916 updateRemote node (Just oldval) Nothing =
917 forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
918 updateRemote _ _ _ = return ()
919 maybeify f Nothing = unmaybeify $ f []
920 maybeify f (Just x) = unmaybeify $ f x
922 unmaybeify [] = Nothing
923 unmaybeify x = Just x
924 incList [] tag = [(tag,1)]
925 incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
926 incList (x:xs) tag = x : incList xs tag
927 decList [] _ = []
928 decList ((atag,1):xs) tag | atag == tag = xs
929 decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
930 decList (x:xs) tag = x:decList xs tag
931 forward node to reg =
932 when (not $ isLocal node (NodeIdentifier to)) $
933 ncSendToNode to $ NCMsg { ctrlMsgSender = ProcessIdentifier from
934 , ctrlMsgSignal = reg
935 }
938-- Unified semantics does not explicitly describe 'whereis'
939ncEffectWhereIs :: ProcessId -> String -> NC ()
940ncEffectWhereIs from label = do
941 mPid <- gets (^. registeredHereFor label)
942 ncSendToProcess from $ unsafeCreateUnencodedMessage $ WhereIsReply label mPid
944-- [Unified: Table 14]
945ncEffectNamedSend :: String -> Message -> NC ()
946ncEffectNamedSend label msg = do
947 mPid <- gets (^. registeredHereFor label)
948 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
949 forM_ mPid (`ncSendToProcess` msg)
951-- [Issue #DP-20]
952ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
953ncEffectLocalSend node to msg =
954 liftIO $ withLocalProc node to $ \p -> do
955 enqueue (processQueue p) msg
956 trace node (MxReceived to msg)
958-- [Issue #DP-20]
959ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
960ncEffectLocalPortSend from msg = do
961 node <- ask
962 let pid = sendPortProcessId from
963 cid = sendPortLocalId from
964 liftIO $ withLocalProc node pid $ \proc -> do
965 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
966 case mChan of
967 -- in the unlikely event we know nothing about this channel id,
968 -- there's little to be done - perhaps some logging/tracing though...
969 Nothing -> return ()
970 Just (TypedChannel chan') -> do
971 -- If ch is Nothing, the process has given up the read end of
972 -- the channel and we simply ignore the incoming message - this
973 ch <- deRefWeak chan'
974 forM_ ch $ \chan -> deliverChan msg chan
975 where deliverChan :: forall a . Message -> TQueue a -> IO ()
976 deliverChan (UnencodedMessage _ raw) chan' =
977 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
978 deliverChan (EncodedMessage _ _) _ =
979 -- this will not happen unless someone screws with Primitives.hs
980 error "invalid local channel delivery"
982-- [Issue #69]
983ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()
984ncEffectKill from to reason = do
985 node <- ask
986 when (isLocal node (ProcessIdentifier to)) $
987 throwException to $ ProcessKillException from reason
989-- [Issue #69]
990ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()
991ncEffectExit from to reason = do
992 node <- ask
993 when (isLocal node (ProcessIdentifier to)) $
994 throwException to $ ProcessExitException from reason
996-- [Issue #89]
997ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()
998ncEffectGetInfo from pid =
999 let lpid = processLocalId pid
1000 them = (ProcessIdentifier pid)
1001 in do
1002 node <- ask
1003 mProc <- liftIO $ withValidLocalState node
1004 $ return . (^. localProcessWithId lpid)
1005 case mProc of
1006 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
1007 from (ProcessInfoNone DiedUnknownId)
1008 Just proc -> do
1009 itsLinks <- gets (^. linksFor them)
1010 itsMons <- gets (^. monitorsFor them)
1011 registered <- gets (^. registeredHere)
1012 size <- liftIO $ queueSize $ processQueue $ proc
1014 let reg = registeredNames registered
1015 dispatch (isLocal node (ProcessIdentifier from))
1016 from
1017 ProcessInfo {
1018 infoNode = (processNodeId pid)
1019 , infoRegisteredNames = reg
1020 , infoMessageQueueLength = size
1021 , infoMonitors = Set.toList itsMons
1022 , infoLinks = Set.toList itsLinks
1024 where dispatch :: (Serializable a, Show a)
1025 => Bool
1026 -> ProcessId
1027 -> a
1028 -> NC ()
1029 dispatch True dest pInfo = postAsMessage dest $ pInfo
1030 dispatch False dest pInfo =
1031 ncSendToProcess dest $ unsafeCreateUnencodedMessage pInfo
1033 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
1034 then (k:ks)
1035 else ks) []
1037ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()
1038ncEffectGetNodeStats from _nid = do
1039 node <- ask
1040 ncState <- StateT.get
1041 nodeState <- liftIO $ withValidLocalState node return
1042 let stats =
1043 NodeStats {
1044 nodeStatsNode = localNodeId node
1045 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
1046 , nodeStatsMonitors = Map.size $ ncState ^. monitors
1047 , nodeStatsLinks = Map.size $ ncState ^. links
1048 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
1050 postAsMessage from stats
1052--------------------------------------------------------------------------------
1053-- Auxiliary --
1054--------------------------------------------------------------------------------
1056notifyDied :: ProcessId -- ^ Who to notify?
1057 -> Identifier -- ^ Who died?
1058 -> DiedReason -- ^ How did they die?
1059 -> Maybe MonitorRef -- ^ 'Nothing' for linking
1060 -> NC ()
1061notifyDied dest src reason mRef = do
1062 node <- ask
1063 case (isLocal node (ProcessIdentifier dest), mRef, src) of
1064 (True, Just ref, ProcessIdentifier pid) ->
1065 postAsMessage dest $ ProcessMonitorNotification ref pid reason
1066 (True, Just ref, NodeIdentifier nid) ->
1067 postAsMessage dest $ NodeMonitorNotification ref nid reason
1068 (True, Just ref, SendPortIdentifier cid) ->
1069 postAsMessage dest $ PortMonitorNotification ref cid reason
1070 (True, Nothing, ProcessIdentifier pid) ->
1071 throwException dest $ ProcessLinkException pid reason
1072 (True, Nothing, NodeIdentifier pid) ->
1073 throwException dest $ NodeLinkException pid reason
1074 (True, Nothing, SendPortIdentifier pid) ->
1075 throwException dest $ PortLinkException pid reason
1076 (False, _, _) ->
1077 -- The change in sender comes from [Unified: Table 10]
1078 ncSendToNode (processNodeId dest) $ NCMsg
1079 { ctrlMsgSender = NodeIdentifier (localNodeId node)
1080 , ctrlMsgSignal = Died src reason
1083-- | [Unified: Table 8]
1084destNid :: ProcessSignal -> Maybe NodeId
1085destNid (Link ident) = Just $ nodeOf ident
1086destNid (Unlink ident) = Just $ nodeOf ident
1087destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)
1088destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
1089destNid (Spawn _ _) = Nothing
1090destNid (Register _ _ _ _) = Nothing
1091destNid (WhereIs _) = Nothing
1092destNid (NamedSend _ _) = Nothing
1093destNid (UnreliableSend _ _) = Nothing
1094-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
1095-- then when a local process dies the monitoring/linking machinery will take
1096-- care of notifying remote nodes
1097destNid (Died _ _) = Nothing
1098destNid (Kill pid _) = Just $ processNodeId pid
1099destNid (Exit pid _) = Just $ processNodeId pid
1100destNid (GetInfo pid) = Just $ processNodeId pid
1101destNid (GetNodeStats nid) = Just nid
1102destNid (LocalSend pid _) = Just $ processNodeId pid
1103destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1104destNid (SigShutdown) = Nothing
1106-- | Check if a process is local to our own node
1107isLocal :: LocalNode -> Identifier -> Bool
1108isLocal nid ident = nodeOf ident == localNodeId nid
1110-- | Lookup a local closure
1111unClosure :: Typeable a => Closure a -> NC (Either String a)
1112unClosure closure = do
1113 rtable <- remoteTable <$> ask
1114 return (Static.unclosure rtable closure)
1116-- | Check if an identifier refers to a valid local object
1117isValidLocalIdentifier :: Identifier -> NC Bool
1118isValidLocalIdentifier ident = do
1119 node <- ask
1120 liftIO . withValidLocalState node $ \nSt ->
1121 case ident of
1122 NodeIdentifier nid ->
1123 return $ nid == localNodeId node
1124 ProcessIdentifier pid -> do
1125 let mProc = nSt ^. localProcessWithId (processLocalId pid)
1126 return $ isJust mProc
1127 SendPortIdentifier cid -> do
1128 let pid = sendPortProcessId cid
1129 mProc = nSt ^. localProcessWithId (processLocalId pid)
1130 case mProc of
1131 Nothing -> return False
1132 Just proc -> withMVar (processState proc) $ \pSt -> do
1133 let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)
1134 return $ isJust mCh
1136--------------------------------------------------------------------------------
1137-- Messages to local processes --
1138--------------------------------------------------------------------------------
1140postAsMessage :: Serializable a => ProcessId -> a -> NC ()
1141postAsMessage pid = postMessage pid . createUnencodedMessage
1143postMessage :: ProcessId -> Message -> NC ()
1144postMessage pid msg = do
1145 node <- ask
1146 liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
1148throwException :: Exception e => ProcessId -> e -> NC ()
1149throwException pid e = do
1150 node <- ask
1151 -- throwTo blocks until the exception is received by the target thread.
1152 -- We fork a helper thread to avoid blocking the node controller.
1153 liftIO $ withLocalProc node pid $ \p -> void $ forkIO $ throwTo (processThread p) e
1155withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1156withLocalProc node pid p =
1157 -- By [Unified: table 6, rule missing_process] messages to dead processes
1158 -- can silently be dropped
1159 let lpid = processLocalId pid in do
1160 withValidLocalState node $ \vst ->
1161 forM_ (vst ^. localProcessWithId lpid) p
1163--------------------------------------------------------------------------------
1164-- Accessors --
1165--------------------------------------------------------------------------------
1167links :: Accessor NCState (Map Identifier (Set ProcessId))
1168links = accessor _links (\ls st -> st { _links = ls })
1170monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
1171monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1173registeredHere :: Accessor NCState (Map String ProcessId)
1174registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
1176registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1177registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1179linksFor :: Identifier -> Accessor NCState (Set ProcessId)
1180linksFor ident = links >>> DAC.mapDefault Set.empty ident
1182monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
1183monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
1185registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1186registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1188registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
1189registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1191-- | @splitNotif ident@ splits a notifications map into those
1192-- notifications that should trigger when 'ident' fails and those links that
1193-- should not.
1195-- There is a hierarchy between identifiers: failure of a node implies failure
1196-- of all processes on that node, and failure of a process implies failure of
1197-- all typed channels to that process. In other words, if 'ident' refers to a
1198-- node, then the /should trigger/ set will include
1200-- * the notifications for the node specifically
1201-- * the notifications for processes on that node, and
1202-- * the notifications for typed channels to processes on that node.
1204-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
1205-- include
1207-- * the notifications for that process specifically and
1208-- * the notifications for typed channels to that process.
1210-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1211splitNotif :: Identifier
1212 -> Map Identifier a
1213 -> (Map Identifier a, Map Identifier a)
1214splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
1216--------------------------------------------------------------------------------
1217-- Auxiliary --
1218--------------------------------------------------------------------------------
1220-- | Modify and evaluate the state
1221modify' :: MonadState s m => (s -> s) -> m ()
1222modify' f = StateT.get >>= \s -> StateT.put $! f s