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