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