git.iko.soy/distributed-process
on 0579726 help sign in
distributed-process/src/Control/Distributed/Process/Node.hs
1222 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.
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.Table as Table
174 ( mxTableCoordinator
175 , startTableCoordinator
176 )
177import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace
178 ( remoteTable
179 )
180import Control.Distributed.Process.Management.Internal.Trace.Tracer
181 ( defaultTracer
182 )
183import Control.Distributed.Process.Management.Internal.Trace.Types
184 ( TraceArg(..)
185 , traceEvent
186 , traceLogFmt
187 , enableTrace
188 )
189import Control.Distributed.Process.Management.Internal.Types
190 ( MxEvent(..)
191 )
192import Control.Distributed.Process.Serializable (Serializable)
193import Control.Distributed.Process.Internal.Messaging
194 ( sendBinary
195 , closeImplicitReconnections
196 , impliesDeathOf
197 )
198import Control.Distributed.Process.Internal.Primitives
199 ( register
200 , receiveWait
201 , match
202 , sendChan
203 , unwrapMessage
204 )
205import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))
206import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
207import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)
208import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC
209 ( mapMaybe
210 , mapDefault
211 )
212import Control.Monad.Catch (try)
213import Unsafe.Coerce
214import Prelude
216--------------------------------------------------------------------------------
217-- Initialization --
218--------------------------------------------------------------------------------
220initRemoteTable :: RemoteTable
221initRemoteTable = Trace.remoteTable $ BuiltIn.remoteTable Static.initRemoteTable
223-- | Initialize a new local node.
224newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
225newLocalNode transport rtable = do
226 mEndPoint <- NT.newEndPoint transport
227 case mEndPoint of
228 Left ex -> throwIO ex
229 Right endPoint -> do
230 localNode <- createBareLocalNode endPoint rtable
231 startServiceProcesses localNode
232 return localNode
234-- | Create a new local node (without any service processes running)
235createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
236createBareLocalNode endPoint rtable = do
237 unq <- randomIO
238 state <- newMVar $ LocalNodeValid $ ValidLocalNodeState
239 { _localProcesses = Map.empty
240 , _localPidCounter = firstNonReservedProcessId
241 , _localPidUnique = unq
242 , _localConnections = Map.empty
243 }
244 ctrlChan <- newChan
245 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
246 , localEndPoint = endPoint
247 , localState = state
248 , localCtrlChan = ctrlChan
249 , localEventBus = MxEventBusInitialising
250 , remoteTable = rtable
251 }
252 tracedNode <- startMxAgent node
254 -- Once the NC terminates, the endpoint isn't much use,
255 void $ forkIO $ Exception.finally (runNodeController tracedNode)
256 (NT.closeEndPoint (localEndPoint node))
258 -- whilst a closed/failing endpoint will terminate the NC
259 void $ forkIO $ Exception.finally (handleIncomingMessages tracedNode)
260 (stopNC node)
262 return tracedNode
263 where
264 stopNC node =
265 writeChan (localCtrlChan node) NCMsg
266 { ctrlMsgSender = NodeIdentifier (localNodeId node)
267 , ctrlMsgSignal = SigShutdown
268 }
270startMxAgent :: LocalNode -> IO LocalNode
271startMxAgent node = do
272 -- see note [tracer/forkProcess races]
273 let fork = forkProcess node
274 mv <- MVar.newEmptyMVar
275 pid <- fork $ mxAgentController fork mv
276 (tracer', wqRef, mxNew') <- MVar.takeMVar mv
277 return node { localEventBus = (MxEventBus pid tracer' wqRef mxNew') }
279startDefaultTracer :: LocalNode -> IO ()
280startDefaultTracer node' = do
281 let t = localEventBus node'
282 case t of
283 MxEventBus _ (Tracer pid _) _ _ -> do
284 runProcess node' $ register "trace.controller" pid
285 pid' <- forkProcess node' defaultTracer
286 enableTrace (localEventBus node') pid'
287 runProcess node' $ register "tracer.initial" pid'
288 _ -> return ()
290-- TODO: we need a better mechanism for defining and registering services
292-- | Start and register the service processes on a node
293startServiceProcesses :: LocalNode -> IO ()
294startServiceProcesses node = do
295 -- tracing /spawns/ relies on the tracer being enabled, but we start
296 -- the default tracer first, even though it might @nsend@ to the logger
297 -- before /that/ process has started - this is a totally harmless race
298 -- however, so we deliberably ignore it
299 startDefaultTracer node
300 tableCoordinatorPid <- fork $ Table.startTableCoordinator fork
301 runProcess node $ register Table.mxTableCoordinator tableCoordinatorPid
302 logger <- forkProcess node loop
303 runProcess node $ do
304 register "logger" logger
305 -- The trace.logger is used for tracing to the console to avoid feedback
306 -- loops during tracing if the user reregisters the "logger" with a custom
307 -- process which uses 'send' or other primitives which are traced.
308 register "trace.logger" logger
309 where
310 loop = do
311 receiveWait
312 [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do
313 liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string
314 loop
315 , match $ \((time, string) :: (String, String)) -> do
316 -- this is a 'trace' message from the local node tracer
317 liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string
318 loop
319 , match $ \(ch :: SendPort ()) -> -- a shutdown request
320 sendChan ch ()
321 ]
323-- | Force-close a local node, killing all processes on that node.
324closeLocalNode :: LocalNode -> IO ()
325closeLocalNode node = do
326 -- Kill processes after refilling the mvar. Otherwise, there is potential for
327 -- deadlock as a dying process tries to get the mvar while masking exceptions
328 -- uninterruptibly.
329 join $ modifyMVar (localState node) $ \st -> case st of
330 LocalNodeValid vst -> do
331 return ( LocalNodeClosed
332 , forM_ (vst ^. localProcesses) $ \lproc ->
333 -- Semantics of 'throwTo' guarantee that target thread will get
334 -- delivered an exception. Therefore, target thread will be
335 -- killed eventually and that's as good as we can do. No need
336 -- to wait for thread to actually finish dying.
337 killThread (processThread lproc)
338 )
339 LocalNodeClosed -> return (LocalNodeClosed, return ())
340 -- This call will have the effect of shutting down the NC as well (see
341 -- 'createBareLocalNode').
342 NT.closeEndPoint (localEndPoint node)
344-- | Run a process on a local node and wait for it to finish
345runProcess :: LocalNode -> Process () -> IO ()
346runProcess node proc = do
347 done <- newEmptyMVar
348 -- TODO; When forkProcess inherits the masking state, protect the forked
349 -- thread against async exceptions that could occur before 'try' is evaluated.
350 void $ forkProcess node $ try proc >>= liftIO . putMVar done
351 takeMVar done >>= either (throwIO :: SomeException -> IO a) return
353-- | Spawn a new process on a local node
354forkProcess :: LocalNode -> Process () -> IO ProcessId
355forkProcess node proc =
356 modifyMVarMasked (localState node) startProcess
357 where
358 startProcess :: LocalNodeState -> IO (LocalNodeState, ProcessId)
359 startProcess (LocalNodeValid vst) = do
360 let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter
361 , lpidUnique = vst ^. localPidUnique
362 }
363 let pid = ProcessId { processNodeId = localNodeId node
364 , processLocalId = lpid
365 }
366 pst <- newMVar LocalProcessState { _monitorCounter = 0
367 , _spawnCounter = 0
368 , _channelCounter = 0
369 , _typedChannels = Map.empty
370 }
371 queue <- newCQueue
372 weakQueue <- mkWeakCQueue queue (return ())
373 (_, lproc) <- fixIO $ \ ~(tid, _) -> do
374 let lproc = LocalProcess { processQueue = queue
375 , processWeakQ = weakQueue
376 , processId = pid
377 , processState = pst
378 , processThread = tid
379 , processNode = node
380 }
381 tid' <- uninterruptibleMask_ $ forkIOWithUnmask $ \unmask -> do
382 reason <- Exception.catches
383 (unmask $ runLocalProcess lproc proc >> return DiedNormal)
384 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
385 mMsg <- unwrapMessage msg :: IO (Maybe String)
386 case mMsg of
387 Nothing -> return $ DiedException $ show ex
388 Just m -> return $ DiedException ("exit-from=" ++ (show from) ++ ",reason=" ++ m)))
389 , (Exception.Handler
390 (return . DiedException . (show :: SomeException -> String)))]
392 -- [Unified: Table 4, rules termination and exiting]
393 mconns <- modifyValidLocalState node (cleanupProcess pid)
394 -- XXX: Revisit after agreeing on the bigger picture for the semantics
395 -- of transport operations.
396 -- https://github.com/haskell-distributed/distributed-process/issues/204
397 forM_ mconns $ forkIO . mapM_ NT.close
399 writeChan (localCtrlChan node) NCMsg
400 { ctrlMsgSender = ProcessIdentifier pid
401 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
402 }
403 return (tid', lproc)
405 -- see note [tracer/forkProcess races]
406 trace node (MxSpawned pid)
408 if lpidCounter lpid == maxBound
409 then do
410 -- TODO: this doesn't look right at all - how do we know
411 -- that newUnique represents a process id that is available!?
412 newUnique <- randomIO
413 return ( LocalNodeValid
414 $ (localProcessWithId lpid ^= Just lproc)
415 . (localPidCounter ^= firstNonReservedProcessId)
416 . (localPidUnique ^= newUnique)
417 $ vst
418 , pid
419 )
420 else
421 return ( LocalNodeValid
422 $ (localProcessWithId lpid ^= Just lproc)
423 . (localPidCounter ^: (+ 1))
424 $ vst
425 , pid
426 )
427 startProcess LocalNodeClosed =
428 throwIO $ NodeClosedException $ localNodeId node
430 cleanupProcess :: ProcessId
431 -> ValidLocalNodeState
432 -> IO (ValidLocalNodeState, [NT.Connection])
433 cleanupProcess pid vst = do
434 let pid' = ProcessIdentifier pid
435 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)
436 return ( (localProcessWithId (processLocalId pid) ^= Nothing)
437 . (localConnections ^= unaffected)
438 $ vst
439 , map fst $ Map.elems affected
440 )
442-- note [tracer/forkProcess races]
443--
444-- Our startTracing function uses forkProcess to start the trace controller
445-- process, and of course forkProcess attempts to call traceEvent once the
446-- process has started. This is harmless, as the localEventBus is not updated
447-- until /after/ the initial forkProcess completes, so the first call to
448-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
449--
451--------------------------------------------------------------------------------
452-- Handle incoming messages --
453--------------------------------------------------------------------------------
455type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
457data IncomingTarget =
458 Uninit
459 | ToProc ProcessId (Weak (CQueue Message))
460 | ToChan TypedChannel
461 | ToNode
463data ConnectionState = ConnectionState {
464 _incoming :: !(Map NT.ConnectionId IncomingConnection)
465 , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId))
466 }
468initConnectionState :: ConnectionState
469initConnectionState = ConnectionState {
470 _incoming = Map.empty
471 , _incomingFrom = Map.empty
472 }
474incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection)
475incoming = accessor _incoming (\conns st -> st { _incoming = conns })
477incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection)
478incomingAt cid = incoming >>> DAC.mapMaybe cid
480incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId)
481incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
482 where
483 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
485handleIncomingMessages :: LocalNode -> IO ()
486handleIncomingMessages node = go initConnectionState
487 `Exception.catch` \(NodeClosedException _) -> return ()
488 where
489 go :: ConnectionState -> IO ()
490 go !st = do
491 event <- NT.receive endpoint
492 case event of
493 NT.ConnectionOpened cid rel theirAddr ->
494 if rel == NT.ReliableOrdered
495 then
496 trace node (MxConnected cid theirAddr)
497 >> go (
498 (incomingAt cid ^= Just (theirAddr, Uninit))
499 . (incomingFrom theirAddr ^: Set.insert cid)
500 $ st
501 )
502 else invalidRequest cid st
503 NT.Received cid payload ->
504 case st ^. incomingAt cid of
505 Just (_, ToProc pid weakQueue) -> do
506 mQueue <- deRefWeak weakQueue
507 forM_ mQueue $ \queue -> do
508 -- TODO: if we find that the queue is Nothing, should we remove
509 -- it from the NC state? (and same for channels, below)
510 let msg = payloadToMessage payload
511 enqueue queue msg -- 'enqueue' is strict
512 trace node (MxReceived pid msg)
513 go st
514 Just (_, ToChan (TypedChannel chan')) -> do
515 mChan <- deRefWeak chan'
516 -- If mChan is Nothing, the process has given up the read end of
517 -- the channel and we simply ignore the incoming message
518 forM_ mChan $ \chan -> atomically $
519 -- We make sure the message is fully decoded when it is enqueued
520 writeTQueue chan $! decode (BSL.fromChunks payload)
521 go st
522 Just (_, ToNode) -> do
523 let ctrlMsg = decode . BSL.fromChunks $ payload
524 writeChan ctrlChan $! ctrlMsg
525 go st
526 Just (src, Uninit) ->
527 case decode (BSL.fromChunks payload) of
528 ProcessIdentifier pid -> do
529 let lpid = processLocalId pid
530 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
531 case mProc of
532 Just proc ->
533 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
534 Nothing ->
535 invalidRequest cid st
536 SendPortIdentifier chId -> do
537 let lcid = sendPortLocalId chId
538 lpid = processLocalId (sendPortProcessId chId)
539 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
540 case mProc of
541 Just proc -> do
542 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
543 case mChannel of
544 Just channel ->
545 go (incomingAt cid ^= Just (src, ToChan channel) $ st)
546 Nothing ->
547 invalidRequest cid st
548 Nothing ->
549 invalidRequest cid st
550 NodeIdentifier nid ->
551 if nid == localNodeId node
552 then go (incomingAt cid ^= Just (src, ToNode) $ st)
553 else invalidRequest cid st
554 Nothing ->
555 invalidRequest cid st
556 NT.ConnectionClosed cid ->
557 case st ^. incomingAt cid of
558 Nothing ->
559 invalidRequest cid st
560 Just (src, _) -> do
561 trace node (MxDisconnected cid src)
562 go ( (incomingAt cid ^= Nothing)
563 . (incomingFrom src ^: Set.delete cid)
564 $ st
565 )
566 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
567 -- [Unified table 9, rule node_disconnect]
568 let nid = NodeIdentifier $ NodeId theirAddr
569 writeChan ctrlChan NCMsg
570 { ctrlMsgSender = nid
571 , ctrlMsgSignal = Died nid DiedDisconnect
572 }
573 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
574 closeImplicitReconnections node nid
575 go ( (incomingFrom theirAddr ^= Set.empty)
576 . (incoming ^: Map.filterWithKey (const . notLost))
577 $ st
578 )
579 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
580 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
581 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
582 fail $ "Cloud Haskell fatal error: transport failed: " ++ str
583 NT.EndPointClosed ->
584 return ()
585 NT.ReceivedMulticast _ _ ->
586 -- If we received a multicast message, something went horribly wrong
587 -- and we just give up
588 fail "Cloud Haskell fatal error: received unexpected multicast"
590 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
591 invalidRequest cid st = do
592 -- TODO: We should treat this as a fatal error on the part of the remote
593 -- node. That is, we should report the remote node as having died, and we
594 -- should close incoming connections (this requires a Transport layer
595 -- extension).
596 traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),
597 (Trace cid)]
598 go ( incomingAt cid ^= Nothing
599 $ st
600 )
602 endpoint = localEndPoint node
603 ctrlChan = localCtrlChan node
605--------------------------------------------------------------------------------
606-- Top-level access to the node controller --
607--------------------------------------------------------------------------------
609runNodeController :: LocalNode -> IO ()
610runNodeController node =
611 runReaderT (evalStateT (unNC nodeController) initNCState) node
612 `Exception.catch` \(NodeClosedException _) -> return ()
614--------------------------------------------------------------------------------
615-- Internal data types --
616--------------------------------------------------------------------------------
618data NCState = NCState
619 { -- Mapping from remote processes to linked local processes
620 _links :: !(Map Identifier (Set ProcessId))
621 -- Mapping from remote processes to monitoring local processes
622 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
623 -- Process registry: names and where they live, mapped to the PIDs
624 , _registeredHere :: !(Map String ProcessId)
625 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
626 }
628newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
629 deriving ( Applicative
630 , Functor
631 , Monad
632 , MonadIO
633 , MonadState NCState
634 , MonadReader LocalNode
635 )
637initNCState :: NCState
638initNCState = NCState { _links = Map.empty
639 , _monitors = Map.empty
640 , _registeredHere = Map.empty
641 , _registeredOnNodes = Map.empty
642 }
644-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This
645-- type is deliberately not exported so it cannot be caught explicitly.
646data ProcessKillException =
647 ProcessKillException !ProcessId !String
648 deriving (Typeable)
650instance Exception ProcessKillException
651instance Show ProcessKillException where
652 show (ProcessKillException pid reason) =
653 "killed-by=" ++ show pid ++ ",reason=" ++ reason
655ncSendToProcess :: ProcessId -> Message -> NC ()
656ncSendToProcess = ncSendToProcessAndTrace True
658ncSendToProcessAndTrace :: Bool -> ProcessId -> Message -> NC ()
659ncSendToProcessAndTrace shouldTrace pid msg = do
660 node <- ask
661 if processNodeId pid == localNodeId node
662 then ncEffectLocalSendAndTrace shouldTrace node pid msg
663 else liftIO $ sendBinary node
664 (NodeIdentifier $ localNodeId node)
665 (NodeIdentifier $ processNodeId pid)
666 WithImplicitReconnect
667 NCMsg { ctrlMsgSender = NodeIdentifier $ localNodeId node
668 , ctrlMsgSignal = UnreliableSend (processLocalId pid) msg
669 }
671ncSendToNode :: NodeId -> NCMsg -> NC ()
672ncSendToNode to msg = do
673 node <- ask
674 liftIO $ if to == localNodeId node
675 then writeChan (localCtrlChan node) $! msg
676 else sendBinary node
677 (NodeIdentifier $ localNodeId node)
678 (NodeIdentifier to)
679 WithImplicitReconnect
680 msg
682--------------------------------------------------------------------------------
683-- Tracing/Debugging --
684--------------------------------------------------------------------------------
686-- [Issue #104 / DP-13]
688traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
689traceNotifyDied node ident reason =
690 -- TODO: sendPortDied notifications
691 liftIO $ withLocalTracer node $ \t ->
692 case ident of
693 (NodeIdentifier nid) -> traceEvent t (MxNodeDied nid reason)
694 (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)
695 _ -> return ()
697traceEventFmtIO :: LocalNode
698 -> String
699 -> [TraceArg]
700 -> IO ()
701traceEventFmtIO node fmt args =
702 withLocalTracer node $ \t -> traceLogFmt t fmt args
704trace :: LocalNode -> MxEvent -> IO ()
705trace node ev = withLocalTracer node $ \t -> traceEvent t ev
707withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()
708withLocalTracer node act = act (localEventBus node)
710--------------------------------------------------------------------------------
711-- Core functionality --
712--------------------------------------------------------------------------------
714-- [Unified: Table 7]
715nodeController :: NC ()
716nodeController = do
717 node <- ask
718 forever' $ do
719 msg <- liftIO $ readChan (localCtrlChan node)
721 -- [Unified: Table 7, rule nc_forward]
722 case destNid (ctrlMsgSignal msg) of
723 Just nid' | nid' /= localNodeId node ->
724 ncSendToNode nid' msg
725 _ ->
726 return ()
728 case msg of
729 NCMsg (ProcessIdentifier from) (Link them) ->
730 ncEffectMonitor from them Nothing
731 NCMsg (ProcessIdentifier from) (Monitor ref) ->
732 ncEffectMonitor from (monitorRefIdent ref) (Just ref)
733 NCMsg (ProcessIdentifier from) (Unlink them) ->
734 ncEffectUnlink from them
735 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->
736 ncEffectUnmonitor from ref
737 NCMsg _from (Died ident reason) ->
738 ncEffectDied ident reason
739 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
740 ncEffectSpawn from proc ref
741 NCMsg (ProcessIdentifier from) (Register label atnode pid force) ->
742 ncEffectRegister from label atnode pid force
743 NCMsg (ProcessIdentifier from) (WhereIs label) ->
744 ncEffectWhereIs from label
745 NCMsg _ (NamedSend label msg') ->
746 ncEffectNamedSend label msg'
747 NCMsg _ (UnreliableSend lpid msg') ->
748 ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg'
749 NCMsg _ (LocalSend to msg') ->
750 ncEffectLocalSend node to msg'
751 NCMsg _ (LocalPortSend to msg') ->
752 ncEffectLocalPortSend to msg'
753 NCMsg (ProcessIdentifier from) (Kill to reason) ->
754 ncEffectKill from to reason
755 NCMsg (ProcessIdentifier from) (Exit to reason) ->
756 ncEffectExit from to reason
757 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
758 ncEffectGetInfo from pid
759 NCMsg _ SigShutdown ->
760 liftIO $ do
761 NT.closeEndPoint (localEndPoint node)
762 `Exception.finally` throwIO (NodeClosedException $ localNodeId node)
763 NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->
764 ncEffectGetNodeStats from nid
765 unexpected ->
766 error $ "nodeController: unexpected message " ++ show unexpected
768-- [Unified: Table 10]
769ncEffectMonitor :: ProcessId -- ^ Who's watching?
770 -> Identifier -- ^ Who's being watched?
771 -> Maybe MonitorRef -- ^ 'Nothing' to link
772 -> NC ()
773ncEffectMonitor from them mRef = do
774 node <- ask
775 shouldLink <-
776 if not (isLocal node them)
777 then return True
778 else isValidLocalIdentifier them
779 case (shouldLink, isLocal node (ProcessIdentifier from)) of
780 (True, _) -> -- [Unified: first rule]
781 case mRef of
782 Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)
783 Nothing -> modify' $ linksFor them ^: Set.insert from
784 (False, True) -> -- [Unified: second rule]
785 notifyDied from them DiedUnknownId mRef
786 (False, False) -> -- [Unified: third rule]
787 -- TODO: this is the right sender according to the Unified semantics,
788 -- but perhaps having 'them' as the sender would make more sense
789 -- (see also: notifyDied)
790 ncSendToNode (processNodeId from) $ NCMsg
791 { ctrlMsgSender = NodeIdentifier (localNodeId node)
792 , ctrlMsgSignal = Died them DiedUnknownId
793 }
795-- [Unified: Table 11]
796ncEffectUnlink :: ProcessId -> Identifier -> NC ()
797ncEffectUnlink from them = do
798 node <- ask
799 when (isLocal node (ProcessIdentifier from)) $
800 case them of
801 ProcessIdentifier pid ->
802 postAsMessage from $ DidUnlinkProcess pid
803 NodeIdentifier nid ->
804 postAsMessage from $ DidUnlinkNode nid
805 SendPortIdentifier cid ->
806 postAsMessage from $ DidUnlinkPort cid
807 modify' $ linksFor them ^: Set.delete from
809-- [Unified: Table 11]
810ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
811ncEffectUnmonitor from ref = do
812 node <- ask
813 when (isLocal node (ProcessIdentifier from)) $
814 postAsMessage from $ DidUnmonitor ref
815 modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
817-- [Unified: Table 12]
818ncEffectDied :: Identifier -> DiedReason -> NC ()
819ncEffectDied ident reason = do
820 node <- ask
821 traceNotifyDied node ident reason
822 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
823 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
825-- _registry :: !(Map (String,NodeId) ProcessId)
827 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
829 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
830 forM_ uss $ \us ->
831 when (localOnly <= isLocal node (ProcessIdentifier us)) $
832 notifyDied us them reason Nothing
834 forM_ (Map.toList affectedMons) $ \(them, refs) ->
835 forM_ refs $ \(us, ref) ->
836 when (localOnly <= isLocal node (ProcessIdentifier us)) $
837 notifyDied us them reason (Just ref)
839 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
841 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
843 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
844 mapM (\(pid,nidlist) ->
845 case ident `impliesDeathOf` ProcessIdentifier pid of
846 True ->
847 do forM_ nidlist $ \(nid,_) ->
848 when (not $ isLocal node (NodeIdentifier nid))
849 (forwardNameDeath node nid)
850 return Nothing
851 False -> return $ Just (pid,nidlist) )
852 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
853 where
854 forwardNameDeath node nid = ncSendToNode nid
855 NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node)
856 , ctrlMsgSignal = Died ident reason
857 }
859-- [Unified: Table 13]
860ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
861ncEffectSpawn pid cProc ref = do
862 mProc <- unClosure cProc
863 -- If the closure does not exist, we spawn a process that throws an exception
864 -- This allows the remote node to find out what's happening
865 -- TODO:
866 let proc = case mProc of
867 Left err -> fail $ "Error: Could not resolve closure: " ++ err
868 Right p -> p
869 node <- ask
870 pid' <- liftIO $ forkProcess node proc
871 ncSendToProcess pid $ unsafeCreateUnencodedMessage $ DidSpawn ref pid'
873-- Unified semantics does not explicitly describe how to implement 'register',
874-- but mentions it's "very similar to nsend" (Table 14)
875-- We send a response indicated if the operation is invalid
876ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
877ncEffectRegister from label atnode mPid reregistration = do
878 node <- ask
879 currentVal <- gets (^. registeredHereFor label)
880 isOk <-
881 case mPid of
882 Nothing -> -- unregister request
883 return $ isJust currentVal
884 Just thepid -> -- register request
885 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
886 return $ (isNothing currentVal /= reregistration) &&
887 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
888 if isLocal node (NodeIdentifier atnode)
889 then do when isOk $
890 do modify' $ registeredHereFor label ^= mPid
891 updateRemote node currentVal mPid
892 case mPid of
893 (Just p) -> liftIO $ trace node (MxRegistered p label)
894 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
895 newVal <- gets (^. registeredHereFor label)
896 ncSendToProcess from $ unsafeCreateUnencodedMessage $
897 RegisterReply label isOk newVal
898 else let operation =
899 case reregistration of
900 True -> flip decList
901 False -> flip incList
902 in case mPid of
903 Nothing -> return ()
904 Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
905 where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
906 do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
907 forward node (processNodeId newval) (Register label atnode (Just newval) False)
908 updateRemote node Nothing (Just newval) =
909 forward node (processNodeId newval) (Register label atnode (Just newval) False)
910 updateRemote node (Just oldval) Nothing =
911 forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
912 updateRemote _ _ _ = return ()
913 maybeify f Nothing = unmaybeify $ f []
914 maybeify f (Just x) = unmaybeify $ f x
916 unmaybeify [] = Nothing
917 unmaybeify x = Just x
918 incList [] tag = [(tag,1)]
919 incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
920 incList (x:xs) tag = x : incList xs tag
921 decList [] _ = []
922 decList ((atag,1):xs) tag | atag == tag = xs
923 decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
924 decList (x:xs) tag = x:decList xs tag
925 forward node to reg =
926 when (not $ isLocal node (NodeIdentifier to)) $
927 ncSendToNode to $ NCMsg { ctrlMsgSender = ProcessIdentifier from
928 , ctrlMsgSignal = reg
929 }
932-- Unified semantics does not explicitly describe 'whereis'
933ncEffectWhereIs :: ProcessId -> String -> NC ()
934ncEffectWhereIs from label = do
935 mPid <- gets (^. registeredHereFor label)
936 ncSendToProcess from $ unsafeCreateUnencodedMessage $ WhereIsReply label mPid
938-- [Unified: Table 14]
939ncEffectNamedSend :: String -> Message -> NC ()
940ncEffectNamedSend label msg = do
941 mPid <- gets (^. registeredHereFor label)
942 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
943 forM_ mPid $ \to ->
944 -- If this is a trace message we don't trace it to avoid entering a loop
945 -- where trace messages produce more trace messages.
946 ncSendToProcessAndTrace (label /= "trace.logger") to msg
948-- [Issue #DP-20]
949ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
950ncEffectLocalSend = ncEffectLocalSendAndTrace True
952ncEffectLocalSendAndTrace :: Bool -> LocalNode -> ProcessId -> Message -> NC ()
953ncEffectLocalSendAndTrace shouldTrace node to msg =
954 liftIO $ withLocalProc node to $ \p -> do
955 enqueue (processQueue p) msg
956 when shouldTrace $ 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