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