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