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