git.iko.soy/distributed-process
on 5eaedc2 help sign in
distributed-process/src/Control/Distributed/Process/Node.hs
1245 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 #-}
8{-# LANGUAGE MagicHash #-}
10-- | Local nodes
11--
12module Control.Distributed.Process.Node
13 ( LocalNode
14 , newLocalNode
15 , closeLocalNode
16 , forkProcess
17 , runProcess
18 , initRemoteTable
19 , localNodeId
20 ) where
22-- TODO: Calls to 'sendBinary' and co (by the NC) may stall the node controller.
24import System.IO (fixIO, hPutStrLn, stderr)
25import System.Mem.Weak (Weak, deRefWeak)
26import qualified Data.ByteString.Lazy as BSL (fromChunks)
27import Data.Binary (decode)
28import Data.Map (Map)
29import qualified Data.Map as Map
30 ( empty
31 , toList
32 , fromList
33 , filter
34 , partitionWithKey
35 , elems
36 , size
37 , filterWithKey
38 , foldlWithKey
39 )
40import Data.Set (Set)
41import qualified Data.Set as Set
42 ( empty
43 , insert
44 , delete
45 , member
46 , toList
47 )
48import Data.Foldable (forM_)
49import Data.Maybe (isJust, fromJust, isNothing, catMaybes)
50import Data.Typeable (Typeable)
51import Control.Category ((>>>))
52import Control.Applicative
53import Control.Monad (void, when, join)
54import Control.Monad.IO.Class (MonadIO, liftIO)
55import Control.Monad.State.Strict (MonadState, StateT, evalStateT, gets)
56import qualified Control.Monad.State.Strict as StateT (get, put)
57import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)
58import Control.Exception
59 ( throwIO
60 , SomeException
61 , Exception
62 , throwTo
63 , uninterruptibleMask_
64 , getMaskingState
65 , MaskingState(..)
66 )
67import qualified Control.Exception as Exception
68 ( Handler(..)
69 , catch
70 , catches
71 , finally
72 )
73import Control.Concurrent (forkIO, forkIOWithUnmask, killThread)
74import Control.Distributed.Process.Internal.StrictMVar
75 ( newMVar
76 , withMVar
77 , modifyMVarMasked
78 , modifyMVar
79 , newEmptyMVar
80 , putMVar
81 , takeMVar
82 )
83import Control.Concurrent.Chan (newChan, writeChan, readChan)
84import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)
85import Control.Concurrent.STM
86 ( atomically
87 )
88import Control.Distributed.Process.Internal.CQueue
89 ( CQueue
90 , enqueue
91 , newCQueue
92 , mkWeakCQueue
93 , queueSize
94 )
95import qualified Network.Transport as NT
96 ( Transport
97 , EndPoint
98 , newEndPoint
99 , receive
100 , Event(..)
101 , EventErrorCode(..)
102 , TransportError(..)
103 , address
104 , closeEndPoint
105 , Connection
106 , ConnectionId
107 , close
108 , EndPointAddress
109 , Reliability(ReliableOrdered)
110 )
111import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
112import System.Random (randomIO)
113import Control.Distributed.Static (RemoteTable, Closure)
114import qualified Control.Distributed.Static as Static
115 ( unclosure
116 , initRemoteTable
117 )
118import Control.Distributed.Process.Internal.Types
119 ( NodeId(..)
120 , LocalProcessId(..)
121 , ProcessId(..)
122 , LocalNode(..)
123 , MxEventBus(..)
124 , LocalNodeState(..)
125 , ValidLocalNodeState(..)
126 , withValidLocalState
127 , modifyValidLocalState
128 , LocalProcess(..)
129 , LocalProcessState(..)
130 , Process(..)
131 , DiedReason(..)
132 , NCMsg(..)
133 , ProcessSignal(..)
134 , localPidCounter
135 , localPidUnique
136 , localProcessWithId
137 , localProcesses
138 , localConnections
139 , forever'
140 , MonitorRef(..)
141 , NodeClosedException(..)
142 , ProcessMonitorNotification(..)
143 , NodeMonitorNotification(..)
144 , PortMonitorNotification(..)
145 , ProcessExitException(..)
146 , ProcessLinkException(..)
147 , NodeLinkException(..)
148 , PortLinkException(..)
149 , DidUnmonitor(..)
150 , DidUnlinkProcess(..)
151 , DidUnlinkNode(..)
152 , DidUnlinkPort(..)
153 , SpawnRef
154 , DidSpawn(..)
155 , Message(..)
156 , TypedChannel(..)
157 , Identifier(..)
158 , nodeOf
159 , ProcessInfo(..)
160 , ProcessInfoNone(..)
161 , NodeStats(..)
162 , SendPortId(..)
163 , typedChannelWithId
164 , RegisterReply(..)
165 , WhereIsReply(..)
166 , payloadToMessage
167 , createUnencodedMessage
168 , unsafeCreateUnencodedMessage
169 , runLocalProcess
170 , firstNonReservedProcessId
171 , ImplicitReconnect(WithImplicitReconnect)
172 )
173import Control.Distributed.Process.Management.Internal.Agent
174 ( mxAgentController
175 )
176import qualified Control.Distributed.Process.Management.Internal.Table as Table
177 ( mxTableCoordinator
178 , startTableCoordinator
179 )
180import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace
181 ( remoteTable
182 )
183import Control.Distributed.Process.Management.Internal.Trace.Tracer
184 ( defaultTracer
185 )
186import Control.Distributed.Process.Management.Internal.Trace.Types
187 ( TraceArg(..)
188 , traceEvent
189 , traceLogFmt
190 , enableTrace
191 )
192import Control.Distributed.Process.Management.Internal.Types
193 ( MxEvent(..)
194 )
195import Control.Distributed.Process.Serializable (Serializable)
196import Control.Distributed.Process.Internal.Messaging
197 ( sendBinary
198 , closeImplicitReconnections
199 , impliesDeathOf
200 )
201import Control.Distributed.Process.Internal.Primitives
202 ( register
203 , receiveWait
204 , match
205 , sendChan
206 , unwrapMessage
207 )
208import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))
209import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
210import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)
211import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC
212 ( mapMaybe
213 , mapDefault
214 )
215import Control.Monad.Catch (try)
216import GHC.IO (IO(..), unsafeUnmask)
217import GHC.Base ( maskAsyncExceptions# )
219import Unsafe.Coerce
220import Prelude
222-- Remove these definitions when the fix for
223-- https://ghc.haskell.org/trac/ghc/ticket/10149
224-- is included in all supported compilers:
225block :: IO a -> IO a
226block (IO io) = IO $ maskAsyncExceptions# io
227unblock :: IO a -> IO a
228unblock = unsafeUnmask
230--------------------------------------------------------------------------------
231-- Initialization --
232--------------------------------------------------------------------------------
234initRemoteTable :: RemoteTable
235initRemoteTable = Trace.remoteTable $ BuiltIn.remoteTable Static.initRemoteTable
237-- | Initialize a new local node.
238newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
239newLocalNode transport rtable = do
240 mEndPoint <- NT.newEndPoint transport
241 case mEndPoint of
242 Left ex -> throwIO ex
243 Right endPoint -> do
244 localNode <- createBareLocalNode endPoint rtable
245 startServiceProcesses localNode
246 return localNode
248-- | Create a new local node (without any service processes running)
249createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
250createBareLocalNode endPoint rtable = do
251 unq <- randomIO
252 state <- newMVar $ LocalNodeValid $ ValidLocalNodeState
253 { _localProcesses = Map.empty
254 , _localPidCounter = firstNonReservedProcessId
255 , _localPidUnique = unq
256 , _localConnections = Map.empty
257 }
258 ctrlChan <- newChan
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 }
266 tracedNode <- startMxAgent node
268 -- Once the NC terminates, the endpoint isn't much use,
269 void $ forkIO $ Exception.finally (runNodeController tracedNode)
270 (NT.closeEndPoint (localEndPoint node))
272 -- whilst a closed/failing endpoint will terminate the NC
273 void $ forkIO $ Exception.finally (handleIncomingMessages tracedNode)
274 (stopNC node)
276 return tracedNode
277 where
278 stopNC node =
279 writeChan (localCtrlChan node) NCMsg
280 { ctrlMsgSender = NodeIdentifier (localNodeId node)
281 , ctrlMsgSignal = SigShutdown
282 }
284startMxAgent :: LocalNode -> IO LocalNode
285startMxAgent node = do
286 -- see note [tracer/forkProcess races]
287 let fork = forkProcess node
288 mv <- MVar.newEmptyMVar
289 pid <- fork $ mxAgentController fork mv
290 (tracer', wqRef, mxNew') <- MVar.takeMVar mv
291 return node { localEventBus = (MxEventBus pid tracer' wqRef mxNew') }
293startDefaultTracer :: LocalNode -> IO ()
294startDefaultTracer node' = do
295 let t = localEventBus node'
296 case t of
297 MxEventBus _ (Tracer pid _) _ _ -> do
298 runProcess node' $ register "trace.controller" pid
299 pid' <- forkProcess node' defaultTracer
300 enableTrace (localEventBus node') pid'
301 runProcess node' $ register "tracer.initial" pid'
302 _ -> return ()
304-- TODO: we need a better mechanism for defining and registering services
306-- | Start and register the service processes on a node
307startServiceProcesses :: LocalNode -> IO ()
308startServiceProcesses node = do
309 -- tracing /spawns/ relies on the tracer being enabled, but we start
310 -- the default tracer first, even though it might @nsend@ to the logger
311 -- before /that/ process has started - this is a totally harmless race
312 -- however, so we deliberably ignore it
313 startDefaultTracer node
314 tableCoordinatorPid <- fork $ Table.startTableCoordinator fork
315 runProcess node $ register Table.mxTableCoordinator tableCoordinatorPid
316 logger <- forkProcess node loop
317 runProcess node $ do
318 register "logger" logger
319 -- The trace.logger is used for tracing to the console to avoid feedback
320 -- loops during tracing if the user reregisters the "logger" with a custom
321 -- process which uses 'send' or other primitives which are traced.
322 register "trace.logger" logger
323 where
324 loop = do
325 receiveWait
326 [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do
327 liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string
328 loop
329 , match $ \((time, string) :: (String, String)) -> do
330 -- this is a 'trace' message from the local node tracer
331 liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string
332 loop
333 , match $ \(ch :: SendPort ()) -> -- a shutdown request
334 sendChan ch ()
335 ]
337-- | Force-close a local node, killing all processes on that node.
338closeLocalNode :: LocalNode -> IO ()
339closeLocalNode node = do
340 -- Kill processes after refilling the mvar. Otherwise, there is potential for
341 -- deadlock as a dying process tries to get the mvar while masking exceptions
342 -- uninterruptibly.
343 join $ modifyMVar (localState node) $ \st -> case st of
344 LocalNodeValid vst -> do
345 return ( LocalNodeClosed
346 , forM_ (vst ^. localProcesses) $ \lproc ->
347 -- Semantics of 'throwTo' guarantee that target thread will get
348 -- delivered an exception. Therefore, target thread will be
349 -- killed eventually and that's as good as we can do. No need
350 -- to wait for thread to actually finish dying.
351 killThread (processThread lproc)
352 )
353 LocalNodeClosed -> return (LocalNodeClosed, return ())
354 -- This call will have the effect of shutting down the NC as well (see
355 -- 'createBareLocalNode').
356 NT.closeEndPoint (localEndPoint node)
358-- | Run a process on a local node and wait for it to finish
359runProcess :: LocalNode -> Process () -> IO ()
360runProcess node proc = do
361 done <- newEmptyMVar
362 -- TODO; When forkProcess inherits the masking state, protect the forked
363 -- thread against async exceptions that could occur before 'try' is evaluated.
364 void $ forkProcess node $ try proc >>= liftIO . putMVar done
365 takeMVar done >>= either (throwIO :: SomeException -> IO a) return
367-- | Spawn a new process on a local node
368forkProcess :: LocalNode -> Process () -> IO ProcessId
369forkProcess node proc = do
370 ms <- getMaskingState
371 modifyMVarMasked (localState node) (startProcess ms)
372 where
373 startProcess :: MaskingState
374 -> LocalNodeState
375 -> IO (LocalNodeState, ProcessId)
376 startProcess ms (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 -- Rewrite this code when this is fixed:
399 -- https://ghc.haskell.org/trac/ghc/ticket/10149
400 let unmask = case ms of
401 Unmasked -> unblock
402 MaskedInterruptible -> block
403 MaskedUninterruptible -> id
404 tid' <- uninterruptibleMask_ $ forkIO $ do
405 reason <- Exception.catches
406 (unmask $ runLocalProcess lproc proc >> return DiedNormal)
407 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
408 mMsg <- unwrapMessage msg :: IO (Maybe String)
409 case mMsg of
410 Nothing -> return $ DiedException $ show ex
411 Just m -> return $ DiedException ("exit-from=" ++ (show from) ++ ",reason=" ++ m)))
412 , (Exception.Handler
413 (return . DiedException . (show :: SomeException -> String)))]
415 -- [Unified: Table 4, rules termination and exiting]
416 mconns <- modifyValidLocalState node (cleanupProcess pid)
417 -- XXX: Revisit after agreeing on the bigger picture for the semantics
418 -- of transport operations.
419 -- https://github.com/haskell-distributed/distributed-process/issues/204
420 forM_ mconns $ forkIO . mapM_ NT.close
422 writeChan (localCtrlChan node) NCMsg
423 { ctrlMsgSender = ProcessIdentifier pid
424 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
425 }
426 return (tid', lproc)
428 -- see note [tracer/forkProcess races]
429 trace node (MxSpawned pid)
431 if lpidCounter lpid == maxBound
432 then do
433 -- TODO: this doesn't look right at all - how do we know
434 -- that newUnique represents a process id that is available!?
435 newUnique <- randomIO
436 return ( LocalNodeValid
437 $ (localProcessWithId lpid ^= Just lproc)
438 . (localPidCounter ^= firstNonReservedProcessId)
439 . (localPidUnique ^= newUnique)
440 $ vst
441 , pid
442 )
443 else
444 return ( LocalNodeValid
445 $ (localProcessWithId lpid ^= Just lproc)
446 . (localPidCounter ^: (+ 1))
447 $ vst
448 , pid
449 )
450 startProcess _ LocalNodeClosed =
451 throwIO $ NodeClosedException $ localNodeId node
453 cleanupProcess :: ProcessId
454 -> ValidLocalNodeState
455 -> IO (ValidLocalNodeState, [NT.Connection])
456 cleanupProcess pid vst = do
457 let pid' = ProcessIdentifier pid
458 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)
459 return ( (localProcessWithId (processLocalId pid) ^= Nothing)
460 . (localConnections ^= unaffected)
461 $ vst
462 , map fst $ Map.elems affected
463 )
465-- note [tracer/forkProcess races]
466--
467-- Our startTracing function uses forkProcess to start the trace controller
468-- process, and of course forkProcess attempts to call traceEvent once the
469-- process has started. This is harmless, as the localEventBus is not updated
470-- until /after/ the initial forkProcess completes, so the first call to
471-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
472--
474--------------------------------------------------------------------------------
475-- Handle incoming messages --
476--------------------------------------------------------------------------------
478type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
480data IncomingTarget =
481 Uninit
482 | ToProc ProcessId (Weak (CQueue Message))
483 | ToChan TypedChannel
484 | ToNode
486data ConnectionState = ConnectionState {
487 _incoming :: !(Map NT.ConnectionId IncomingConnection)
488 , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId))
489 }
491initConnectionState :: ConnectionState
492initConnectionState = ConnectionState {
493 _incoming = Map.empty
494 , _incomingFrom = Map.empty
495 }
497incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection)
498incoming = accessor _incoming (\conns st -> st { _incoming = conns })
500incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection)
501incomingAt cid = incoming >>> DAC.mapMaybe cid
503incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId)
504incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
505 where
506 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
508handleIncomingMessages :: LocalNode -> IO ()
509handleIncomingMessages node = go initConnectionState
510 `Exception.catch` \(NodeClosedException _) -> return ()
511 where
512 go :: ConnectionState -> IO ()
513 go !st = do
514 event <- NT.receive endpoint
515 case event of
516 NT.ConnectionOpened cid rel theirAddr ->
517 if rel == NT.ReliableOrdered
518 then
519 trace node (MxConnected cid theirAddr)
520 >> go (
521 (incomingAt cid ^= Just (theirAddr, Uninit))
522 . (incomingFrom theirAddr ^: Set.insert cid)
523 $ st
524 )
525 else invalidRequest cid st
526 NT.Received cid payload ->
527 case st ^. incomingAt cid of
528 Just (_, ToProc pid weakQueue) -> do
529 mQueue <- deRefWeak weakQueue
530 forM_ mQueue $ \queue -> do
531 -- TODO: if we find that the queue is Nothing, should we remove
532 -- it from the NC state? (and same for channels, below)
533 let msg = payloadToMessage payload
534 enqueue queue msg -- 'enqueue' is strict
535 trace node (MxReceived pid msg)
536 go st
537 Just (_, ToChan (TypedChannel chan')) -> do
538 mChan <- deRefWeak chan'
539 -- If mChan is Nothing, the process has given up the read end of
540 -- the channel and we simply ignore the incoming message
541 forM_ mChan $ \chan -> atomically $
542 -- We make sure the message is fully decoded when it is enqueued
543 writeTQueue chan $! decode (BSL.fromChunks payload)
544 go st
545 Just (_, ToNode) -> do
546 let ctrlMsg = decode . BSL.fromChunks $ payload
547 writeChan ctrlChan $! ctrlMsg
548 go st
549 Just (src, Uninit) ->
550 case decode (BSL.fromChunks payload) of
551 ProcessIdentifier pid -> do
552 let lpid = processLocalId pid
553 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
554 case mProc of
555 Just proc ->
556 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
557 Nothing ->
558 invalidRequest cid st
559 SendPortIdentifier chId -> do
560 let lcid = sendPortLocalId chId
561 lpid = processLocalId (sendPortProcessId chId)
562 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
563 case mProc of
564 Just proc -> do
565 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
566 case mChannel of
567 Just channel ->
568 go (incomingAt cid ^= Just (src, ToChan channel) $ st)
569 Nothing ->
570 invalidRequest cid st
571 Nothing ->
572 invalidRequest cid st
573 NodeIdentifier nid ->
574 if nid == localNodeId node
575 then go (incomingAt cid ^= Just (src, ToNode) $ st)
576 else invalidRequest cid st
577 Nothing ->
578 invalidRequest cid st
579 NT.ConnectionClosed cid ->
580 case st ^. incomingAt cid of
581 Nothing ->
582 invalidRequest cid st
583 Just (src, _) -> do
584 trace node (MxDisconnected cid src)
585 go ( (incomingAt cid ^= Nothing)
586 . (incomingFrom src ^: Set.delete cid)
587 $ st
588 )
589 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
590 -- [Unified table 9, rule node_disconnect]
591 let nid = NodeIdentifier $ NodeId theirAddr
592 writeChan ctrlChan NCMsg
593 { ctrlMsgSender = nid
594 , ctrlMsgSignal = Died nid DiedDisconnect
595 }
596 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
597 closeImplicitReconnections node nid
598 go ( (incomingFrom theirAddr ^= Set.empty)
599 . (incoming ^: Map.filterWithKey (const . notLost))
600 $ st
601 )
602 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
603 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
604 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
605 fail $ "Cloud Haskell fatal error: transport failed: " ++ str
606 NT.EndPointClosed ->
607 return ()
608 NT.ReceivedMulticast _ _ ->
609 -- If we received a multicast message, something went horribly wrong
610 -- and we just give up
611 fail "Cloud Haskell fatal error: received unexpected multicast"
613 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
614 invalidRequest cid st = do
615 -- TODO: We should treat this as a fatal error on the part of the remote
616 -- node. That is, we should report the remote node as having died, and we
617 -- should close incoming connections (this requires a Transport layer
618 -- extension).
619 traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),
620 (Trace cid)]
621 go ( incomingAt cid ^= Nothing
622 $ st
623 )
625 endpoint = localEndPoint node
626 ctrlChan = localCtrlChan node
628--------------------------------------------------------------------------------
629-- Top-level access to the node controller --
630--------------------------------------------------------------------------------
632runNodeController :: LocalNode -> IO ()
633runNodeController node =
634 runReaderT (evalStateT (unNC nodeController) initNCState) node
635 `Exception.catch` \(NodeClosedException _) -> return ()
637--------------------------------------------------------------------------------
638-- Internal data types --
639--------------------------------------------------------------------------------
641data NCState = NCState
642 { -- Mapping from remote processes to linked local processes
643 _links :: !(Map Identifier (Set ProcessId))
644 -- Mapping from remote processes to monitoring local processes
645 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
646 -- Process registry: names and where they live, mapped to the PIDs
647 , _registeredHere :: !(Map String ProcessId)
648 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
649 }
651newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
652 deriving ( Applicative
653 , Functor
654 , Monad
655 , MonadIO
656 , MonadState NCState
657 , MonadReader LocalNode
658 )
660initNCState :: NCState
661initNCState = NCState { _links = Map.empty
662 , _monitors = Map.empty
663 , _registeredHere = Map.empty
664 , _registeredOnNodes = Map.empty
665 }
667-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This
668-- type is deliberately not exported so it cannot be caught explicitly.
669data ProcessKillException =
670 ProcessKillException !ProcessId !String
671 deriving (Typeable)
673instance Exception ProcessKillException
674instance Show ProcessKillException where
675 show (ProcessKillException pid reason) =
676 "killed-by=" ++ show pid ++ ",reason=" ++ reason
678ncSendToProcess :: ProcessId -> Message -> NC ()
679ncSendToProcess = ncSendToProcessAndTrace True
681ncSendToProcessAndTrace :: Bool -> ProcessId -> Message -> NC ()
682ncSendToProcessAndTrace shouldTrace pid msg = do
683 node <- ask
684 if processNodeId pid == localNodeId node
685 then ncEffectLocalSendAndTrace shouldTrace node pid msg
686 else liftIO $ 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 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 $ \to ->
967 -- If this is a trace message we don't trace it to avoid entering a loop
968 -- where trace messages produce more trace messages.
969 ncSendToProcessAndTrace (label /= "trace.logger") to msg
971-- [Issue #DP-20]
972ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
973ncEffectLocalSend = ncEffectLocalSendAndTrace True
975ncEffectLocalSendAndTrace :: Bool -> LocalNode -> ProcessId -> Message -> NC ()
976ncEffectLocalSendAndTrace shouldTrace node to msg =
977 liftIO $ withLocalProc node to $ \p -> do
978 enqueue (processQueue p) msg
979 when shouldTrace $ trace node (MxReceived to msg)
981-- [Issue #DP-20]
982ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
983ncEffectLocalPortSend from msg = do
984 node <- ask
985 let pid = sendPortProcessId from
986 cid = sendPortLocalId from
987 liftIO $ withLocalProc node pid $ \proc -> do
988 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
989 case mChan of
990 -- in the unlikely event we know nothing about this channel id,
991 -- there's little to be done - perhaps some logging/tracing though...
992 Nothing -> return ()
993 Just (TypedChannel chan') -> do
994 -- If ch is Nothing, the process has given up the read end of
995 -- the channel and we simply ignore the incoming message - this
996 ch <- deRefWeak chan'
997 forM_ ch $ \chan -> deliverChan msg chan
998 where deliverChan :: forall a . Message -> TQueue a -> IO ()
999 deliverChan (UnencodedMessage _ raw) chan' =
1000 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1001 deliverChan (EncodedMessage _ _) _ =
1002 -- this will not happen unless someone screws with Primitives.hs
1003 error "invalid local channel delivery"
1005-- [Issue #69]
1006ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()
1007ncEffectKill from to reason = do
1008 node <- ask
1009 when (isLocal node (ProcessIdentifier to)) $
1010 throwException to $ ProcessKillException from reason
1012-- [Issue #69]
1013ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()
1014ncEffectExit from to reason = do
1015 node <- ask
1016 when (isLocal node (ProcessIdentifier to)) $
1017 throwException to $ ProcessExitException from reason
1019-- [Issue #89]
1020ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()
1021ncEffectGetInfo from pid =
1022 let lpid = processLocalId pid
1023 them = (ProcessIdentifier pid)
1024 in do
1025 node <- ask
1026 mProc <- liftIO $ withValidLocalState node
1027 $ return . (^. localProcessWithId lpid)
1028 case mProc of
1029 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
1030 from (ProcessInfoNone DiedUnknownId)
1031 Just proc -> do
1032 itsLinks <- gets (^. linksFor them)
1033 itsMons <- gets (^. monitorsFor them)
1034 registered <- gets (^. registeredHere)
1035 size <- liftIO $ queueSize $ processQueue $ proc
1037 let reg = registeredNames registered
1038 dispatch (isLocal node (ProcessIdentifier from))
1039 from
1040 ProcessInfo {
1041 infoNode = (processNodeId pid)
1042 , infoRegisteredNames = reg
1043 , infoMessageQueueLength = size
1044 , infoMonitors = Set.toList itsMons
1045 , infoLinks = Set.toList itsLinks
1047 where dispatch :: (Serializable a, Show a)
1048 => Bool
1049 -> ProcessId
1050 -> a
1051 -> NC ()
1052 dispatch True dest pInfo = postAsMessage dest $ pInfo
1053 dispatch False dest pInfo =
1054 ncSendToProcess dest $ unsafeCreateUnencodedMessage pInfo
1056 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
1057 then (k:ks)
1058 else ks) []
1060ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()
1061ncEffectGetNodeStats from _nid = do
1062 node <- ask
1063 ncState <- StateT.get
1064 nodeState <- liftIO $ withValidLocalState node return
1065 let stats =
1066 NodeStats {
1067 nodeStatsNode = localNodeId node
1068 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
1069 , nodeStatsMonitors = Map.size $ ncState ^. monitors
1070 , nodeStatsLinks = Map.size $ ncState ^. links
1071 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
1073 postAsMessage from stats
1075--------------------------------------------------------------------------------
1076-- Auxiliary --
1077--------------------------------------------------------------------------------
1079notifyDied :: ProcessId -- ^ Who to notify?
1080 -> Identifier -- ^ Who died?
1081 -> DiedReason -- ^ How did they die?
1082 -> Maybe MonitorRef -- ^ 'Nothing' for linking
1083 -> NC ()
1084notifyDied dest src reason mRef = do
1085 node <- ask
1086 case (isLocal node (ProcessIdentifier dest), mRef, src) of
1087 (True, Just ref, ProcessIdentifier pid) ->
1088 postAsMessage dest $ ProcessMonitorNotification ref pid reason
1089 (True, Just ref, NodeIdentifier nid) ->
1090 postAsMessage dest $ NodeMonitorNotification ref nid reason
1091 (True, Just ref, SendPortIdentifier cid) ->
1092 postAsMessage dest $ PortMonitorNotification ref cid reason
1093 (True, Nothing, ProcessIdentifier pid) ->
1094 throwException dest $ ProcessLinkException pid reason
1095 (True, Nothing, NodeIdentifier pid) ->
1096 throwException dest $ NodeLinkException pid reason
1097 (True, Nothing, SendPortIdentifier pid) ->
1098 throwException dest $ PortLinkException pid reason
1099 (False, _, _) ->
1100 -- The change in sender comes from [Unified: Table 10]
1101 ncSendToNode (processNodeId dest) $ NCMsg
1102 { ctrlMsgSender = NodeIdentifier (localNodeId node)
1103 , ctrlMsgSignal = Died src reason
1106-- | [Unified: Table 8]
1107destNid :: ProcessSignal -> Maybe NodeId
1108destNid (Link ident) = Just $ nodeOf ident
1109destNid (Unlink ident) = Just $ nodeOf ident
1110destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)
1111destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
1112destNid (Spawn _ _) = Nothing
1113destNid (Register _ _ _ _) = Nothing
1114destNid (WhereIs _) = Nothing
1115destNid (NamedSend _ _) = Nothing
1116destNid (UnreliableSend _ _) = Nothing
1117-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
1118-- then when a local process dies the monitoring/linking machinery will take
1119-- care of notifying remote nodes
1120destNid (Died _ _) = Nothing
1121destNid (Kill pid _) = Just $ processNodeId pid
1122destNid (Exit pid _) = Just $ processNodeId pid
1123destNid (GetInfo pid) = Just $ processNodeId pid
1124destNid (GetNodeStats nid) = Just nid
1125destNid (LocalSend pid _) = Just $ processNodeId pid
1126destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1127destNid (SigShutdown) = Nothing
1129-- | Check if a process is local to our own node
1130isLocal :: LocalNode -> Identifier -> Bool
1131isLocal nid ident = nodeOf ident == localNodeId nid
1133-- | Lookup a local closure
1134unClosure :: Typeable a => Closure a -> NC (Either String a)
1135unClosure closure = do
1136 rtable <- remoteTable <$> ask
1137 return (Static.unclosure rtable closure)
1139-- | Check if an identifier refers to a valid local object
1140isValidLocalIdentifier :: Identifier -> NC Bool
1141isValidLocalIdentifier ident = do
1142 node <- ask
1143 liftIO . withValidLocalState node $ \nSt ->
1144 case ident of
1145 NodeIdentifier nid ->
1146 return $ nid == localNodeId node
1147 ProcessIdentifier pid -> do
1148 let mProc = nSt ^. localProcessWithId (processLocalId pid)
1149 return $ isJust mProc
1150 SendPortIdentifier cid -> do
1151 let pid = sendPortProcessId cid
1152 mProc = nSt ^. localProcessWithId (processLocalId pid)
1153 case mProc of
1154 Nothing -> return False
1155 Just proc -> withMVar (processState proc) $ \pSt -> do
1156 let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)
1157 return $ isJust mCh
1159--------------------------------------------------------------------------------
1160-- Messages to local processes --
1161--------------------------------------------------------------------------------
1163postAsMessage :: Serializable a => ProcessId -> a -> NC ()
1164postAsMessage pid = postMessage pid . createUnencodedMessage
1166postMessage :: ProcessId -> Message -> NC ()
1167postMessage pid msg = do
1168 node <- ask
1169 liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
1171throwException :: Exception e => ProcessId -> e -> NC ()
1172throwException pid e = do
1173 node <- ask
1174 -- throwTo blocks until the exception is received by the target thread.
1175 -- We fork a helper thread to avoid blocking the node controller.
1176 liftIO $ withLocalProc node pid $ \p -> void $ forkIO $ throwTo (processThread p) e
1178withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1179withLocalProc node pid p =
1180 -- By [Unified: table 6, rule missing_process] messages to dead processes
1181 -- can silently be dropped
1182 let lpid = processLocalId pid in do
1183 withValidLocalState node $ \vst ->
1184 forM_ (vst ^. localProcessWithId lpid) p
1186--------------------------------------------------------------------------------
1187-- Accessors --
1188--------------------------------------------------------------------------------
1190links :: Accessor NCState (Map Identifier (Set ProcessId))
1191links = accessor _links (\ls st -> st { _links = ls })
1193monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
1194monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1196registeredHere :: Accessor NCState (Map String ProcessId)
1197registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
1199registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1200registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1202linksFor :: Identifier -> Accessor NCState (Set ProcessId)
1203linksFor ident = links >>> DAC.mapDefault Set.empty ident
1205monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
1206monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
1208registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1209registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1211registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
1212registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1214-- | @splitNotif ident@ splits a notifications map into those
1215-- notifications that should trigger when 'ident' fails and those links that
1216-- should not.
1218-- There is a hierarchy between identifiers: failure of a node implies failure
1219-- of all processes on that node, and failure of a process implies failure of
1220-- all typed channels to that process. In other words, if 'ident' refers to a
1221-- node, then the /should trigger/ set will include
1223-- * the notifications for the node specifically
1224-- * the notifications for processes on that node, and
1225-- * the notifications for typed channels to processes on that node.
1227-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
1228-- include
1230-- * the notifications for that process specifically and
1231-- * the notifications for typed channels to that process.
1233-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1234splitNotif :: Identifier
1235 -> Map Identifier a
1236 -> (Map Identifier a, Map Identifier a)
1237splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
1239--------------------------------------------------------------------------------
1240-- Auxiliary --
1241--------------------------------------------------------------------------------
1243-- | Modify and evaluate the state
1244modify' :: MonadState s m => (s -> s) -> m ()
1245modify' f = StateT.get >>= \s -> StateT.put $! f s