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