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