git.iko.soy/distributed-process
on 12cdcfc help sign in
distributed-process/src/Control/Distributed/Process/Node.hs
1254 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.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 "attempt to connect with unsupported reliability " ++ show rel
521 NT.Received cid payload ->
522 case st ^. incomingAt cid of
523 Just (_, ToProc pid weakQueue) -> do
524 mQueue <- deRefWeak weakQueue
525 forM_ mQueue $ \queue -> do
526 -- TODO: if we find that the queue is Nothing, should we remove
527 -- it from the NC state? (and same for channels, below)
528 let msg = payloadToMessage payload
529 enqueue queue msg -- 'enqueue' is strict
530 trace node (MxReceived pid msg)
531 go st
532 Just (_, ToChan (TypedChannel chan')) -> do
533 mChan <- deRefWeak chan'
534 -- If mChan is Nothing, the process has given up the read end of
535 -- the channel and we simply ignore the incoming message
536 forM_ mChan $ \chan -> atomically $
537 -- We make sure the message is fully decoded when it is enqueued
538 writeTQueue chan $! decode (BSL.fromChunks payload)
539 go st
540 Just (_, ToNode) -> do
541 let ctrlMsg = decode . BSL.fromChunks $ payload
542 writeChan ctrlChan $! ctrlMsg
543 go st
544 Just (src, Uninit) ->
545 case decode (BSL.fromChunks payload) of
546 ProcessIdentifier pid -> do
547 let lpid = processLocalId pid
548 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
549 case mProc of
550 Just proc ->
551 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
552 Nothing ->
553 invalidRequest cid st $
554 "incoming attempt to connect to unknown process "
555 ++ show pid
556 SendPortIdentifier chId -> do
557 let lcid = sendPortLocalId chId
558 lpid = processLocalId (sendPortProcessId chId)
559 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
560 case mProc of
561 Just proc -> do
562 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
563 case mChannel of
564 Just channel ->
565 go (incomingAt cid ^= Just (src, ToChan channel) $ st)
566 Nothing ->
567 invalidRequest cid st $
568 "incoming attempt to connect to unknown channel of"
569 ++ " process " ++ show (sendPortProcessId chId)
570 Nothing ->
571 invalidRequest cid st $
572 "incoming attempt to connect to channel of unknown"
573 ++ " process " ++ show (sendPortProcessId chId)
574 NodeIdentifier nid ->
575 if nid == localNodeId node
576 then go (incomingAt cid ^= Just (src, ToNode) $ st)
577 else invalidRequest cid st $
578 "incoming attempt to connect to a different node -"
579 ++ " I'm " ++ show (localNodeId node)
580 ++ " but the remote peer wants to connect to "
581 ++ show nid
582 Nothing ->
583 invalidRequest cid st
584 "message received from an unknown connection"
585 NT.ConnectionClosed cid ->
586 case st ^. incomingAt cid of
587 Nothing ->
588 invalidRequest cid st "closed unknown connection"
589 Just (src, _) -> do
590 trace node (MxDisconnected cid src)
591 go ( (incomingAt cid ^= Nothing)
592 . (incomingFrom src ^: Set.delete cid)
593 $ st
594 )
595 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
596 -- [Unified table 9, rule node_disconnect]
597 let nid = NodeIdentifier $ NodeId theirAddr
598 writeChan ctrlChan NCMsg
599 { ctrlMsgSender = nid
600 , ctrlMsgSignal = Died nid DiedDisconnect
601 }
602 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
603 closeImplicitReconnections node nid
604 go ( (incomingFrom theirAddr ^= Set.empty)
605 . (incoming ^: Map.filterWithKey (const . notLost))
606 $ st
607 )
608 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
609 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
610 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
611 fail $ "Cloud Haskell fatal error: transport failed: " ++ str
612 NT.EndPointClosed ->
613 return ()
614 NT.ReceivedMulticast _ _ ->
615 -- If we received a multicast message, something went horribly wrong
616 -- and we just give up
617 fail "Cloud Haskell fatal error: received unexpected multicast"
619 invalidRequest :: NT.ConnectionId -> ConnectionState -> String -> IO ()
620 invalidRequest cid st msg = do
621 -- TODO: We should treat this as a fatal error on the part of the remote
622 -- node. That is, we should report the remote node as having died, and we
623 -- should close incoming connections (this requires a Transport layer
624 -- extension).
625 traceEventFmtIO node "" [ TraceStr $ " [network] invalid request"
626 ++ " (" ++ msg ++ "): "
627 , (Trace cid)
628 ]
629 go ( incomingAt cid ^= Nothing
630 $ st
631 )
633 endpoint = localEndPoint node
634 ctrlChan = localCtrlChan node
636--------------------------------------------------------------------------------
637-- Top-level access to the node controller --
638--------------------------------------------------------------------------------
640runNodeController :: LocalNode -> IO ()
641runNodeController node =
642 runReaderT (evalStateT (unNC nodeController) initNCState) node
643 `Exception.catch` \(NodeClosedException _) -> return ()
645--------------------------------------------------------------------------------
646-- Internal data types --
647--------------------------------------------------------------------------------
649data NCState = NCState
650 { -- Mapping from remote processes to linked local processes
651 _links :: !(Map Identifier (Set ProcessId))
652 -- Mapping from remote processes to monitoring local processes
653 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
654 -- Process registry: names and where they live, mapped to the PIDs
655 , _registeredHere :: !(Map String ProcessId)
656 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
657 }
659newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
660 deriving ( Applicative
661 , Functor
662 , Monad
663 , MonadIO
664 , MonadState NCState
665 , MonadReader LocalNode
666 )
668initNCState :: NCState
669initNCState = NCState { _links = Map.empty
670 , _monitors = Map.empty
671 , _registeredHere = Map.empty
672 , _registeredOnNodes = Map.empty
673 }
675-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This
676-- type is deliberately not exported so it cannot be caught explicitly.
677data ProcessKillException =
678 ProcessKillException !ProcessId !String
679 deriving (Typeable)
681instance Exception ProcessKillException
682instance Show ProcessKillException where
683 show (ProcessKillException pid reason) =
684 "killed-by=" ++ show pid ++ ",reason=" ++ reason
686ncSendToProcess :: ProcessId -> Message -> NC ()
687ncSendToProcess = ncSendToProcessAndTrace True
689ncSendToProcessAndTrace :: Bool -> ProcessId -> Message -> NC ()
690ncSendToProcessAndTrace shouldTrace pid msg = do
691 node <- ask
692 if processNodeId pid == localNodeId node
693 then ncEffectLocalSendAndTrace shouldTrace node pid msg
694 else liftIO $ sendBinary node
695 (NodeIdentifier $ localNodeId node)
696 (NodeIdentifier $ processNodeId pid)
697 WithImplicitReconnect
698 NCMsg { ctrlMsgSender = NodeIdentifier $ localNodeId node
699 , ctrlMsgSignal = UnreliableSend (processLocalId pid) msg
700 }
702ncSendToNode :: NodeId -> NCMsg -> NC ()
703ncSendToNode to msg = do
704 node <- ask
705 liftIO $ if to == localNodeId node
706 then writeChan (localCtrlChan node) $! msg
707 else sendBinary node
708 (NodeIdentifier $ localNodeId node)
709 (NodeIdentifier to)
710 WithImplicitReconnect
711 msg
713--------------------------------------------------------------------------------
714-- Tracing/Debugging --
715--------------------------------------------------------------------------------
717-- [Issue #104 / DP-13]
719traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
720traceNotifyDied node ident reason =
721 -- TODO: sendPortDied notifications
722 liftIO $ withLocalTracer node $ \t ->
723 case ident of
724 (NodeIdentifier nid) -> traceEvent t (MxNodeDied nid reason)
725 (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)
726 _ -> return ()
728traceEventFmtIO :: LocalNode
729 -> String
730 -> [TraceArg]
731 -> IO ()
732traceEventFmtIO node fmt args =
733 withLocalTracer node $ \t -> traceLogFmt t fmt args
735trace :: LocalNode -> MxEvent -> IO ()
736trace node ev = withLocalTracer node $ \t -> traceEvent t ev
738withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()
739withLocalTracer node act = act (localEventBus node)
741--------------------------------------------------------------------------------
742-- Core functionality --
743--------------------------------------------------------------------------------
745-- [Unified: Table 7]
746nodeController :: NC ()
747nodeController = do
748 node <- ask
749 forever' $ do
750 msg <- liftIO $ readChan (localCtrlChan node)
752 -- [Unified: Table 7, rule nc_forward]
753 case destNid (ctrlMsgSignal msg) of
754 Just nid' | nid' /= localNodeId node ->
755 ncSendToNode nid' msg
756 _ ->
757 return ()
759 case msg of
760 NCMsg (ProcessIdentifier from) (Link them) ->
761 ncEffectMonitor from them Nothing
762 NCMsg (ProcessIdentifier from) (Monitor ref) ->
763 ncEffectMonitor from (monitorRefIdent ref) (Just ref)
764 NCMsg (ProcessIdentifier from) (Unlink them) ->
765 ncEffectUnlink from them
766 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->
767 ncEffectUnmonitor from ref
768 NCMsg _from (Died ident reason) ->
769 ncEffectDied ident reason
770 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
771 ncEffectSpawn from proc ref
772 NCMsg (ProcessIdentifier from) (Register label atnode pid force) ->
773 ncEffectRegister from label atnode pid force
774 NCMsg (ProcessIdentifier from) (WhereIs label) ->
775 ncEffectWhereIs from label
776 NCMsg _ (NamedSend label msg') ->
777 ncEffectNamedSend label msg'
778 NCMsg _ (UnreliableSend lpid msg') ->
779 ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg'
780 NCMsg _ (LocalSend to msg') ->
781 ncEffectLocalSend node to msg'
782 NCMsg _ (LocalPortSend to msg') ->
783 ncEffectLocalPortSend to msg'
784 NCMsg (ProcessIdentifier from) (Kill to reason) ->
785 ncEffectKill from to reason
786 NCMsg (ProcessIdentifier from) (Exit to reason) ->
787 ncEffectExit from to reason
788 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
789 ncEffectGetInfo from pid
790 NCMsg _ SigShutdown ->
791 liftIO $ do
792 NT.closeEndPoint (localEndPoint node)
793 `Exception.finally` throwIO (NodeClosedException $ localNodeId node)
794 NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->
795 ncEffectGetNodeStats from nid
796 unexpected ->
797 error $ "nodeController: unexpected message " ++ show unexpected
799-- [Unified: Table 10]
800ncEffectMonitor :: ProcessId -- ^ Who's watching?
801 -> Identifier -- ^ Who's being watched?
802 -> Maybe MonitorRef -- ^ 'Nothing' to link
803 -> NC ()
804ncEffectMonitor from them mRef = do
805 node <- ask
806 shouldLink <-
807 if not (isLocal node them)
808 then return True
809 else isValidLocalIdentifier them
810 case (shouldLink, isLocal node (ProcessIdentifier from)) of
811 (True, _) -> -- [Unified: first rule]
812 case mRef of
813 Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)
814 Nothing -> modify' $ linksFor them ^: Set.insert from
815 (False, True) -> -- [Unified: second rule]
816 notifyDied from them DiedUnknownId mRef
817 (False, False) -> -- [Unified: third rule]
818 -- TODO: this is the right sender according to the Unified semantics,
819 -- but perhaps having 'them' as the sender would make more sense
820 -- (see also: notifyDied)
821 ncSendToNode (processNodeId from) $ NCMsg
822 { ctrlMsgSender = NodeIdentifier (localNodeId node)
823 , ctrlMsgSignal = Died them DiedUnknownId
824 }
826-- [Unified: Table 11]
827ncEffectUnlink :: ProcessId -> Identifier -> NC ()
828ncEffectUnlink from them = do
829 node <- ask
830 when (isLocal node (ProcessIdentifier from)) $
831 case them of
832 ProcessIdentifier pid ->
833 postAsMessage from $ DidUnlinkProcess pid
834 NodeIdentifier nid ->
835 postAsMessage from $ DidUnlinkNode nid
836 SendPortIdentifier cid ->
837 postAsMessage from $ DidUnlinkPort cid
838 modify' $ linksFor them ^: Set.delete from
840-- [Unified: Table 11]
841ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
842ncEffectUnmonitor from ref = do
843 node <- ask
844 when (isLocal node (ProcessIdentifier from)) $
845 postAsMessage from $ DidUnmonitor ref
846 modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
848-- [Unified: Table 12]
849ncEffectDied :: Identifier -> DiedReason -> NC ()
850ncEffectDied ident reason = do
851 node <- ask
852 traceNotifyDied node ident reason
853 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
854 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
856-- _registry :: !(Map (String,NodeId) ProcessId)
858 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
860 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
861 forM_ uss $ \us ->
862 when (localOnly <= isLocal node (ProcessIdentifier us)) $
863 notifyDied us them reason Nothing
865 forM_ (Map.toList affectedMons) $ \(them, refs) ->
866 forM_ refs $ \(us, ref) ->
867 when (localOnly <= isLocal node (ProcessIdentifier us)) $
868 notifyDied us them reason (Just ref)
870 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
872 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
874 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
875 mapM (\(pid,nidlist) ->
876 case ident `impliesDeathOf` ProcessIdentifier pid of
877 True ->
878 do forM_ nidlist $ \(nid,_) ->
879 when (not $ isLocal node (NodeIdentifier nid))
880 (forwardNameDeath node nid)
881 return Nothing
882 False -> return $ Just (pid,nidlist) )
883 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
884 where
885 forwardNameDeath node nid = ncSendToNode nid
886 NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node)
887 , ctrlMsgSignal = Died ident reason
888 }
890-- [Unified: Table 13]
891ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
892ncEffectSpawn pid cProc ref = do
893 mProc <- unClosure cProc
894 -- If the closure does not exist, we spawn a process that throws an exception
895 -- This allows the remote node to find out what's happening
896 -- TODO:
897 let proc = case mProc of
898 Left err -> fail $ "Error: Could not resolve closure: " ++ err
899 Right p -> p
900 node <- ask
901 pid' <- liftIO $ forkProcess node proc
902 ncSendToProcess pid $ unsafeCreateUnencodedMessage $ DidSpawn ref pid'
904-- Unified semantics does not explicitly describe how to implement 'register',
905-- but mentions it's "very similar to nsend" (Table 14)
906-- We send a response indicated if the operation is invalid
907ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
908ncEffectRegister from label atnode mPid reregistration = do
909 node <- ask
910 currentVal <- gets (^. registeredHereFor label)
911 isOk <-
912 case mPid of
913 Nothing -> -- unregister request
914 return $ isJust currentVal
915 Just thepid -> -- register request
916 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
917 return $ (isNothing currentVal /= reregistration) &&
918 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
919 if isLocal node (NodeIdentifier atnode)
920 then do when isOk $
921 do modify' $ registeredHereFor label ^= mPid
922 updateRemote node currentVal mPid
923 case mPid of
924 (Just p) -> liftIO $ trace node (MxRegistered p label)
925 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
926 newVal <- gets (^. registeredHereFor label)
927 ncSendToProcess from $ unsafeCreateUnencodedMessage $
928 RegisterReply label isOk newVal
929 else let operation =
930 case reregistration of
931 True -> flip decList
932 False -> flip incList
933 in case mPid of
934 Nothing -> return ()
935 Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
936 where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
937 do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
938 forward node (processNodeId newval) (Register label atnode (Just newval) False)
939 updateRemote node Nothing (Just newval) =
940 forward node (processNodeId newval) (Register label atnode (Just newval) False)
941 updateRemote node (Just oldval) Nothing =
942 forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
943 updateRemote _ _ _ = return ()
944 maybeify f Nothing = unmaybeify $ f []
945 maybeify f (Just x) = unmaybeify $ f x
947 unmaybeify [] = Nothing
948 unmaybeify x = Just x
949 incList [] tag = [(tag,1)]
950 incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
951 incList (x:xs) tag = x : incList xs tag
952 decList [] _ = []
953 decList ((atag,1):xs) tag | atag == tag = xs
954 decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
955 decList (x:xs) tag = x:decList xs tag
956 forward node to reg =
957 when (not $ isLocal node (NodeIdentifier to)) $
958 ncSendToNode to $ NCMsg { ctrlMsgSender = ProcessIdentifier from
959 , ctrlMsgSignal = reg
960 }
963-- Unified semantics does not explicitly describe 'whereis'
964ncEffectWhereIs :: ProcessId -> String -> NC ()
965ncEffectWhereIs from label = do
966 mPid <- gets (^. registeredHereFor label)
967 ncSendToProcess from $ unsafeCreateUnencodedMessage $ WhereIsReply label mPid
969-- [Unified: Table 14]
970ncEffectNamedSend :: String -> Message -> NC ()
971ncEffectNamedSend label msg = do
972 mPid <- gets (^. registeredHereFor label)
973 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
974 forM_ mPid $ \to ->
975 -- If this is a trace message we don't trace it to avoid entering a loop
976 -- where trace messages produce more trace messages.
977 ncSendToProcessAndTrace (label /= "trace.logger") to msg
979-- [Issue #DP-20]
980ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
981ncEffectLocalSend = ncEffectLocalSendAndTrace True
983ncEffectLocalSendAndTrace :: Bool -> LocalNode -> ProcessId -> Message -> NC ()
984ncEffectLocalSendAndTrace shouldTrace node to msg =
985 liftIO $ withLocalProc node to $ \p -> do
986 enqueue (processQueue p) msg
987 when shouldTrace $ trace node (MxReceived to msg)
989-- [Issue #DP-20]
990ncEffectLocalPortSend :: SendPortId -> Message -> NC ()
991ncEffectLocalPortSend from msg = do
992 node <- ask
993 let pid = sendPortProcessId from
994 cid = sendPortLocalId from
995 liftIO $ withLocalProc node pid $ \proc -> do
996 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
997 case mChan of
998 -- in the unlikely event we know nothing about this channel id,
999 -- there's little to be done - perhaps some logging/tracing though...
1000 Nothing -> return ()
1001 Just (TypedChannel chan') -> do
1002 -- If ch is Nothing, the process has given up the read end of
1003 -- the channel and we simply ignore the incoming message - this
1004 ch <- deRefWeak chan'
1005 forM_ ch $ \chan -> deliverChan msg chan
1006 where deliverChan :: forall a . Message -> TQueue a -> IO ()
1007 deliverChan (UnencodedMessage _ raw) chan' =
1008 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1009 deliverChan (EncodedMessage _ _) _ =
1010 -- this will not happen unless someone screws with Primitives.hs
1011 error "invalid local channel delivery"
1013-- [Issue #69]
1014ncEffectKill :: ProcessId -> ProcessId -> String -> NC ()
1015ncEffectKill from to reason = do
1016 node <- ask
1017 when (isLocal node (ProcessIdentifier to)) $
1018 throwException to $ ProcessKillException from reason
1020-- [Issue #69]
1021ncEffectExit :: ProcessId -> ProcessId -> Message -> NC ()
1022ncEffectExit from to reason = do
1023 node <- ask
1024 when (isLocal node (ProcessIdentifier to)) $
1025 throwException to $ ProcessExitException from reason
1027-- [Issue #89]
1028ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()
1029ncEffectGetInfo from pid =
1030 let lpid = processLocalId pid
1031 them = (ProcessIdentifier pid)
1032 in do
1033 node <- ask
1034 mProc <- liftIO $ withValidLocalState node
1035 $ return . (^. localProcessWithId lpid)
1036 case mProc of
1037 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
1038 from (ProcessInfoNone DiedUnknownId)
1039 Just proc -> do
1040 itsLinks <- gets (^. linksFor them)
1041 itsMons <- gets (^. monitorsFor them)
1042 registered <- gets (^. registeredHere)
1043 size <- liftIO $ queueSize $ processQueue $ proc
1045 let reg = registeredNames registered
1046 dispatch (isLocal node (ProcessIdentifier from))
1047 from
1048 ProcessInfo {
1049 infoNode = (processNodeId pid)
1050 , infoRegisteredNames = reg
1051 , infoMessageQueueLength = size
1052 , infoMonitors = Set.toList itsMons
1053 , infoLinks = Set.toList itsLinks
1055 where dispatch :: (Serializable a, Show a)
1056 => Bool
1057 -> ProcessId
1058 -> a
1059 -> NC ()
1060 dispatch True dest pInfo = postAsMessage dest $ pInfo
1061 dispatch False dest pInfo =
1062 ncSendToProcess dest $ unsafeCreateUnencodedMessage pInfo
1064 registeredNames = Map.foldlWithKey (\ks k v -> if v == pid
1065 then (k:ks)
1066 else ks) []
1068ncEffectGetNodeStats :: ProcessId -> NodeId -> NC ()
1069ncEffectGetNodeStats from _nid = do
1070 node <- ask
1071 ncState <- StateT.get
1072 nodeState <- liftIO $ withValidLocalState node return
1073 let stats =
1074 NodeStats {
1075 nodeStatsNode = localNodeId node
1076 , nodeStatsRegisteredNames = Map.size $ ncState ^. registeredHere
1077 , nodeStatsMonitors = Map.size $ ncState ^. monitors
1078 , nodeStatsLinks = Map.size $ ncState ^. links
1079 , nodeStatsProcesses = Map.size (nodeState ^. localProcesses)
1081 postAsMessage from stats
1083--------------------------------------------------------------------------------
1084-- Auxiliary --
1085--------------------------------------------------------------------------------
1087notifyDied :: ProcessId -- ^ Who to notify?
1088 -> Identifier -- ^ Who died?
1089 -> DiedReason -- ^ How did they die?
1090 -> Maybe MonitorRef -- ^ 'Nothing' for linking
1091 -> NC ()
1092notifyDied dest src reason mRef = do
1093 node <- ask
1094 case (isLocal node (ProcessIdentifier dest), mRef, src) of
1095 (True, Just ref, ProcessIdentifier pid) ->
1096 postAsMessage dest $ ProcessMonitorNotification ref pid reason
1097 (True, Just ref, NodeIdentifier nid) ->
1098 postAsMessage dest $ NodeMonitorNotification ref nid reason
1099 (True, Just ref, SendPortIdentifier cid) ->
1100 postAsMessage dest $ PortMonitorNotification ref cid reason
1101 (True, Nothing, ProcessIdentifier pid) ->
1102 throwException dest $ ProcessLinkException pid reason
1103 (True, Nothing, NodeIdentifier pid) ->
1104 throwException dest $ NodeLinkException pid reason
1105 (True, Nothing, SendPortIdentifier pid) ->
1106 throwException dest $ PortLinkException pid reason
1107 (False, _, _) ->
1108 -- The change in sender comes from [Unified: Table 10]
1109 ncSendToNode (processNodeId dest) $ NCMsg
1110 { ctrlMsgSender = NodeIdentifier (localNodeId node)
1111 , ctrlMsgSignal = Died src reason
1114-- | [Unified: Table 8]
1115destNid :: ProcessSignal -> Maybe NodeId
1116destNid (Link ident) = Just $ nodeOf ident
1117destNid (Unlink ident) = Just $ nodeOf ident
1118destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)
1119destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
1120destNid (Spawn _ _) = Nothing
1121destNid (Register _ _ _ _) = Nothing
1122destNid (WhereIs _) = Nothing
1123destNid (NamedSend _ _) = Nothing
1124destNid (UnreliableSend _ _) = Nothing
1125-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
1126-- then when a local process dies the monitoring/linking machinery will take
1127-- care of notifying remote nodes
1128destNid (Died _ _) = Nothing
1129destNid (Kill pid _) = Just $ processNodeId pid
1130destNid (Exit pid _) = Just $ processNodeId pid
1131destNid (GetInfo pid) = Just $ processNodeId pid
1132destNid (GetNodeStats nid) = Just nid
1133destNid (LocalSend pid _) = Just $ processNodeId pid
1134destNid (LocalPortSend cid _) = Just $ processNodeId (sendPortProcessId cid)
1135destNid (SigShutdown) = Nothing
1137-- | Check if a process is local to our own node
1138isLocal :: LocalNode -> Identifier -> Bool
1139isLocal nid ident = nodeOf ident == localNodeId nid
1141-- | Lookup a local closure
1142unClosure :: Typeable a => Closure a -> NC (Either String a)
1143unClosure closure = do
1144 rtable <- remoteTable <$> ask
1145 return (Static.unclosure rtable closure)
1147-- | Check if an identifier refers to a valid local object
1148isValidLocalIdentifier :: Identifier -> NC Bool
1149isValidLocalIdentifier ident = do
1150 node <- ask
1151 liftIO . withValidLocalState node $ \nSt ->
1152 case ident of
1153 NodeIdentifier nid ->
1154 return $ nid == localNodeId node
1155 ProcessIdentifier pid -> do
1156 let mProc = nSt ^. localProcessWithId (processLocalId pid)
1157 return $ isJust mProc
1158 SendPortIdentifier cid -> do
1159 let pid = sendPortProcessId cid
1160 mProc = nSt ^. localProcessWithId (processLocalId pid)
1161 case mProc of
1162 Nothing -> return False
1163 Just proc -> withMVar (processState proc) $ \pSt -> do
1164 let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)
1165 return $ isJust mCh
1167--------------------------------------------------------------------------------
1168-- Messages to local processes --
1169--------------------------------------------------------------------------------
1171postAsMessage :: Serializable a => ProcessId -> a -> NC ()
1172postAsMessage pid = postMessage pid . createUnencodedMessage
1174postMessage :: ProcessId -> Message -> NC ()
1175postMessage pid msg = do
1176 node <- ask
1177 liftIO $ withLocalProc node pid $ \p -> enqueue (processQueue p) msg
1179throwException :: Exception e => ProcessId -> e -> NC ()
1180throwException pid e = do
1181 node <- ask
1182 -- throwTo blocks until the exception is received by the target thread.
1183 -- We cannot easily make it happen asynchronpusly because then 'unlink'
1184 -- semantics would break.
1185 liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1187withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1188withLocalProc node pid p =
1189 -- By [Unified: table 6, rule missing_process] messages to dead processes
1190 -- can silently be dropped
1191 let lpid = processLocalId pid in do
1192 join $ withValidLocalState node $ \vst ->
1193 return $ forM_ (vst ^. localProcessWithId lpid) p
1195--------------------------------------------------------------------------------
1196-- Accessors --
1197--------------------------------------------------------------------------------
1199links :: Accessor NCState (Map Identifier (Set ProcessId))
1200links = accessor _links (\ls st -> st { _links = ls })
1202monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
1203monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1205registeredHere :: Accessor NCState (Map String ProcessId)
1206registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
1208registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1209registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1211linksFor :: Identifier -> Accessor NCState (Set ProcessId)
1212linksFor ident = links >>> DAC.mapDefault Set.empty ident
1214monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
1215monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
1217registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1218registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1220registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
1221registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1223-- | @splitNotif ident@ splits a notifications map into those
1224-- notifications that should trigger when 'ident' fails and those links that
1225-- should not.
1227-- There is a hierarchy between identifiers: failure of a node implies failure
1228-- of all processes on that node, and failure of a process implies failure of
1229-- all typed channels to that process. In other words, if 'ident' refers to a
1230-- node, then the /should trigger/ set will include
1232-- * the notifications for the node specifically
1233-- * the notifications for processes on that node, and
1234-- * the notifications for typed channels to processes on that node.
1236-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
1237-- include
1239-- * the notifications for that process specifically and
1240-- * the notifications for typed channels to that process.
1242-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1243splitNotif :: Identifier
1244 -> Map Identifier a
1245 -> (Map Identifier a, Map Identifier a)
1246splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
1248--------------------------------------------------------------------------------
1249-- Auxiliary --
1250--------------------------------------------------------------------------------
1252-- | Modify and evaluate the state
1253modify' :: MonadState s m => (s -> s) -> m ()
1254modify' f = StateT.get >>= \s -> StateT.put $! f s