git.iko.soy/distributed-process
on 050c72f help sign in
distributed-process/src/Control/Distributed/Process/Node.hs
1262 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, killThread)
74import Control.Distributed.Process.Internal.StrictMVar
75 ( newMVar
76 , withMVar
77 , modifyMVarMasked
78 , modifyMVar
79 , newEmptyMVar
80 , putMVar
81 , takeMVar
82 )
83import Control.Concurrent.Chan (newChan, writeChan, readChan)
84import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)
85import Control.Concurrent.STM
86 ( atomically
87 )
88import Control.Distributed.Process.Internal.CQueue
89 ( CQueue
90 , enqueue
91 , newCQueue
92 , mkWeakCQueue
93 , queueSize
94 )
95import qualified Network.Transport as NT
96 ( Transport
97 , EndPoint
98 , newEndPoint
99 , receive
100 , Event(..)
101 , EventErrorCode(..)
102 , TransportError(..)
103 , address
104 , closeEndPoint
105 , Connection
106 , ConnectionId
107 , close
108 , EndPointAddress
109 , Reliability(ReliableOrdered)
110 )
111import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
112import System.Random (randomIO)
113import Control.Distributed.Static (RemoteTable, Closure)
114import qualified Control.Distributed.Static as Static
115 ( unclosure
116 , initRemoteTable
117 )
118import Control.Distributed.Process.Internal.Types
119 ( NodeId(..)
120 , LocalProcessId(..)
121 , ProcessId(..)
122 , LocalNode(..)
123 , MxEventBus(..)
124 , LocalNodeState(..)
125 , ValidLocalNodeState(..)
126 , withValidLocalState
127 , modifyValidLocalState
128 , LocalProcess(..)
129 , LocalProcessState(..)
130 , Process(..)
131 , DiedReason(..)
132 , NCMsg(..)
133 , ProcessSignal(..)
134 , localPidCounter
135 , localPidUnique
136 , localProcessWithId
137 , localProcesses
138 , localConnections
139 , forever'
140 , MonitorRef(..)
141 , NodeClosedException(..)
142 , ProcessMonitorNotification(..)
143 , NodeMonitorNotification(..)
144 , PortMonitorNotification(..)
145 , ProcessExitException(..)
146 , ProcessLinkException(..)
147 , NodeLinkException(..)
148 , PortLinkException(..)
149 , DidUnmonitor(..)
150 , DidUnlinkProcess(..)
151 , DidUnlinkNode(..)
152 , DidUnlinkPort(..)
153 , SpawnRef
154 , DidSpawn(..)
155 , Message(..)
156 , TypedChannel(..)
157 , Identifier(..)
158 , nodeOf
159 , ProcessInfo(..)
160 , ProcessInfoNone(..)
161 , NodeStats(..)
162 , SendPortId(..)
163 , typedChannelWithId
164 , RegisterReply(..)
165 , WhereIsReply(..)
166 , payloadToMessage
167 , createUnencodedMessage
168 , unsafeCreateUnencodedMessage
169 , runLocalProcess
170 , firstNonReservedProcessId
171 , ImplicitReconnect(WithImplicitReconnect)
172 )
173import Control.Distributed.Process.Management.Internal.Agent
174 ( mxAgentController
175 )
176import qualified Control.Distributed.Process.Management.Internal.Table as Table
177 ( mxTableCoordinator
178 , startTableCoordinator
179 )
180import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace
181 ( remoteTable
182 )
183import Control.Distributed.Process.Management.Internal.Trace.Tracer
184 ( defaultTracer
185 )
186import Control.Distributed.Process.Management.Internal.Trace.Types
187 ( TraceArg(..)
188 , traceEvent
189 , traceLogFmt
190 , enableTrace
191 )
192import Control.Distributed.Process.Management.Internal.Types
193 ( MxEvent(..)
194 )
195import Control.Distributed.Process.Serializable (Serializable)
196import Control.Distributed.Process.Internal.Messaging
197 ( sendBinary
198 , closeImplicitReconnections
199 , impliesDeathOf
200 )
201import Control.Distributed.Process.Internal.Primitives
202 ( register
203 , receiveWait
204 , match
205 , sendChan
206 , unwrapMessage
207 )
208import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))
209import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)
210import Control.Distributed.Process.Internal.WeakTQueue (TQueue, writeTQueue)
211import qualified Control.Distributed.Process.Internal.StrictContainerAccessors as DAC
212 ( mapMaybe
213 , mapDefault
214 )
215import Control.Monad.Catch (try)
216import GHC.IO (IO(..), unsafeUnmask)
217import GHC.Base ( maskAsyncExceptions# )
219import Unsafe.Coerce
220import Prelude
222-- Remove these definitions when the fix for
223-- https://ghc.haskell.org/trac/ghc/ticket/10149
224-- is included in all supported compilers:
225block :: IO a -> IO a
226block (IO io) = IO $ maskAsyncExceptions# io
227unblock :: IO a -> IO a
228unblock = unsafeUnmask
230--------------------------------------------------------------------------------
231-- Initialization --
232--------------------------------------------------------------------------------
234initRemoteTable :: RemoteTable
235initRemoteTable = Trace.remoteTable $ BuiltIn.remoteTable Static.initRemoteTable
237-- | Initialize a new local node.
238newLocalNode :: NT.Transport -> RemoteTable -> IO LocalNode
239newLocalNode transport rtable = do
240 mEndPoint <- NT.newEndPoint transport
241 case mEndPoint of
242 Left ex -> throwIO ex
243 Right endPoint -> do
244 localNode <- createBareLocalNode endPoint rtable
245 startServiceProcesses localNode
246 return localNode
248-- | Create a new local node (without any service processes running)
249createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode
250createBareLocalNode endPoint rtable = do
251 unq <- randomIO
252 state <- newMVar $ LocalNodeValid $ ValidLocalNodeState
253 { _localProcesses = Map.empty
254 , _localPidCounter = firstNonReservedProcessId
255 , _localPidUnique = unq
256 , _localConnections = Map.empty
257 }
258 ctrlChan <- newChan
259 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
260 , localEndPoint = endPoint
261 , localState = state
262 , localCtrlChan = ctrlChan
263 , localEventBus = MxEventBusInitialising
264 , remoteTable = rtable
265 }
266 tracedNode <- startMxAgent node
268 -- Once the NC terminates, the endpoint isn't much use,
269 void $ forkIO $ Exception.finally (runNodeController tracedNode)
270 (NT.closeEndPoint (localEndPoint node))
272 -- whilst a closed/failing endpoint will terminate the NC
273 void $ forkIO $ Exception.finally (handleIncomingMessages tracedNode)
274 (stopNC node)
276 return tracedNode
277 where
278 stopNC node =
279 writeChan (localCtrlChan node) NCMsg
280 { ctrlMsgSender = NodeIdentifier (localNodeId node)
281 , ctrlMsgSignal = SigShutdown
282 }
284startMxAgent :: LocalNode -> IO LocalNode
285startMxAgent node = do
286 -- see note [tracer/forkProcess races]
287 let fork = forkProcess node
288 mv <- MVar.newEmptyMVar
289 pid <- fork $ mxAgentController fork mv
290 (tracer', wqRef, mxNew') <- MVar.takeMVar mv
291 return node { localEventBus = (MxEventBus pid tracer' wqRef mxNew') }
293startDefaultTracer :: LocalNode -> IO ()
294startDefaultTracer node' = do
295 let t = localEventBus node'
296 case t of
297 MxEventBus _ (Tracer pid _) _ _ -> do
298 runProcess node' $ register "trace.controller" pid
299 pid' <- forkProcess node' defaultTracer
300 enableTrace (localEventBus node') pid'
301 runProcess node' $ register "tracer.initial" pid'
302 _ -> return ()
304-- TODO: we need a better mechanism for defining and registering services
306-- | Start and register the service processes on a node
307startServiceProcesses :: LocalNode -> IO ()
308startServiceProcesses node = do
309 -- tracing /spawns/ relies on the tracer being enabled, but we start
310 -- the default tracer first, even though it might @nsend@ to the logger
311 -- before /that/ process has started - this is a totally harmless race
312 -- however, so we deliberably ignore it
313 startDefaultTracer node
314 tableCoordinatorPid <- fork $ Table.startTableCoordinator fork
315 runProcess node $ register Table.mxTableCoordinator tableCoordinatorPid
316 logger <- forkProcess node loop
317 runProcess node $ do
318 register "logger" logger
319 -- The trace.logger is used for tracing to the console to avoid feedback
320 -- loops during tracing if the user reregisters the "logger" with a custom
321 -- process which uses 'send' or other primitives which are traced.
322 register "trace.logger" logger
323 where
324 fork = forkProcess node
326 loop = do
327 receiveWait
328 [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do
329 liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string
330 loop
331 , match $ \((time, string) :: (String, String)) -> do
332 -- this is a 'trace' message from the local node tracer
333 liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string
334 loop
335 , match $ \(ch :: SendPort ()) -> -- a shutdown request
336 sendChan ch ()
337 ]
339-- | Force-close a local node, killing all processes on that node.
340closeLocalNode :: LocalNode -> IO ()
341closeLocalNode node = do
342 -- Kill processes after refilling the mvar. Otherwise, there is potential for
343 -- deadlock as a dying process tries to get the mvar while masking exceptions
344 -- uninterruptibly.
345 join $ modifyMVar (localState node) $ \st -> case st of
346 LocalNodeValid vst -> do
347 return ( LocalNodeClosed
348 , forM_ (vst ^. localProcesses) $ \lproc ->
349 -- Semantics of 'throwTo' guarantee that target thread will get
350 -- delivered an exception. Therefore, target thread will be
351 -- killed eventually and that's as good as we can do. No need
352 -- to wait for thread to actually finish dying.
353 killThread (processThread lproc)
354 )
355 LocalNodeClosed -> return (LocalNodeClosed, return ())
356 -- This call will have the effect of shutting down the NC as well (see
357 -- 'createBareLocalNode').
358 NT.closeEndPoint (localEndPoint node)
360-- | Run a process on a local node and wait for it to finish
361runProcess :: LocalNode -> Process () -> IO ()
362runProcess node proc = do
363 done <- newEmptyMVar
364 -- TODO; When forkProcess inherits the masking state, protect the forked
365 -- thread against async exceptions that could occur before 'try' is evaluated.
366 void $ forkProcess node $ try proc >>= liftIO . putMVar done
367 takeMVar done >>= either (throwIO :: SomeException -> IO a) return
369-- | Spawn a new process on a local node
370forkProcess :: LocalNode -> Process () -> IO ProcessId
371forkProcess node proc = do
372 ms <- getMaskingState
373 modifyMVarMasked (localState node) (startProcess ms)
374 where
375 startProcess :: MaskingState
376 -> LocalNodeState
377 -> IO (LocalNodeState, ProcessId)
378 startProcess ms (LocalNodeValid vst) = do
379 let lpid = LocalProcessId { lpidCounter = vst ^. localPidCounter
380 , lpidUnique = vst ^. localPidUnique
381 }
382 let pid = ProcessId { processNodeId = localNodeId node
383 , processLocalId = lpid
384 }
385 pst <- newMVar LocalProcessState { _monitorCounter = 0
386 , _spawnCounter = 0
387 , _channelCounter = 0
388 , _typedChannels = Map.empty
389 }
390 queue <- newCQueue
391 weakQueue <- mkWeakCQueue queue (return ())
392 (_, lproc) <- fixIO $ \ ~(tid, _) -> do
393 let lproc = LocalProcess { processQueue = queue
394 , processWeakQ = weakQueue
395 , processId = pid
396 , processState = pst
397 , processThread = tid
398 , processNode = node
399 }
400 -- Rewrite this code when this is fixed:
401 -- https://ghc.haskell.org/trac/ghc/ticket/10149
402 let unmask = case ms of
403 Unmasked -> unblock
404 MaskedInterruptible -> block
405 MaskedUninterruptible -> id
406 tid' <- uninterruptibleMask_ $ forkIO $ do
407 reason <- Exception.catches
408 (unmask $ runLocalProcess lproc proc >> return DiedNormal)
409 [ (Exception.Handler (\ex@(ProcessExitException from msg) -> do
410 mMsg <- unwrapMessage msg :: IO (Maybe String)
411 case mMsg of
412 Nothing -> return $ DiedException $ show ex
413 Just m -> return $ DiedException ("exit-from=" ++ (show from) ++ ",reason=" ++ m)))
414 , (Exception.Handler
415 (return . DiedException . (show :: SomeException -> String)))]
417 -- [Unified: Table 4, rules termination and exiting]
418 mconns <- modifyValidLocalState node (cleanupProcess pid)
419 -- XXX: Revisit after agreeing on the bigger picture for the semantics
420 -- of transport operations.
421 -- https://github.com/haskell-distributed/distributed-process/issues/204
422 forM_ mconns $ forkIO . mapM_ NT.close
424 writeChan (localCtrlChan node) NCMsg
425 { ctrlMsgSender = ProcessIdentifier pid
426 , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
427 }
428 return (tid', lproc)
430 -- see note [tracer/forkProcess races]
431 trace node (MxSpawned pid)
433 if lpidCounter lpid == maxBound
434 then do
435 -- TODO: this doesn't look right at all - how do we know
436 -- that newUnique represents a process id that is available!?
437 newUnique <- randomIO
438 return ( LocalNodeValid
439 $ (localProcessWithId lpid ^= Just lproc)
440 . (localPidCounter ^= firstNonReservedProcessId)
441 . (localPidUnique ^= newUnique)
442 $ vst
443 , pid
444 )
445 else
446 return ( LocalNodeValid
447 $ (localProcessWithId lpid ^= Just lproc)
448 . (localPidCounter ^: (+ 1))
449 $ vst
450 , pid
451 )
452 startProcess _ LocalNodeClosed =
453 throwIO $ NodeClosedException $ localNodeId node
455 cleanupProcess :: ProcessId
456 -> ValidLocalNodeState
457 -> IO (ValidLocalNodeState, [NT.Connection])
458 cleanupProcess pid vst = do
459 let pid' = ProcessIdentifier pid
460 let (affected, unaffected) = Map.partitionWithKey (\(fr, _to) !_v -> impliesDeathOf pid' fr) (vst ^. localConnections)
461 return ( (localProcessWithId (processLocalId pid) ^= Nothing)
462 . (localConnections ^= unaffected)
463 $ vst
464 , map fst $ Map.elems affected
465 )
467-- note [tracer/forkProcess races]
468--
469-- Our startTracing function uses forkProcess to start the trace controller
470-- process, and of course forkProcess attempts to call traceEvent once the
471-- process has started. This is harmless, as the localEventBus is not updated
472-- until /after/ the initial forkProcess completes, so the first call to
473-- traceEvent behaves as if tracing were disabled (i.e., it is ignored).
474--
476--------------------------------------------------------------------------------
477-- Handle incoming messages --
478--------------------------------------------------------------------------------
480type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
482data IncomingTarget =
483 Uninit
484 | ToProc ProcessId (Weak (CQueue Message))
485 | ToChan TypedChannel
486 | ToNode
488data ConnectionState = ConnectionState {
489 _incoming :: !(Map NT.ConnectionId IncomingConnection)
490 , _incomingFrom :: !(Map NT.EndPointAddress (Set NT.ConnectionId))
491 }
493initConnectionState :: ConnectionState
494initConnectionState = ConnectionState {
495 _incoming = Map.empty
496 , _incomingFrom = Map.empty
497 }
499incoming :: Accessor ConnectionState (Map NT.ConnectionId IncomingConnection)
500incoming = accessor _incoming (\conns st -> st { _incoming = conns })
502incomingAt :: NT.ConnectionId -> Accessor ConnectionState (Maybe IncomingConnection)
503incomingAt cid = incoming >>> DAC.mapMaybe cid
505incomingFrom :: NT.EndPointAddress -> Accessor ConnectionState (Set NT.ConnectionId)
506incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
507 where
508 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
510handleIncomingMessages :: LocalNode -> IO ()
511handleIncomingMessages node = go initConnectionState
512 `Exception.catch` \(NodeClosedException _) -> return ()
513 where
514 go :: ConnectionState -> IO ()
515 go !st = do
516 event <- NT.receive endpoint
517 case event of
518 NT.ConnectionOpened cid rel theirAddr ->
519 if rel == NT.ReliableOrdered
520 then
521 trace node (MxConnected cid theirAddr)
522 >> go (
523 (incomingAt cid ^= Just (theirAddr, Uninit))
524 . (incomingFrom theirAddr ^: Set.insert cid)
525 $ st
526 )
527 else invalidRequest cid st $
528 "attempt to connect with unsupported reliability " ++ show rel
529 NT.Received cid payload ->
530 case st ^. incomingAt cid of
531 Just (_, ToProc pid weakQueue) -> do
532 mQueue <- deRefWeak weakQueue
533 forM_ mQueue $ \queue -> do
534 -- TODO: if we find that the queue is Nothing, should we remove
535 -- it from the NC state? (and same for channels, below)
536 let msg = payloadToMessage payload
537 enqueue queue msg -- 'enqueue' is strict
538 trace node (MxReceived pid msg)
539 go st
540 Just (_, ToChan (TypedChannel chan')) -> do
541 mChan <- deRefWeak chan'
542 -- If mChan is Nothing, the process has given up the read end of
543 -- the channel and we simply ignore the incoming message
544 forM_ mChan $ \chan -> atomically $
545 -- We make sure the message is fully decoded when it is enqueued
546 writeTQueue chan $! decode (BSL.fromChunks payload)
547 go st
548 Just (_, ToNode) -> do
549 let ctrlMsg = decode . BSL.fromChunks $ payload
550 writeChan ctrlChan $! ctrlMsg
551 go st
552 Just (src, Uninit) ->
553 case decode (BSL.fromChunks payload) of
554 ProcessIdentifier pid -> do
555 let lpid = processLocalId pid
556 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
557 case mProc of
558 Just proc ->
559 go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
560 Nothing ->
561 -- incoming attempt to connect to unknown process - might
562 -- be dead already
563 go (incomingAt cid ^= Nothing $ st)
564 SendPortIdentifier chId -> do
565 let lcid = sendPortLocalId chId
566 lpid = processLocalId (sendPortProcessId chId)
567 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
568 case mProc of
569 Just proc -> do
570 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
571 case mChannel of
572 Just channel ->
573 go (incomingAt cid ^= Just (src, ToChan channel) $ st)
574 Nothing ->
575 invalidRequest cid st $
576 "incoming attempt to connect to unknown channel of"
577 ++ " process " ++ show (sendPortProcessId chId)
578 Nothing ->
579 -- incoming attempt to connect to channel of unknown
580 -- process - might be dead already
581 go (incomingAt cid ^= Nothing $ st)
582 NodeIdentifier nid ->
583 if nid == localNodeId node
584 then go (incomingAt cid ^= Just (src, ToNode) $ st)
585 else invalidRequest cid st $
586 "incoming attempt to connect to a different node -"
587 ++ " I'm " ++ show (localNodeId node)
588 ++ " but the remote peer wants to connect to "
589 ++ show nid
590 Nothing ->
591 invalidRequest cid st
592 "message received from an unknown connection"
593 NT.ConnectionClosed cid ->
594 case st ^. incomingAt cid of
595 Nothing ->
596 invalidRequest cid st "closed unknown connection"
597 Just (src, _) -> do
598 trace node (MxDisconnected cid src)
599 go ( (incomingAt cid ^= Nothing)
600 . (incomingFrom src ^: Set.delete cid)
601 $ st
602 )
603 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr) _) -> do
604 -- [Unified table 9, rule node_disconnect]
605 let nid = NodeIdentifier $ NodeId theirAddr
606 writeChan ctrlChan NCMsg
607 { ctrlMsgSender = nid
608 , ctrlMsgSignal = Died nid DiedDisconnect
609 }
610 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
611 closeImplicitReconnections node nid
612 go ( (incomingFrom theirAddr ^= Set.empty)
613 . (incoming ^: Map.filterWithKey (const . notLost))
614 $ st
615 )
616 NT.ErrorEvent (NT.TransportError NT.EventEndPointFailed str) ->
617 fail $ "Cloud Haskell fatal error: end point failed: " ++ str
618 NT.ErrorEvent (NT.TransportError NT.EventTransportFailed str) ->
619 fail $ "Cloud Haskell fatal error: transport failed: " ++ str
620 NT.EndPointClosed ->
621 return ()
622 NT.ReceivedMulticast _ _ ->
623 -- If we received a multicast message, something went horribly wrong
624 -- and we just give up
625 fail "Cloud Haskell fatal error: received unexpected multicast"
627 invalidRequest :: NT.ConnectionId -> ConnectionState -> String -> IO ()
628 invalidRequest cid st msg = do
629 -- TODO: We should treat this as a fatal error on the part of the remote
630 -- node. That is, we should report the remote node as having died, and we
631 -- should close incoming connections (this requires a Transport layer
632 -- extension).
633 traceEventFmtIO node "" [ TraceStr $ " [network] invalid request"
634 ++ " (" ++ msg ++ "): "
635 , (Trace cid)
636 ]
637 go ( incomingAt cid ^= Nothing
638 $ st
639 )
641 endpoint = localEndPoint node
642 ctrlChan = localCtrlChan node
644--------------------------------------------------------------------------------
645-- Top-level access to the node controller --
646--------------------------------------------------------------------------------
648runNodeController :: LocalNode -> IO ()
649runNodeController node =
650 runReaderT (evalStateT (unNC nodeController) initNCState) node
651 `Exception.catch` \(NodeClosedException _) -> return ()
653--------------------------------------------------------------------------------
654-- Internal data types --
655--------------------------------------------------------------------------------
657data NCState = NCState
658 { -- Mapping from remote processes to linked local processes
659 _links :: !(Map Identifier (Set ProcessId))
660 -- Mapping from remote processes to monitoring local processes
661 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
662 -- Process registry: names and where they live, mapped to the PIDs
663 , _registeredHere :: !(Map String ProcessId)
664 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
665 }
667newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
668 deriving ( Applicative
669 , Functor
670 , Monad
671 , MonadIO
672 , MonadState NCState
673 , MonadReader LocalNode
674 )
676initNCState :: NCState
677initNCState = NCState { _links = Map.empty
678 , _monitors = Map.empty
679 , _registeredHere = Map.empty
680 , _registeredOnNodes = Map.empty
681 }
683-- | Thrown in response to the user invoking 'kill' (see Primitives.hs). This
684-- type is deliberately not exported so it cannot be caught explicitly.
685data ProcessKillException =
686 ProcessKillException !ProcessId !String
687 deriving (Typeable)
689instance Exception ProcessKillException
690instance Show ProcessKillException where
691 show (ProcessKillException pid reason) =
692 "killed-by=" ++ show pid ++ ",reason=" ++ reason
694ncSendToProcess :: ProcessId -> Message -> NC ()
695ncSendToProcess = ncSendToProcessAndTrace True
697ncSendToProcessAndTrace :: Bool -> ProcessId -> Message -> NC ()
698ncSendToProcessAndTrace shouldTrace pid msg = do
699 node <- ask
700 if processNodeId pid == localNodeId node
701 then ncEffectLocalSendAndTrace shouldTrace node pid msg
702 else liftIO $ sendBinary node
703 (NodeIdentifier $ localNodeId node)
704 (NodeIdentifier $ processNodeId pid)
705 WithImplicitReconnect
706 NCMsg { ctrlMsgSender = NodeIdentifier $ localNodeId node
707 , ctrlMsgSignal = UnreliableSend (processLocalId pid) msg
708 }
710ncSendToNode :: NodeId -> NCMsg -> NC ()
711ncSendToNode to msg = do
712 node <- ask
713 liftIO $ if to == localNodeId node
714 then writeChan (localCtrlChan node) $! msg
715 else sendBinary node
716 (NodeIdentifier $ localNodeId node)
717 (NodeIdentifier to)
718 WithImplicitReconnect
719 msg
721--------------------------------------------------------------------------------
722-- Tracing/Debugging --
723--------------------------------------------------------------------------------
725-- [Issue #104 / DP-13]
727traceNotifyDied :: LocalNode -> Identifier -> DiedReason -> NC ()
728traceNotifyDied node ident reason =
729 -- TODO: sendPortDied notifications
730 liftIO $ withLocalTracer node $ \t ->
731 case ident of
732 (NodeIdentifier nid) -> traceEvent t (MxNodeDied nid reason)
733 (ProcessIdentifier pid) -> traceEvent t (MxProcessDied pid reason)
734 _ -> return ()
736traceEventFmtIO :: LocalNode
737 -> String
738 -> [TraceArg]
739 -> IO ()
740traceEventFmtIO node fmt args =
741 withLocalTracer node $ \t -> traceLogFmt t fmt args
743trace :: LocalNode -> MxEvent -> IO ()
744trace node ev = withLocalTracer node $ \t -> traceEvent t ev
746withLocalTracer :: LocalNode -> (MxEventBus -> IO ()) -> IO ()
747withLocalTracer node act = act (localEventBus node)
749--------------------------------------------------------------------------------
750-- Core functionality --
751--------------------------------------------------------------------------------
753-- [Unified: Table 7]
754nodeController :: NC ()
755nodeController = do
756 node <- ask
757 forever' $ do
758 msg <- liftIO $ readChan (localCtrlChan node)
760 -- [Unified: Table 7, rule nc_forward]
761 case destNid (ctrlMsgSignal msg) of
762 Just nid' | nid' /= localNodeId node ->
763 ncSendToNode nid' msg
764 _ ->
765 return ()
767 case msg of
768 NCMsg (ProcessIdentifier from) (Link them) ->
769 ncEffectMonitor from them Nothing
770 NCMsg (ProcessIdentifier from) (Monitor ref) ->
771 ncEffectMonitor from (monitorRefIdent ref) (Just ref)
772 NCMsg (ProcessIdentifier from) (Unlink them) ->
773 ncEffectUnlink from them
774 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->
775 ncEffectUnmonitor from ref
776 NCMsg _from (Died ident reason) ->
777 ncEffectDied ident reason
778 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
779 ncEffectSpawn from proc ref
780 NCMsg (ProcessIdentifier from) (Register label atnode pid force) ->
781 ncEffectRegister from label atnode pid force
782 NCMsg (ProcessIdentifier from) (WhereIs label) ->
783 ncEffectWhereIs from label
784 NCMsg _ (NamedSend label msg') ->
785 ncEffectNamedSend label msg'
786 NCMsg _ (UnreliableSend lpid msg') ->
787 ncEffectLocalSend node (ProcessId (localNodeId node) lpid) msg'
788 NCMsg _ (LocalSend to msg') ->
789 ncEffectLocalSend node to msg'
790 NCMsg _ (LocalPortSend to msg') ->
791 ncEffectLocalPortSend to msg'
792 NCMsg (ProcessIdentifier from) (Kill to reason) ->
793 ncEffectKill from to reason
794 NCMsg (ProcessIdentifier from) (Exit to reason) ->
795 ncEffectExit from to reason
796 NCMsg (ProcessIdentifier from) (GetInfo pid) ->
797 ncEffectGetInfo from pid
798 NCMsg _ SigShutdown ->
799 liftIO $ do
800 NT.closeEndPoint (localEndPoint node)
801 `Exception.finally` throwIO (NodeClosedException $ localNodeId node)
802 NCMsg (ProcessIdentifier from) (GetNodeStats nid) ->
803 ncEffectGetNodeStats from nid
804 unexpected ->
805 error $ "nodeController: unexpected message " ++ show unexpected
807-- [Unified: Table 10]
808ncEffectMonitor :: ProcessId -- ^ Who's watching?
809 -> Identifier -- ^ Who's being watched?
810 -> Maybe MonitorRef -- ^ 'Nothing' to link
811 -> NC ()
812ncEffectMonitor from them mRef = do
813 node <- ask
814 shouldLink <-
815 if not (isLocal node them)
816 then return True
817 else isValidLocalIdentifier them
818 case (shouldLink, isLocal node (ProcessIdentifier from)) of
819 (True, _) -> -- [Unified: first rule]
820 case mRef of
821 Just ref -> modify' $ monitorsFor them ^: Set.insert (from, ref)
822 Nothing -> modify' $ linksFor them ^: Set.insert from
823 (False, True) -> -- [Unified: second rule]
824 notifyDied from them DiedUnknownId mRef
825 (False, False) -> -- [Unified: third rule]
826 -- TODO: this is the right sender according to the Unified semantics,
827 -- but perhaps having 'them' as the sender would make more sense
828 -- (see also: notifyDied)
829 ncSendToNode (processNodeId from) $ NCMsg
830 { ctrlMsgSender = NodeIdentifier (localNodeId node)
831 , ctrlMsgSignal = Died them DiedUnknownId
832 }
834-- [Unified: Table 11]
835ncEffectUnlink :: ProcessId -> Identifier -> NC ()
836ncEffectUnlink from them = do
837 node <- ask
838 when (isLocal node (ProcessIdentifier from)) $
839 case them of
840 ProcessIdentifier pid ->
841 postAsMessage from $ DidUnlinkProcess pid
842 NodeIdentifier nid ->
843 postAsMessage from $ DidUnlinkNode nid
844 SendPortIdentifier cid ->
845 postAsMessage from $ DidUnlinkPort cid
846 modify' $ linksFor them ^: Set.delete from
848-- [Unified: Table 11]
849ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
850ncEffectUnmonitor from ref = do
851 node <- ask
852 when (isLocal node (ProcessIdentifier from)) $
853 postAsMessage from $ DidUnmonitor ref
854 modify' $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
856-- [Unified: Table 12]
857ncEffectDied :: Identifier -> DiedReason -> NC ()
858ncEffectDied ident reason = do
859 node <- ask
860 traceNotifyDied node ident reason
861 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
862 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
864-- _registry :: !(Map (String,NodeId) ProcessId)
866 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
868 forM_ (Map.toList affectedLinks) $ \(them, uss) ->
869 forM_ uss $ \us ->
870 when (localOnly <= isLocal node (ProcessIdentifier us)) $
871 notifyDied us them reason Nothing
873 forM_ (Map.toList affectedMons) $ \(them, refs) ->
874 forM_ refs $ \(us, ref) ->
875 when (localOnly <= isLocal node (ProcessIdentifier us)) $
876 notifyDied us them reason (Just ref)
878 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
880 modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
882 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
883 mapM (\(pid,nidlist) ->
884 case ident `impliesDeathOf` ProcessIdentifier pid of
885 True ->
886 do forM_ nidlist $ \(nid,_) ->
887 when (not $ isLocal node (NodeIdentifier nid))
888 (forwardNameDeath node nid)
889 return Nothing
890 False -> return $ Just (pid,nidlist) )
891 modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
892 where
893 forwardNameDeath node nid = ncSendToNode nid
894 NCMsg { ctrlMsgSender = NodeIdentifier (localNodeId node)
895 , ctrlMsgSignal = Died ident reason
896 }
898-- [Unified: Table 13]
899ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
900ncEffectSpawn pid cProc ref = do
901 mProc <- unClosure cProc
902 -- If the closure does not exist, we spawn a process that throws an exception
903 -- This allows the remote node to find out what's happening
904 -- TODO:
905 let proc = case mProc of
906 Left err -> fail $ "Error: Could not resolve closure: " ++ err
907 Right p -> p
908 node <- ask
909 pid' <- liftIO $ forkProcess node proc
910 ncSendToProcess pid $ unsafeCreateUnencodedMessage $ DidSpawn ref pid'
912-- Unified semantics does not explicitly describe how to implement 'register',
913-- but mentions it's "very similar to nsend" (Table 14)
914-- We send a response indicated if the operation is invalid
915ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
916ncEffectRegister from label atnode mPid reregistration = do
917 node <- ask
918 currentVal <- gets (^. registeredHereFor label)
919 isOk <-
920 case mPid of
921 Nothing -> -- unregister request
922 return $ isJust currentVal
923 Just thepid -> -- register request
924 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
925 return $ (isNothing currentVal /= reregistration) &&
926 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
927 if isLocal node (NodeIdentifier atnode)
928 then do when isOk $
929 do modify' $ registeredHereFor label ^= mPid
930 updateRemote node currentVal mPid
931 case mPid of
932 (Just p) -> liftIO $ trace node (MxRegistered p label)
933 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)
934 newVal <- gets (^. registeredHereFor label)
935 ncSendToProcess from $ unsafeCreateUnencodedMessage $
936 RegisterReply label isOk newVal
937 else let operation =
938 case reregistration of
939 True -> flip decList
940 False -> flip incList
941 in case mPid of
942 Nothing -> return ()
943 Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
944 where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
945 do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
946 forward node (processNodeId newval) (Register label atnode (Just newval) False)
947 updateRemote node Nothing (Just newval) =
948 forward node (processNodeId newval) (Register label atnode (Just newval) False)
949 updateRemote node (Just oldval) Nothing =
950 forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
951 updateRemote _ _ _ = return ()
952 maybeify f Nothing = unmaybeify $ f []
953 maybeify f (Just x) = unmaybeify $ f x
955 unmaybeify [] = Nothing
956 unmaybeify x = Just x
957 incList [] tag = [(tag,1)]
958 incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
959 incList (x:xs) tag = x : incList xs tag
960 decList [] _ = []
961 decList ((atag,1):xs) tag | atag == tag = xs
962 decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
963 decList (x:xs) tag = x:decList xs tag
964 forward node to reg =
965 when (not $ isLocal node (NodeIdentifier to)) $
966 ncSendToNode to $ NCMsg { ctrlMsgSender = ProcessIdentifier from
967 , ctrlMsgSignal = reg
968 }
971-- Unified semantics does not explicitly describe 'whereis'
972ncEffectWhereIs :: ProcessId -> String -> NC ()
973ncEffectWhereIs from label = do
974 mPid <- gets (^. registeredHereFor label)
975 ncSendToProcess from $ unsafeCreateUnencodedMessage $ WhereIsReply label mPid
977-- [Unified: Table 14]
978ncEffectNamedSend :: String -> Message -> NC ()
979ncEffectNamedSend label msg = do
980 mPid <- gets (^. registeredHereFor label)
981 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
982 forM_ mPid $ \to ->
983 -- If this is a trace message we don't trace it to avoid entering a loop
984 -- where trace messages produce more trace messages.
985 ncSendToProcessAndTrace (label /= "trace.logger") to msg
987-- [Issue #DP-20]
988ncEffectLocalSend :: LocalNode -> ProcessId -> Message -> NC ()
989ncEffectLocalSend = ncEffectLocalSendAndTrace True
991ncEffectLocalSendAndTrace :: Bool -> LocalNode -> ProcessId -> Message -> NC ()
992ncEffectLocalSendAndTrace shouldTrace node to msg =
993 liftIO $ withLocalProc node to $ \p -> do
994 enqueue (processQueue p) msg
995 when shouldTrace $ 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)
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 cannot easily make it happen asynchronpusly because then 'unlink'
1192 -- semantics would break.
1193 liftIO $ withLocalProc node pid $ \p -> throwTo (processThread p) e
1195withLocalProc :: LocalNode -> ProcessId -> (LocalProcess -> IO ()) -> IO ()
1196withLocalProc node pid p =
1197 -- By [Unified: table 6, rule missing_process] messages to dead processes
1198 -- can silently be dropped
1199 let lpid = processLocalId pid in do
1200 join $ withValidLocalState node $ \vst ->
1201 return $ forM_ (vst ^. localProcessWithId lpid) p
1203--------------------------------------------------------------------------------
1204-- Accessors --
1205--------------------------------------------------------------------------------
1207links :: Accessor NCState (Map Identifier (Set ProcessId))
1208links = accessor _links (\ls st -> st { _links = ls })
1210monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
1211monitors = accessor _monitors (\ms st -> st { _monitors = ms })
1213registeredHere :: Accessor NCState (Map String ProcessId)
1214registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
1216registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
1217registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
1219linksFor :: Identifier -> Accessor NCState (Set ProcessId)
1220linksFor ident = links >>> DAC.mapDefault Set.empty ident
1222monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
1223monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
1225registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
1226registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
1228registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
1229registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
1231-- | @splitNotif ident@ splits a notifications map into those
1232-- notifications that should trigger when 'ident' fails and those links that
1233-- should not.
1235-- There is a hierarchy between identifiers: failure of a node implies failure
1236-- of all processes on that node, and failure of a process implies failure of
1237-- all typed channels to that process. In other words, if 'ident' refers to a
1238-- node, then the /should trigger/ set will include
1240-- * the notifications for the node specifically
1241-- * the notifications for processes on that node, and
1242-- * the notifications for typed channels to processes on that node.
1244-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
1245-- include
1247-- * the notifications for that process specifically and
1248-- * the notifications for typed channels to that process.
1250-- See https://github.com/haskell/containers/issues/14 for the bang on _v.
1251splitNotif :: Identifier
1252 -> Map Identifier a
1253 -> (Map Identifier a, Map Identifier a)
1254splitNotif ident = Map.partitionWithKey (\k !_v -> ident `impliesDeathOf` k)
1256--------------------------------------------------------------------------------
1257-- Auxiliary --
1258--------------------------------------------------------------------------------
1260-- | Modify and evaluate the state
1261modify' :: MonadState s m => (s -> s) -> m ()
1262modify' f = StateT.get >>= \s -> StateT.put $! f s