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