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