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