git.iko.soy/distributed-process
on 66400bb help sign in
distributed-process/src/Control/Distributed/Process/Node.hs
1248 lines 50 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 fork = forkProcess node
326 loop = do
327 receiveWait
328 [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do
329 liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string
330 loop
331 , match $ \((time, string) :: (String, String)) -> do
332 -- this is a 'trace' message from the local node tracer
333 liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string
334 loop
335 , match $ \(ch :: SendPort ()) -> -- a shutdown request
336 sendChan ch ()
337 ]
339-- | Force-close a local node, killing all processes on that node.
340closeLocalNode :: LocalNode -> IO ()
341closeLocalNode node = do
342 -- Kill processes after refilling the mvar. Otherwise, there is potential for
343 -- deadlock as a dying process tries to get the mvar while masking exceptions
344 -- uninterruptibly.
345 join $ modifyMVar (localState node) $ \st -> case st of
346 LocalNodeValid vst -> do
347 return ( LocalNodeClosed
348 , forM_ (vst ^. localProcesses) $ \lproc ->
349 -- Semantics of 'throwTo' guarantee that target thread will get
350 -- delivered an exception. Therefore, target thread will be
351 -- killed eventually and that's as good as we can do. No need
352 -- to wait for thread to actually finish dying.
353 killThread (processThread lproc)
354 )
355 LocalNodeClosed -> return (LocalNodeClosed, return ())
356 -- This call will have the effect of shutting down the NC as well (see
357 -- 'createBareLocalNode').
358 NT.closeEndPoint (localEndPoint node)
360-- | Run a process on a local node and wait for it to finish
361runProcess :: LocalNode -> Process () -> IO ()
362runProcess node proc = do
363 done <- newEmptyMVar
364 -- TODO; When forkProcess inherits the masking state, protect the forked
365 -- thread against async exceptions that could occur before 'try' is evaluated.
366 void $ forkProcess node $ try proc >>= liftIO . putMVar done
367 takeMVar done >>= either (throwIO :: SomeException -> IO a) return
369-- | Spawn a new process on a local node
370forkProcess :: LocalNode -> Process () -> IO ProcessId
371forkProcess node proc = do
372 ms <- getMaskingState
373 modifyMVarMasked (localState node) (startProcess ms)
374 where
375 startProcess :: MaskingState
376 -> LocalNodeState
377 -> IO (LocalNodeState, ProcessId)
378 startProcess ms (LocalNodeValid vst) = do
379 let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter
380 , lpidUnique = vst ^. localPidUnique
381 }
382 let pid = ProcessId { processNodeId = localNodeId node
383 , processLocalId = lpid
384 }
385 pst <- newMVar LocalProcessState { _monitorCounter = 0
386 , _spawnCounter = 0
387 , _channelCounter = 0
388 , _typedChannels = Map.empty
389 }
390 queue <- newCQueue
391 weakQueue <- mkWeakCQueue queue (return ())
392 (_, lproc) <- fixIO $ \ ~(tid, _) -> do
393 let lproc = LocalProcess { processQueue = queue
394 , processWeakQ = weakQueue
395 , processId = pid
396 , processState = pst
397 , processThread = tid
398 , processNode = node
399 }
400 -- Rewrite this code when this is fixed:
401 -- https://ghc.haskell.org/trac/ghc/ticket/10149
402 let unmask = case ms of
403 Unmasked -> unblock
404 MaskedInterruptible -> block
405 MaskedUninterruptible -> id
406 tid' <- uninterruptibleMask_ $ forkIO $ do
407 reason <- Exception.catches
408 (unmask $ runLocalProcess lproc proc >> return DiedNormal)
409 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
410 mMsg <- unwrapMessage msg :: IO (Maybe String)
411 case mMsg of
412 Nothing -> return $ DiedException $ show ex
413 Just m -> return $ DiedException ("exit-from=" ++ (show from) ++ ",reason=" ++ m)))
414 , (Exception.Handler
415 (return . DiedException . (show :: SomeException -> String)))]
417 -- [Unified: Table 4, rules termination and exiting]
418 mconns <- modifyValidLocalState node (cleanupProcess pid)
419 -- XXX: Revisit after agreeing on the bigger picture for the semantics
420 -- of transport operations.
421 -- https://github.com/haskell-distributed/distributed-process/issues/204
422 forM_ mconns $ forkIO . mapM_ NT.close
424 writeChan (localCtrlChan node) NCMsg
425 { ctrlMsgSender = ProcessIdentifier pid
426 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
427 }
428 return (tid', lproc)
430 -- see note [tracer/forkProcess races]
431 trace node (MxSpawned pid)
433 if lpidCounter lpid == maxBound
434 then do
435 -- TODO: this doesn't look right at all - how do we know
436 -- that newUnique represents a process id that is available!?
437 newUnique <- randomIO
438 return ( LocalNodeValid
439 $ (localProcessWithId lpid ^= Just lproc)
440 . (localPidCounter ^= firstNonReservedProcessId)
441 . (localPidUnique ^= newUnique)
442 $ vst
443 , pid
444 )
445 else
446 return ( LocalNodeValid
447 $ (localProcessWithId lpid ^= Just lproc)
448 . (localPidCounter ^: (+ 1))
449 $ vst
450 , pid
451 )
452 startProcess _ LocalNodeClosed =
453 throwIO $ NodeClosedException $ localNodeId node
455 cleanupProcess :: ProcessId
456 -> ValidLocalNodeState
457 -> IO (ValidLocalNodeState, [NT.Connection])
458 cleanupProcess pid vst = do
459 let pid' = ProcessIdentifier pid
460 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)
461 return ( (localProcessWithId (processLocalId pid) ^= Nothing)
462 . (localConnections ^= unaffected)
463 $ vst
464 , map fst $ Map.elems affected
465 )
467-- note [tracer/forkProcess races]
468--
469-- Our startTracing function uses forkProcess to start the trace controller
470-- process, and of course forkProcess attempts to call traceEvent once the
471-- process has started. This is harmless, as the localEventBus is not updated
472-- until /after/ the initial forkProcess completes, so the first call to
473-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
474--
476--------------------------------------------------------------------------------
477-- Handle incoming messages --
478--------------------------------------------------------------------------------
480type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
482data IncomingTarget =
483 Uninit
484 | ToProc ProcessId (Weak (CQueue Message))
485 | ToChan TypedChannel
486 | ToNode
488data ConnectionState = ConnectionState {
489 _incoming :: !(Map NT.ConnectionId IncomingConnection)
490 , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId))
491 }
493initConnectionState :: ConnectionState
494initConnectionState = ConnectionState {
495 _incoming = Map.empty
496 , _incomingFrom = Map.empty
497 }
499incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection)
500incoming = accessor _incoming (\conns st -> st { _incoming = conns })
502incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection)
503incomingAt cid = incoming >>> DAC.mapMaybe cid
505incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId)
506incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
507 where
508 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
510handleIncomingMessages :: LocalNode -> IO ()
511handleIncomingMessages node = go initConnectionState
512 `Exception.catch` \(NodeClosedException _) -> return ()
513 where
514 go :: ConnectionState -> IO ()
515 go !st = do
516 event <- NT.receive endpoint
517 case event of
518 NT.ConnectionOpened cid rel theirAddr ->
519 if rel == NT.ReliableOrdered
520 then
521 trace node (MxConnected cid theirAddr)
522 >> go (
523 (incomingAt cid ^= Just (theirAddr, Uninit))
524 . (incomingFrom theirAddr ^: Set.insert cid)
525 $ st
526 )
527 else invalidRequest cid st
528 NT.Received cid payload ->
529 case st ^. incomingAt cid of
530 Just (_, ToProc pid weakQueue) -> do
531 mQueue <- deRefWeak weakQueue
532 forM_ mQueue $ \queue -> do
533 -- TODO: if we find that the queue is Nothing, should we remove
534 -- it from the NC state? (and same for channels, below)
535 let msg = payloadToMessage payload
536 enqueue queue msg -- 'enqueue' is strict
537 trace node (MxReceived pid msg)
538 go st
539 Just (_, ToChan (TypedChannel chan')) -> do
540 mChan <- deRefWeak chan'
541 -- If mChan is Nothing, the process has given up the read end of
542 -- the channel and we simply ignore the incoming message
543 forM_ mChan $ \chan -> atomically $
544 -- We make sure the message is fully decoded when it is enqueued
545 writeTQueue chan $! decode (BSL.fromChunks payload)
546 go st
547 Just (_, ToNode) -> do
548 let ctrlMsg = decode . BSL.fromChunks $ payload
549 writeChan ctrlChan $! ctrlMsg
550 go st
551 Just (src, Uninit) ->
552 case decode (BSL.fromChunks payload) of
553 ProcessIdentifier pid -> do
554 let lpid = processLocalId pid
555 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
556 case mProc of
557 Just proc ->
558 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
559 Nothing ->
560 invalidRequest cid st
561 SendPortIdentifier chId -> do
562 let lcid = sendPortLocalId chId
563 lpid = processLocalId (sendPortProcessId chId)
564 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
565 case mProc of
566 Just proc -> do
567 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
568 case mChannel of
569 Just channel ->
570 go (incomingAt cid ^= Just (src, ToChan channel) $ st)
571 Nothing ->
572 invalidRequest cid st
573 Nothing ->
574 invalidRequest cid st
575 NodeIdentifier nid ->
576 if nid == localNodeId node
577 then go (incomingAt cid ^= Just (src, ToNode) $ st)
578 else invalidRequest cid st
579 Nothing ->
580 invalidRequest cid st
581 NT.ConnectionClosed cid ->
582 case st ^. incomingAt cid of
583 Nothing ->
584 invalidRequest cid st
585 Just (src, _) -> do
586 trace node (MxDisconnected cid src)
587 go ( (incomingAt cid ^= Nothing)
588 . (incomingFrom src ^: Set.delete cid)
589 $ st
590 )
591 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
592 -- [Unified table 9, rule node_disconnect]
593 let nid = NodeIdentifier $ NodeId theirAddr
594 writeChan ctrlChan NCMsg
595 { ctrlMsgSender = nid
596 , ctrlMsgSignal = Died nid DiedDisconnect
597 }
598 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
599 closeImplicitReconnections node nid
600 go ( (incomingFrom theirAddr ^= Set.empty)
601 . (incoming ^: Map.filterWithKey (const . notLost))
602 $ st
603 )
604 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
605 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
606 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
607 fail $ "Cloud Haskell fatal error: transport failed: " ++ str
608 NT.EndPointClosed ->
609 return ()
610 NT.ReceivedMulticast _ _ ->
611 -- If we received a multicast message, something went horribly wrong
612 -- and we just give up
613 fail "Cloud Haskell fatal error: received unexpected multicast"
615 invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
616 invalidRequest cid st = do
617 -- TODO: We should treat this as a fatal error on the part of the remote
618 -- node. That is, we should report the remote node as having died, and we
619 -- should close incoming connections (this requires a Transport layer
620 -- extension).
621 traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),
622 (Trace cid)]
623 go ( incomingAt cid ^= Nothing
624 $ st
625 )
627 endpoint = localEndPoint node
628 ctrlChan = localCtrlChan node
630--------------------------------------------------------------------------------
631-- Top-level access to the node controller --
632--------------------------------------------------------------------------------
634runNodeController :: LocalNode -> IO ()
635runNodeController node =
636 runReaderT (evalStateT (unNC nodeController) initNCState) node
637 `Exception.catch` \(NodeClosedException _) -> return ()
639--------------------------------------------------------------------------------
640-- Internal data types --
641--------------------------------------------------------------------------------
643data NCState = NCState
644 { -- Mapping from remote processes to linked local processes
645 _links :: !(Map Identifier (Set ProcessId))
646 -- Mapping from remote processes to monitoring local processes
647 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
648 -- Process registry: names and where they live, mapped to the PIDs
649 , _registeredHere :: !(Map String ProcessId)
650 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
651 }
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 { _links = Map.empty
664 , _monitors = Map.empty
665 , _registeredHere = Map.empty
666 , _registeredOnNodes = Map.empty
667 }
669-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This
670-- type is deliberately not exported so it cannot be caught explicitly.
671data ProcessKillException =
672 ProcessKillException !ProcessId !String
673 deriving (Typeable)
675instance Exception ProcessKillException
676instance Show ProcessKillException where
677 show (ProcessKillException pid reason) =
678 "killed-by=" ++ show pid ++ ",reason=" ++ reason
680ncSendToProcess :: ProcessId -> Message -> NC ()
681ncSendToProcess = ncSendToProcessAndTrace True
683ncSendToProcessAndTrace :: Bool -> ProcessId -> Message -> NC ()
684ncSendToProcessAndTrace shouldTrace pid msg = do
685 node <- ask
686 if processNodeId pid == localNodeId node
687 then ncEffectLocalSendAndTrace shouldTrace node pid msg
688 else liftIO $ sendBinary node
689 (NodeIdentifier $ localNodeId node)
690 (NodeIdentifier $ processNodeId pid)
691 WithImplicitReconnect
692 NCMsg { ctrlMsgSender = NodeIdentifier $ localNodeId node
693 , ctrlMsgSignal = UnreliableSend (processLocalId pid) msg
694 }
696ncSendToNode :: NodeId -> NCMsg -> NC ()
697ncSendToNode to msg = do
698 node <- ask
699 liftIO $ if to == localNodeId node
700 then writeChan (localCtrlChan node) $! msg
701 else sendBinary node
702 (NodeIdentifier $ localNodeId node)
703 (NodeIdentifier to)
704 WithImplicitReconnect
705 msg
707--------------------------------------------------------------------------------
708-- Tracing/Debugging --
709--------------------------------------------------------------------------------
711-- [Issue #104 / DP-13]
713traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
714traceNotifyDied node ident reason =
715 -- TODO: sendPortDied notifications
716 liftIO $ withLocalTracer node $ \t ->
717 case ident of
718 (NodeIdentifier nid) -> traceEvent t (MxNodeDied nid reason)
719 (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)
720 _ -> return ()
722traceEventFmtIO :: LocalNode
723 -> String
724 -> [TraceArg]
725 -> IO ()
726traceEventFmtIO node fmt args =
727 withLocalTracer node $ \t -> traceLogFmt t fmt args
729trace :: LocalNode -> MxEvent -> IO ()
730trace node ev = withLocalTracer node $ \t -> traceEvent t ev
732withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()
733withLocalTracer node act = act (localEventBus node)
735--------------------------------------------------------------------------------
736-- Core functionality --
737--------------------------------------------------------------------------------
739-- [Unified: Table 7]
740nodeController :: NC ()
741nodeController = do
742 node <- ask
743 forever' $ do
744 msg <- liftIO $ readChan (localCtrlChan node)
746 -- [Unified: Table 7, rule nc_forward]
747 case destNid (ctrlMsgSignal msg) of
748 Just nid' | nid' /= localNodeId node ->
749 ncSendToNode nid' msg
750 _ ->
751 return ()
753 case msg of
754 NCMsg (ProcessIdentifier from) (Link them) ->
755 ncEffectMonitor from them Nothing
756 NCMsg (ProcessIdentifier from) (Monitor ref) ->
757 ncEffectMonitor from (monitorRefIdent ref) (Just ref)
758 NCMsg (ProcessIdentifier from) (Unlink them) ->
759 ncEffectUnlink from them
760 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->
761 ncEffectUnmonitor from ref
762 NCMsg _from (Died ident reason) ->
763 ncEffectDied ident reason
764 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
765 ncEffectSpawn from proc ref
766 NCMsg (ProcessIdentifier from) (Register label atnode pid force) ->
767 ncEffectRegister from label atnode pid force
768 NCMsg (ProcessIdentifier from) (WhereIs label) ->
769 ncEffectWhereIs from label
770 NCMsg _ (NamedSend label msg') ->
771 ncEffectNamedSend label msg'
772 NCMsg _ (UnreliableSend lpid msg') ->
773 ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg'
774 NCMsg _ (LocalSend to msg') ->
775 ncEffectLocalSend node to msg'
776 NCMsg _ (LocalPortSend to msg') ->
777 ncEffectLocalPortSend to msg'
778 NCMsg (ProcessIdentifier from) (Kill to reason) ->
779 ncEffectKill from to reason
780 NCMsg (ProcessIdentifier from) (Exit to reason) ->
781 ncEffectExit from to reason
782 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
783 ncEffectGetInfo from pid
784 NCMsg _ SigShutdown ->
785 liftIO $ do
786 NT.closeEndPoint (localEndPoint node)
787 `Exception.finally` throwIO (NodeClosedException $ localNodeId node)
788 NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->
789 ncEffectGetNodeStats from nid
790 unexpected ->
791 error $ "nodeController: unexpected message " ++ show unexpected
793-- [Unified: Table 10]
794ncEffectMonitor :: ProcessId -- ^ Who's watching?
795 -> Identifier -- ^ Who's being watched?
796 -> Maybe MonitorRef -- ^ 'Nothing' to link
797 -> NC ()
798ncEffectMonitor from them mRef = do
799 node <- ask
800 shouldLink <-
801 if not (isLocal node them)
802 then return True
803 else isValidLocalIdentifier them
804 case (shouldLink, isLocal node (ProcessIdentifier from)) of
805 (True, _) -> -- [Unified: first rule]
806 case mRef of
807 Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)
808 Nothing -> modify' $ linksFor them ^: Set.insert from
809 (False, True) -> -- [Unified: second rule]
810 notifyDied from them DiedUnknownId mRef
811 (False, False) -> -- [Unified: third rule]
812 -- TODO: this is the right sender according to the Unified semantics,
813 -- but perhaps having 'them' as the sender would make more sense
814 -- (see also: notifyDied)
815 ncSendToNode (processNodeId from) $ NCMsg
816 { ctrlMsgSender = NodeIdentifier (localNodeId node)
817 , ctrlMsgSignal = Died them DiedUnknownId
818 }
820-- [Unified: Table 11]
821ncEffectUnlink :: ProcessId -> Identifier -> NC ()
822ncEffectUnlink from them = do
823 node <- ask
824 when (isLocal node (ProcessIdentifier from)) $
825 case them of
826 ProcessIdentifier pid ->
827 postAsMessage from $ DidUnlinkProcess pid
828 NodeIdentifier nid ->
829 postAsMessage from $ DidUnlinkNode nid
830 SendPortIdentifier cid ->
831 postAsMessage from $ DidUnlinkPort cid
832 modify' $ linksFor them ^: Set.delete from
834-- [Unified: Table 11]
835ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
836ncEffectUnmonitor from ref = do
837 node <- ask
838 when (isLocal node (ProcessIdentifier from)) $
839 postAsMessage from $ DidUnmonitor ref
840 modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
842-- [Unified: Table 12]
843ncEffectDied :: Identifier -> DiedReason -> NC ()
844ncEffectDied ident reason = do
845 node <- ask
846 traceNotifyDied node ident reason
847 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
848 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
850-- _registry :: !(Map (String,NodeId) ProcessId)
852 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
854 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
855 forM_ uss $ \us ->
856 when (localOnly <= isLocal node (ProcessIdentifier us)) $
857 notifyDied us them reason Nothing
859 forM_ (Map.toList affectedMons) $ \(them, refs) ->
860 forM_ refs $ \(us, ref) ->
861 when (localOnly <= isLocal node (ProcessIdentifier us)) $
862 notifyDied us them reason (Just ref)
864 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
866 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
868 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
869 mapM (\(pid,nidlist) ->
870 case ident `impliesDeathOf` ProcessIdentifier pid of
871 True ->
872 do forM_ nidlist $ \(nid,_) ->
873 when (not $ isLocal node (NodeIdentifier nid))
874 (forwardNameDeath node nid)
875 return Nothing
876 False -> return $ Just (pid,nidlist) )
877 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
878 where
879 forwardNameDeath node nid = ncSendToNode nid
880 NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node)
881 , ctrlMsgSignal = Died ident reason
882 }
884-- [Unified: Table 13]
885ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
886ncEffectSpawn pid cProc ref = do
887 mProc <- unClosure cProc
888 -- If the closure does not exist, we spawn a process that throws an exception
889 -- This allows the remote node to find out what's happening
890 -- TODO:
891 let proc = case mProc of
892 Left err -> fail $ "Error: Could not resolve closure: " ++ err
893 Right p -> p
894 node <- ask
895 pid' <- liftIO $ forkProcess node proc
896 ncSendToProcess pid $ unsafeCreateUnencodedMessage $ DidSpawn ref pid'
898-- Unified semantics does not explicitly describe how to implement 'register',
899-- but mentions it's "very similar to nsend" (Table 14)
900-- We send a response indicated if the operation is invalid
901ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
902ncEffectRegister from label atnode mPid reregistration = do
903 node <- ask
904 currentVal <- gets (^. registeredHereFor label)
905 isOk <-
906 case mPid of
907 Nothing -> -- unregister request
908 return $ isJust currentVal
909 Just thepid -> -- register request
910 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
911 return $ (isNothing currentVal /= reregistration) &&
912 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
913 if isLocal node (NodeIdentifier atnode)
914 then do when isOk $
915 do modify' $ registeredHereFor label ^= mPid
916 updateRemote node currentVal mPid
917 case mPid of
918 (Just p) -> liftIO $ trace node (MxRegistered p label)
919 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
920 newVal <- gets (^. registeredHereFor label)
921 ncSendToProcess from $ unsafeCreateUnencodedMessage $
922 RegisterReply label isOk newVal
923 else let operation =
924 case reregistration of
925 True -> flip decList
926 False -> flip incList
927 in case mPid of
928 Nothing -> return ()
929 Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
930 where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
931 do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
932 forward node (processNodeId newval) (Register label atnode (Just newval) False)
933 updateRemote node Nothing (Just newval) =
934 forward node (processNodeId newval) (Register label atnode (Just newval) False)
935 updateRemote node (Just oldval) Nothing =
936 forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
937 updateRemote _ _ _ = return ()
938 maybeify f Nothing = unmaybeify $ f []
939 maybeify f (Just x) = unmaybeify $ f x
941 unmaybeify [] = Nothing
942 unmaybeify x = Just x
943 incList [] tag = [(tag,1)]
944 incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
945 incList (x:xs) tag = x : incList xs tag
946 decList [] _ = []
947 decList ((atag,1):xs) tag | atag == tag = xs
948 decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
949 decList (x:xs) tag = x:decList xs tag
950 forward node to reg =
951 when (not $ isLocal node (NodeIdentifier to)) $
952 ncSendToNode to $ NCMsg { ctrlMsgSender = ProcessIdentifier from
953 , ctrlMsgSignal = reg
954 }
957-- Unified semantics does not explicitly describe 'whereis'
958ncEffectWhereIs :: ProcessId -> String -> NC ()
959ncEffectWhereIs from label = do
960 mPid <- gets (^. registeredHereFor label)
961 ncSendToProcess from $ unsafeCreateUnencodedMessage $ WhereIsReply label mPid
963-- [Unified: Table 14]
964ncEffectNamedSend :: String -> Message -> NC ()
965ncEffectNamedSend label msg = do
966 mPid <- gets (^. registeredHereFor label)
967 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
968 forM_ mPid $ \to ->
969 -- If this is a trace message we don't trace it to avoid entering a loop
970 -- where trace messages produce more trace messages.
971 ncSendToProcessAndTrace (label /= "trace.logger") to msg
973-- [Issue #DP-20]
974ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
975ncEffectLocalSend = ncEffectLocalSendAndTrace True
977ncEffectLocalSendAndTrace :: Bool -> LocalNode -> ProcessId -> Message -> NC ()
978ncEffectLocalSendAndTrace shouldTrace node to msg =
979 liftIO $ withLocalProc node to $ \p -> do
980 enqueue (processQueue p) msg
981 when shouldTrace $ trace node (MxReceived to msg)
983-- [Issue #DP-20]
984ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
985ncEffectLocalPortSend from msg = do
986 node <- ask
987 let pid = sendPortProcessId from
988 cid = sendPortLocalId from
989 liftIO $ withLocalProc node pid $ \proc -> do
990 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
991 case mChan of
992 -- in the unlikely event we know nothing about this channel id,
993 -- there's little to be done - perhaps some logging/tracing though...
994 Nothing -> return ()
995 Just (TypedChannel chan') -> do
996 -- If ch is Nothing, the process has given up the read end of
997 -- the channel and we simply ignore the incoming message - this
998 ch <- deRefWeak chan'
999 forM_ ch $ \chan -> deliverChan msg chan
1000 where deliverChan :: forall a . Message -> TQueue a -> IO ()
1001 deliverChan (UnencodedMessage _ raw) chan' =
1002 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1003 deliverChan (EncodedMessage _ _) _ =
1004 -- this will not happen unless someone screws with Primitives.hs
1005 error "invalid local channel delivery"
1007-- [Issue #69]
1008ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()
1009ncEffectKill from to reason = do
1010 node <- ask
1011 when (isLocal node (ProcessIdentifier to)) $
1012 throwException to $ ProcessKillException from reason
1014-- [Issue #69]
1015ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()
1016ncEffectExit from to reason = do
1017 node <- ask
1018 when (isLocal node (ProcessIdentifier to)) $
1019 throwException to $ ProcessExitException from reason
1021-- [Issue #89]
1022ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()
1023ncEffectGetInfo from pid =
1024 let lpid = processLocalId pid
1025 them = (ProcessIdentifier pid)
1026 in do
1027 node <- ask
1028 mProc <- liftIO $ withValidLocalState node
1029 $ return . (^. localProcessWithId lpid)
1030 case mProc of
1031 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
1032 from (ProcessInfoNone DiedUnknownId)
1033 Just proc -> do
1034 itsLinks <- gets (^. linksFor them)
1035 itsMons <- gets (^. monitorsFor them)
1036 registered <- gets (^. registeredHere)
1037 size <- liftIO $ queueSize $ processQueue $ proc
1039 let reg = registeredNames registered
1040 dispatch (isLocal node (ProcessIdentifier from))
1041 from
1042 ProcessInfo {
1043 infoNode = (processNodeId pid)
1044 , infoRegisteredNames = reg
1045 , infoMessageQueueLength = size
1046 , infoMonitors = Set.toList itsMons
1047 , infoLinks = Set.toList itsLinks
1049 where dispatch :: (Serializable a, Show a)
1050 => Bool
1051 -> ProcessId
1052 -> a
1053 -> NC ()
1054 dispatch True dest pInfo = postAsMessage dest $ pInfo
1055 dispatch False dest pInfo =
1056 ncSendToProcess dest $ unsafeCreateUnencodedMessage pInfo
1058 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
1059 then (k:ks)
1060 else ks) []
1062ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()
1063ncEffectGetNodeStats from _nid = do
1064 node <- ask
1065 ncState <- StateT.get
1066 nodeState <- liftIO $ withValidLocalState node return
1067 let stats =
1068 NodeStats {
1069 nodeStatsNode = localNodeId node
1070 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
1071 , nodeStatsMonitors = Map.size $ ncState ^. monitors
1072 , nodeStatsLinks = Map.size $ ncState ^. links
1073 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
1075 postAsMessage from stats
1077--------------------------------------------------------------------------------
1078-- Auxiliary --
1079--------------------------------------------------------------------------------
1081notifyDied :: ProcessId -- ^ Who to notify?
1082 -> Identifier -- ^ Who died?
1083 -> DiedReason -- ^ How did they die?
1084 -> Maybe MonitorRef -- ^ 'Nothing' for linking
1085 -> NC ()
1086notifyDied dest src reason mRef = do
1087 node <- ask
1088 case (isLocal node (ProcessIdentifier dest), mRef, src) of
1089 (True, Just ref, ProcessIdentifier pid) ->
1090 postAsMessage dest $ ProcessMonitorNotification ref pid reason
1091 (True, Just ref, NodeIdentifier nid) ->
1092 postAsMessage dest $ NodeMonitorNotification ref nid reason
1093 (True, Just ref, SendPortIdentifier cid) ->
1094 postAsMessage dest $ PortMonitorNotification ref cid reason
1095 (True, Nothing, ProcessIdentifier pid) ->
1096 throwException dest $ ProcessLinkException pid reason
1097 (True, Nothing, NodeIdentifier pid) ->
1098 throwException dest $ NodeLinkException pid reason
1099 (True, Nothing, SendPortIdentifier pid) ->
1100 throwException dest $ PortLinkException pid reason
1101 (False, _, _) ->
1102 -- The change in sender comes from [Unified: Table 10]
1103 ncSendToNode (processNodeId dest) $ NCMsg
1104 { ctrlMsgSender = NodeIdentifier (localNodeId node)
1105 , ctrlMsgSignal = Died src reason
1108-- | [Unified: Table 8]
1109destNid :: ProcessSignal -> Maybe NodeId
1110destNid (Link ident) = Just $ nodeOf ident
1111destNid (Unlink ident) = Just $ nodeOf ident
1112destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)
1113destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
1114destNid (Spawn _ _) = Nothing
1115destNid (Register _ _ _ _) = Nothing
1116destNid (WhereIs _) = Nothing
1117destNid (NamedSend _ _) = Nothing
1118destNid (UnreliableSend _ _) = Nothing
1119-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
1120-- then when a local process dies the monitoring/linking machinery will take
1121-- care of notifying remote nodes
1122destNid (Died _ _) = Nothing
1123destNid (Kill pid _) = Just $ processNodeId pid
1124destNid (Exit pid _) = Just $ processNodeId pid
1125destNid (GetInfo pid) = Just $ processNodeId pid
1126destNid (GetNodeStats nid) = Just nid
1127destNid (LocalSend pid _) = Just $ processNodeId pid
1128destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1129destNid (SigShutdown) = Nothing
1131-- | Check if a process is local to our own node
1132isLocal :: LocalNode -> Identifier -> Bool
1133isLocal nid ident = nodeOf ident == localNodeId nid
1135-- | Lookup a local closure
1136unClosure :: Typeable a => Closure a -> NC (Either String a)
1137unClosure closure = do
1138 rtable <- remoteTable <$> ask
1139 return (Static.unclosure rtable closure)
1141-- | Check if an identifier refers to a valid local object
1142isValidLocalIdentifier :: Identifier -> NC Bool
1143isValidLocalIdentifier ident = do
1144 node <- ask
1145 liftIO . withValidLocalState node $ \nSt ->
1146 case ident of
1147 NodeIdentifier nid ->
1148 return $ nid == localNodeId node
1149 ProcessIdentifier pid -> do
1150 let mProc = nSt ^. localProcessWithId (processLocalId pid)
1151 return $ isJust mProc
1152 SendPortIdentifier cid -> do
1153 let pid = sendPortProcessId cid
1154 mProc = nSt ^. localProcessWithId (processLocalId pid)
1155 case mProc of
1156 Nothing -> return False
1157 Just proc -> withMVar (processState proc) $ \pSt -> do
1158 let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)
1159 return $ isJust mCh
1161--------------------------------------------------------------------------------
1162-- Messages to local processes --
1163--------------------------------------------------------------------------------
1165postAsMessage :: Serializable a => ProcessId -> a -> NC ()
1166postAsMessage pid = postMessage pid . createUnencodedMessage
1168postMessage :: ProcessId -> Message -> NC ()
1169postMessage pid msg = do
1170 node <- ask
1171 liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
1173throwException :: Exception e => ProcessId -> e -> NC ()
1174throwException pid e = do
1175 node <- ask
1176 -- throwTo blocks until the exception is received by the target thread.
1177 -- We cannot easily make it happen asynchronpusly because then 'unlink'
1178 -- semantics would break.
1179 liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1181withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1182withLocalProc node pid p =
1183 -- By [Unified: table 6, rule missing_process] messages to dead processes
1184 -- can silently be dropped
1185 let lpid = processLocalId pid in do
1186 join $ withValidLocalState node $ \vst ->
1187 return $ forM_ (vst ^. localProcessWithId lpid) p
1189--------------------------------------------------------------------------------
1190-- Accessors --
1191--------------------------------------------------------------------------------
1193links :: Accessor NCState (Map Identifier (Set ProcessId))
1194links = accessor _links (\ls st -> st { _links = ls })
1196monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
1197monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1199registeredHere :: Accessor NCState (Map String ProcessId)
1200registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
1202registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1203registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1205linksFor :: Identifier -> Accessor NCState (Set ProcessId)
1206linksFor ident = links >>> DAC.mapDefault Set.empty ident
1208monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
1209monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
1211registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1212registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1214registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
1215registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1217-- | @splitNotif ident@ splits a notifications map into those
1218-- notifications that should trigger when 'ident' fails and those links that
1219-- should not.
1221-- There is a hierarchy between identifiers: failure of a node implies failure
1222-- of all processes on that node, and failure of a process implies failure of
1223-- all typed channels to that process. In other words, if 'ident' refers to a
1224-- node, then the /should trigger/ set will include
1226-- * the notifications for the node specifically
1227-- * the notifications for processes on that node, and
1228-- * the notifications for typed channels to processes on that node.
1230-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
1231-- include
1233-- * the notifications for that process specifically and
1234-- * the notifications for typed channels to that process.
1236-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1237splitNotif :: Identifier
1238 -> Map Identifier a
1239 -> (Map Identifier a, Map Identifier a)
1240splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
1242--------------------------------------------------------------------------------
1243-- Auxiliary --
1244--------------------------------------------------------------------------------
1246-- | Modify and evaluate the state
1247modify' :: MonadState s m => (s -> s) -> m ()
1248modify' f = StateT.get >>= \s -> StateT.put $! f s