@@ -17,16 +17,18 @@
1717--
1818-- In order to guarantee (3), we stipulate that
1919--
20−-- 3a. Once a connection to a remote process fails, that process is considered
21−-- forever unreachable. When the remote process restarts, it will receive a
22−-- brand new ProcessId.
20+-- 3a. We do not garbage collect connections because Network.Transport provides
21+-- ordering guarantees only *per connection*.
2322--
24−-- 3b. We do not garbage collect (lightweight) connections, because we have
25−-- ordering guarantees from Network.Transport only per lightweight
26−-- connection. We could lift this restriction later, if we wish, by adding
27−-- some acknowledgement control messages: we can drop one lightweight
28−-- connection and open another once we know that all messages sent on the
29−-- former have been received.
23+-- 3b. Once a connection breaks, we have no way of knowing which messages
24+-- arrived and which did not; hence, once a connection fails, we assume the
25+-- remote process to be forever unreachable. Otherwise we might sent m1 and
26+-- m2, get notified of the broken connection, reconnect, send m3, but only
27+-- m1 and m3 arrive.
28+--
29+-- 3c. As a consequence of (3b) we should not reuse PIDs. If a process dies,
30+-- we consider it forever unreachable. Hence, new processes should get new
31+-- IDs or they too would be considered unreachable.
3032--
3133-- Main reference for Cloud Haskell is
3234--
@@ -70,13 +72,10 @@ module Control.Distributed.Process
7072 , expect
7173 , send
7274 , getSelfPid
75+ -- * Monitoring and linking
7376 , monitor
74− , monitorProcess
75− , linkProcess
76− -- * Monitoring
77− , MonitorAction(..)
78− , ProcessMonitorException(..)
79− , SignalReason(..)
77+ , MonitorReply(..)
78+ , DiedReason(..)
8079 -- * Initialization
8180 , newLocalNode
8281 , forkProcess
@@ -94,20 +93,23 @@ import qualified Data.ByteString.Lazy as BSL ( ByteString
9493 , fromChunks
9594 , splitAt
9695 )
97−import Data.Binary (Binary, decode, encode, put, get, getWord8, putWord8)
96+import Data.Binary (Binary, decode, encode, put, get, putWord8, getWord8)
9897import Data.Map (Map)
99−import qualified Data.Map as Map (empty, lookup, insert, delete, keys)
98+import qualified Data.Map as Map (empty, lookup, insert, delete, keys, toList)
10099import qualified Data.List as List (delete)
100+import Data.Set (Set)
101+import qualified Data.Set as Set (empty, insert)
101102import Data.Int (Int32)
102103import Data.Typeable (Typeable)
103104import Data.Foldable (forM_)
104−import Control.Monad (void, liftM, liftM2)
105+import Data.Maybe (isNothing)
106+import Control.Monad (void, liftM2, liftM3, when)
105107import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
106108import Control.Monad.IO.Class (MonadIO, liftIO)
107109import Control.Applicative ((<$>))
108110import Control.Category ((>>>))
109−import Control.Exception (Exception, throwIO, catch, throwTo)
110−import Control.Concurrent (forkIO, myThreadId)
111+import Control.Exception (Exception, throwIO, catch)
112+import Control.Concurrent (forkIO)
111113import Control.Concurrent.MVar ( MVar
112114 , newMVar
113115 , withMVar
@@ -117,6 +119,7 @@ import Control.Concurrent.MVar ( MVar
117119 , putMVar
118120 , takeMVar
119121 )
122+import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
120123import Control.Distributed.Process.Internal.CQueue ( CQueue
121124 , dequeueMatching
122125 , enqueue
@@ -145,6 +148,7 @@ import qualified Network.Transport as NT ( Transport
145148 , TransportError(..)
146149 , address
147150 , closeEndPoint
151+ , ConnectionId
148152 )
149153import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)
150154import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
@@ -159,10 +163,14 @@ data LocalProcessId = LocalProcessId
159163 }
160164 deriving (Eq, Ord, Typeable, Show)
161165
166+-- We identify node IDs and endpoint IDs
167+newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress }
168+ deriving (Show, Eq, Ord, Binary)
169+
162170-- | A process ID combines a local process with with an endpoint address
163171-- (in other words, we identify nodes and endpoints)
164172data ProcessId = ProcessId
165− { processAddress :: NT.EndPointAddress
173+ { processNodeId :: NodeId
166174 , processLocalId :: LocalProcessId
167175 }
168176 deriving (Eq, Ord, Typeable, Show)
@@ -175,16 +183,19 @@ data Message = Message
175183
176184-- | Local nodes
177185data LocalNode = LocalNode
178− { localEndPoint :: NT.EndPoint
179− , localState :: MVar LocalNodeState
186+ { localNodeId :: NodeId
187+ , localEndPoint :: NT.EndPoint
188+ , localState :: MVar LocalNodeState
189+ , localCtrlChan :: Chan NodeCtrlMsg
180190 }
181191
182192-- | Local node state
183193data LocalNodeState = LocalNodeState
184− { _localProcesses :: Map LocalProcessId LocalProcess
185− , _localPidCounter :: Int32
186− , _localPidUnique :: Int32
187− , _monitorActions :: Map ProcessId [IO ()]
194+ { _localProcesses :: Map LocalProcessId LocalProcess
195+ , _localPidCounter :: Int32
196+ , _localPidUnique :: Int32
197+ , _monitorActions :: Map ProcessId [IO ()]
198+ , _localMonitorCounter :: Int32
188199 }
189200
190201-- | Processes running on our local node
@@ -229,40 +240,16 @@ send them msg = do
229240getSelfPid :: Process ProcessId
230241getSelfPid = processId <$> ask
231242
232−-- | Variation on 'monitorProcess' where the monitor is the current process
233−--
234−-- TODO: we need to decide what to do with multiple calls to monitor with
235−-- the same argument. The Erlang manual is very specific: it creates multiple
236−-- monitors (http://www.erlang.org/doc/reference_manual/processes.html#id82613)
237−monitor :: ProcessId -> MonitorAction -> Process ()
238−monitor them action = do
239− proc <- ask
240− liftIO $ do
241− tid <- myThreadId
242− let err = ProcessMonitorException them SrNoPing
243− act = case action of
244− MaMonitor -> enqueue (processQueue proc) $ createMessage err
245− MaLink -> throwTo tid err
246− MaLinkError -> fail "monitor: MaLinkError not implemented"
247− modifyMVar_ (localState . processNode $ proc) $
248− return . (monitorActionsFor them ^: (act :))
249−
250−{-
251− ourState <- processState <$> ask
252− liftIO $ modifyMVar_ ourState $ \st ->
253− return . (monitorActionFor them ^= Just action) $ st
254−-}
255−
256−-- | Have one process monitor another
257−--
258−-- TODO: this is a weird function to expose at the API level ('monitor' is more
259−-- natural and more Erlang-like)
260−monitorProcess :: ProcessId -> ProcessId -> MonitorAction -> Process ()
261−monitorProcess = undefined
262−
263−-- | Link failure in two processes
264−linkProcess :: ProcessId -> Process ()
265−linkProcess = undefined
243+monitor :: ProcessId -> Process MonitorRef
244+monitor them = do
245+ us <- getSelfPid
246+ monitorRef <- getMonitorRef
247+ ourCtrlChan <- localCtrlChan . processNode <$> ask
248+ liftIO . writeChan ourCtrlChan $ NodeCtrlMsg
249+ { ctrlMsgSender = Left us
250+ , ctrlMsgSignal = Monitor them monitorRef
251+ }
252+ return monitorRef
266253
267254--------------------------------------------------------------------------------
268255-- Auxiliary API --
@@ -290,93 +277,166 @@ expectTimeout timeout = do
290277 dequeueMatching queue (Just timeout) ((== fp) . messageFingerprint)
291278 return $ fmap (decode . messageEncoding) msg
292279
280+
293281--------------------------------------------------------------------------------
294−-- Monitoring --
295−-- --
296−-- TODO: Many of these definitions are not available in the paper, and are --
297−-- taken from 'remote'. Do we want to stick to them precisely? --
282+-- The node controller --
298283--------------------------------------------------------------------------------
299284
300−-- | The different kinds of monitoring available between processes.
301−data MonitorAction =
302− -- MaMonitor means that the monitor process will be sent a
303− -- ProcessMonitorException message when the monitee terminates for any
304− -- reason.
305− MaMonitor
306− -- MaLink means that the monitor process will receive an asynchronous
307− -- exception of type ProcessMonitorException when the monitee terminates
308− -- for any reason
309− | MaLink
310− -- MaLinkError means that the monitor process will receive an asynchronous
311− -- exception of type ProcessMonitorException when the monitee terminates
312− -- abnormally
313− | MaLinkError
314− deriving (Typeable, Show, Ord, Eq)
315−
316−-- | The main form of notification to a monitoring process that a monitored
317−-- process has terminated. This data structure can be delivered to the monitor
318−-- either as a message (if the monitor is of type 'MaMonitor') or as an
319−-- asynchronous exception (if the monitor is of type 'MaLink' or
320−-- 'MaLinkError'). It contains the PID of the monitored process and the reason
321−-- for its nofication.
322−data ProcessMonitorException = ProcessMonitorException ProcessId SignalReason
285+data ProcessSignal =
286+ Monitor ProcessId MonitorRef
287+ | Died Identifier DiedReason
288+
289+type MonitorRef = Int32
290+
291+data MonitorReply = ProcessDied MonitorRef ProcessId DiedReason
323292 deriving (Typeable)
324293
325−-- | Part of the notification system of process monitoring, indicating why the
326−-- monitor is being notified.
327−data SignalReason =
328− -- the monitee terminated normally
329− SrNormal
330− -- the monitee terminated with an uncaught exception, which is given as a
331− -- string
332− | SrException String
333− -- the monitee is believed to have ended or be inaccessible, as the node on
334− -- which its running is not responding to pings. This may indicate a
335− -- network bisection or that the remote node has crashed.
336− | SrNoPing
337− -- the monitee was not running at the time of the attempt to establish
338− -- monitoring
339− | SrInvalid
340− deriving (Typeable,Show)
341−
342−instance Binary MonitorAction where
343− put MaMonitor = putWord8 0
344− put MaLink = putWord8 1
345− put MaLinkError = putWord8 2
346−
347− get = do x <- getWord8
348− case x of
349− 0 -> return MaMonitor
350− 1 -> return MaLink
351− 2 -> return MaLinkError
352− _ -> fail "Invalid MonitorAction"
353−
354−instance Binary ProcessMonitorException where
355− put (ProcessMonitorException pid sr) = put pid >> put sr
356− get = liftM2 ProcessMonitorException get get
357−
358−instance Binary SignalReason where
359− put SrNormal = putWord8 0
360− put (SrException s) = putWord8 1 >> put s
361− put SrNoPing = putWord8 2
362− put SrInvalid = putWord8 3
363−
364− get = do a <- getWord8
365− case a of
366− 0 -> return SrNormal
367− 1 -> liftM SrException get
368− 2 -> return SrNoPing
369− 3 -> return SrInvalid
370− _ -> fail "Invalid SignalReason"
371−
372−instance Exception ProcessMonitorException
373−
374−instance Show ProcessMonitorException where
375− show (ProcessMonitorException pid why) =
376− "ProcessMonitorException: "
377− ++ show pid
378− ++ " has terminated because "
379− ++ show why
294+data DiedReason =
295+ DiedNormal
296+ | DiedDisconnect
297+ | DiedNodeDown
298+
299+type Identifier = Either ProcessId NodeId
300+
301+data NodeCtrlMsg = NodeCtrlMsg
302+ { ctrlMsgSender :: Identifier
303+ , ctrlMsgSignal :: ProcessSignal
304+ }
305+
306+data NodeCtrlState = NodeCtrlState
307+ { -- Mapping from remote processes to linked local processes
308+ _nodeCtrlLinks :: Map NodeId (Map LocalProcessId (Set ProcessId))
309+ -- Mapping from remote processes to monitoring local processes
310+ , _nodeCtrlMons :: Map NodeId (Map LocalProcessId (Map ProcessId (Set (MonitorRef))))
311+ -- The node controller maintains its own set of connections
312+ -- TODO: still not convinced that this is correct
313+ , _nodeCtrlConns :: Map Identifier NT.Connection
314+ }
315+
316+destNid :: ProcessSignal -> Maybe NodeId
317+destNid (Monitor pid _) = Just $ processNodeId pid
318+destNid (Died (Right _) _) = Nothing
319+destNid (Died (Left _pid) _) = fail "destNid: TODO"
320+
321+initNodeCtrlState :: NodeCtrlState
322+initNodeCtrlState = NodeCtrlState
323+ { _nodeCtrlLinks = Map.empty
324+ , _nodeCtrlMons = Map.empty
325+ , _nodeCtrlConns = Map.empty
326+ }
327+
328+nodeController :: LocalNode -> IO ()
329+nodeController node = go initNodeCtrlState
330+ where
331+ -- [Unified: Table 7]
332+ go :: NodeCtrlState -> IO ()
333+ go st = do
334+ msg <- readChan ourChan
335+ st' <- case destNid (ctrlMsgSignal msg) of
336+ Just nid' | nid' /= localNodeId node -> do
337+ (st', mConn) <- connTo st (Right nid')
338+ didSend <- case mConn of
339+ Just conn -> do
340+ didSend <- NT.send conn (BSL.toChunks . encode $ msg)
341+ case didSend of
342+ Left _ -> return False
343+ Right () -> return True
344+ Nothing ->
345+ return False
346+ if didSend
347+ then
348+ return st'
349+ else do
350+ writeChan ourChan $ NodeCtrlMsg
351+ { ctrlMsgSender = Right nid'
352+ , ctrlMsgSignal = Died (Right nid') DiedDisconnect
353+ }
354+ return st'
355+ _ ->
356+ -- No need to forward
357+ return st
358+ st'' <- processCtrlMsg node st' msg
359+ go st''
360+
361+ ourChan = localCtrlChan node
362+ ourEndPoint = localEndPoint node
363+
364+ connTo :: NodeCtrlState
365+ -> Identifier
366+ -> IO (NodeCtrlState, Maybe NT.Connection)
367+ connTo st them = do
368+ case st ^. nodeCtrlConnFor them of
369+ Nothing -> do
370+ (addr, firstMsg) <- return $ case them of
371+ Left pid -> ( nodeAddress (processNodeId pid)
372+ , idToPayload (Just $ processLocalId pid)
373+ )
374+ Right nid -> ( nodeAddress nid
375+ , idToPayload Nothing
376+ )
377+ mConn <- NT.connect ourEndPoint
378+ addr
379+ NT.ReliableOrdered
380+ NT.defaultConnectHints
381+ case mConn of
382+ Right conn -> do
383+ didSend <- NT.send conn firstMsg
384+ case didSend of
385+ Left _ ->
386+ return (st, Nothing)
387+ Right () ->
388+ return (nodeCtrlConnFor them ^= Just conn $ st, Just conn)
389+ Left _ ->
390+ return (st, Nothing)
391+ Just conn ->
392+ return (st, Just conn)
393+
394+-- [Unified: ncEffect]
395+processCtrlMsg :: LocalNode
396+ -> NodeCtrlState
397+ -> NodeCtrlMsg
398+ -> IO (NodeCtrlState)
399+-- [Unified: Table 10]
400+processCtrlMsg node st (NodeCtrlMsg from (Monitor them ref)) = do
401+ shouldLink <-
402+ if processNodeId them /= localNodeId node
403+ then return True
404+ else withMVar (localState node) $
405+ return . isNothing . (^. localProcessWithId (processLocalId them))
406+ if shouldLink
407+ then
408+ return (nodeCtrlMonsFor them (fromLeft from) ^: Set.insert ref $ st)
409+ else
410+ fail "processCtrlMsg: TODO"
411+
412+-- [Unified Table 9 and Table 12]
413+processCtrlMsg node st (NodeCtrlMsg _from (Died (Right nid) reason)) = do
414+ let localLinks :: Map LocalProcessId (Set ProcessId)
415+ localLinks = st ^. nodeCtrlLinksForNode nid
416+
417+ localMons :: Map LocalProcessId (Map ProcessId (Set MonitorRef))
418+ localMons = st ^. nodeCtrlMonsForNode nid
419+
420+ forM_ (Map.toList localLinks) $ \(_them, _uss) -> do
421+ fail "disconnected: linking not implemented"
422+
423+ forM_ (Map.toList localMons) $ \(them, uss) -> do
424+ forM_ (Map.toList uss) $ \(us, refs) -> do
425+ -- We only need to notify local processes
426+ when (processNodeId us == localNodeId node) $ do
427+ let lpid = processLocalId us
428+ rpid = ProcessId nid them
429+ forM_ refs $ \ref -> do
430+ mProc <- withMVar (localState node) $
431+ return . (^. localProcessWithId lpid)
432+ forM_ mProc $ \proc ->
433+ enqueue (processQueue proc) . createMessage $
434+ ProcessDied ref rpid reason
435+
436+ return . (nodeCtrlLinksForNode nid ^= Map.empty)
437+ . (nodeCtrlMonsForNode nid ^= Map.empty)
438+ $ st
439+
380440
381441--------------------------------------------------------------------------------
382442-- Initialization --
@@ -390,14 +450,19 @@ newLocalNode transport = do
390450 Right endPoint -> do
391451 unq <- randomIO
392452 state <- newMVar LocalNodeState
393− { _localProcesses = Map.empty
394− , _localPidCounter = 0
395− , _localPidUnique = unq
396− , _monitorActions = Map.empty
453+ { _localProcesses = Map.empty
454+ , _localPidCounter = 0
455+ , _localPidUnique = unq
456+ , _monitorActions = Map.empty
457+ , _localMonitorCounter = 0
397458 }
398− let node = LocalNode { localEndPoint = endPoint
459+ ctrlChan <- newChan
460+ let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
461+ , localEndPoint = endPoint
399462 , localState = state
463+ , localCtrlChan = ctrlChan
400464 }
465+ void . forkIO $ nodeController node
401466 void . forkIO $ handleIncomingMessages node
402467 return node
403468
@@ -414,7 +479,7 @@ forkProcess node proc = do
414479 let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
415480 , lpidUnique = st ^. localPidUnique
416481 }
417− let pid = ProcessId { processAddress = NT.address (localEndPoint node)
482+ let pid = ProcessId { processNodeId = localNodeId node
418483 , processLocalId = lpid
419484 }
420485 pst <- newMVar LocalProcessState { _connections = Map.empty
@@ -436,6 +501,7 @@ forkProcess node proc = do
436501handleIncomingMessages :: LocalNode -> IO ()
437502handleIncomingMessages node = go [] Map.empty
438503 where
504+ go :: [NT.ConnectionId] -> Map NT.ConnectionId LocalProcess -> IO ()
439505 go halfOpenConns openConns = do
440506 event <- NT.receive endpoint
441507 case event of
@@ -449,7 +515,7 @@ handleIncomingMessages node = go [] Map.empty
449515 go halfOpenConns openConns
450516 Nothing -> if cid `elem` halfOpenConns
451517 then do
452− let lpid = payloadToLpid payload
518+ let Just lpid = payloadToId payload
453519 mProc <- withMVar state $ return . (^. localProcessWithId lpid)
454520 case mProc of
455521 Just proc ->
@@ -464,7 +530,8 @@ handleIncomingMessages node = go [] Map.empty
464530 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr _) _) -> do
465531 -- TODO: we could cache this information
466532 pids <- withMVar state $ \st ->
467− return $ filter ((== theirAddr) . processAddress) . Map.keys
533+ return $ filter ((== theirAddr) . nodeAddress . processNodeId)
534+ . Map.keys
468535 $ st ^. monitorActions
469536 forM_ pids $ remoteProcessFailed node
470537 NT.ErrorEvent _ ->
@@ -493,7 +560,7 @@ createConnectionTo :: ProcessId -> Process (Maybe NT.Connection)
493560createConnectionTo them = do
494561 proc <- ask
495562 mConn <- liftIO $ NT.connect (localEndPoint . processNode $ proc)
496− (processAddress them)
563+ (nodeAddress . processNodeId $ them)
497564 NT.ReliableOrdered
498565 NT.defaultConnectHints
499566 case mConn of
@@ -512,7 +579,7 @@ createConnectionTo them = do
512579 liftIO $ NT.close conn
513580 return . Just $ conn'
514581 Nothing -> do
515− sendBinary them conn $ lpidToPayload (processLocalId them)
582+ sendBinary them conn . idToPayload . Just . processLocalId $ them
516583 return . Just $ conn
517584 Left _err -> do
518585 -- TODO: should probably pass this error to remoteProcessFailed
@@ -541,16 +608,26 @@ payloadToMessage payload = Message fp msg
541608 $ BSL.fromChunks payload
542609 fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp
543610
544−lpidToPayload :: LocalProcessId -> [BSS.ByteString]
545−lpidToPayload lpid = [ NTI.encodeInt32 (lpidCounter lpid)
546− , NTI.encodeInt32 (lpidUnique lpid)
547− ]
548−
549−payloadToLpid :: [BSS.ByteString] -> LocalProcessId
550−payloadToLpid bss = let (bs1, bs2) = BSS.splitAt 4 . BSS.concat $ bss
551− in LocalProcessId { lpidCounter = NTI.decodeInt32 bs1
552− , lpidUnique = NTI.decodeInt32 bs2
553− }
611+-- | The first message we send across a connection to indicate the intended
612+-- recipient. Pass Nothing for the remote node controller
613+idToPayload :: Maybe LocalProcessId -> [BSS.ByteString]
614+idToPayload Nothing = [ NTI.encodeInt32 (0 :: Int) ]
615+idToPayload (Just lpid) = [ NTI.encodeInt32 (1 :: Int)
616+ , NTI.encodeInt32 (lpidCounter lpid)
617+ , NTI.encodeInt32 (lpidUnique lpid)
618+ ]
619+
620+-- | Inverse of 'idToPayload'
621+payloadToId :: [BSS.ByteString] -> Maybe LocalProcessId
622+payloadToId bss = let (bs1, bss') = BSS.splitAt 4 . BSS.concat $ bss
623+ (bs2, bs3) = BSS.splitAt 4 bss' in
624+ case NTI.decodeInt32 bs1 :: Int of
625+ 0 -> Nothing
626+ 1 -> Just $ LocalProcessId
627+ { lpidCounter = NTI.decodeInt32 bs2
628+ , lpidUnique = NTI.decodeInt32 bs3
629+ }
630+ _ -> fail "payloadToId"
554631
555632remoteProcessFailed :: LocalNode -> ProcessId -> IO ()
556633remoteProcessFailed node them = do
@@ -580,6 +657,14 @@ remoteProcessFailed node them = do
580657createMessage :: Serializable a => a -> Message
581658createMessage a = Message (fingerprint a) (encode a)
582659
660+getMonitorRef :: Process MonitorRef
661+getMonitorRef = do
662+ node <- processNode <$> ask
663+ liftIO $ modifyMVar (localState node) $ \st ->
664+ return ( localMonitorCounter ^: (+ 1) $ st
665+ , st ^. localMonitorCounter
666+ )
667+
583668-- This is most definitely NOT exported
584669runLocalProcess :: LocalProcess -> Process a -> IO a
585670runLocalProcess lproc proc = runReaderT (unProcess proc) lproc
@@ -593,9 +678,38 @@ instance Binary LocalProcessId where
593678 get = liftM2 LocalProcessId get get
594679
595680instance Binary ProcessId where
596− put pid = put (processAddress pid) >> put (processLocalId pid)
681+ put pid = put (processNodeId pid) >> put (processLocalId pid)
597682 get = liftM2 ProcessId get get
598683
684+instance Binary MonitorReply where
685+ put (ProcessDied ref pid reason) = put ref >> put pid >> put reason
686+ get = liftM3 ProcessDied get get get
687+
688+instance Binary NodeCtrlMsg where
689+ put msg = put (ctrlMsgSender msg) >> put (ctrlMsgSignal msg)
690+ get = liftM2 NodeCtrlMsg get get
691+
692+instance Binary ProcessSignal where
693+ put (Monitor pid ref) = putWord8 0 >> put pid >> put ref
694+ put (Died who reason) = putWord8 1 >> put who >> put reason
695+ get = do
696+ header <- getWord8
697+ case header of
698+ 0 -> liftM2 Monitor get get
699+ _ -> fail "ProcessSignal.get: invalid"
700+
701+instance Binary DiedReason where
702+ put DiedNormal = putWord8 0
703+ put DiedDisconnect = putWord8 1
704+ put DiedNodeDown = putWord8 2
705+ get = do
706+ header <- getWord8
707+ case header of
708+ 0 -> return DiedNormal
709+ 1 -> return DiedDisconnect
710+ 2 -> return DiedNodeDown
711+ _ -> fail "DiedReason.get: invalid"
712+
599713--------------------------------------------------------------------------------
600714-- Accessors --
601715--------------------------------------------------------------------------------
@@ -612,6 +726,9 @@ localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq
612726monitorActions :: Accessor LocalNodeState (Map ProcessId [IO ()])
613727monitorActions = accessor _monitorActions (\as st -> st { _monitorActions = as })
614728
729+localMonitorCounter :: Accessor LocalNodeState Int32
730+localMonitorCounter = accessor _localMonitorCounter (\ctr st -> st { _localMonitorCounter = ctr })
731+
615732localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)
616733localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid
617734
@@ -623,3 +740,40 @@ connections = accessor _connections (\conns st -> st { _connections = conns })
623740
624741connectionTo :: ProcessId -> Accessor LocalProcessState (Maybe NT.Connection)
625742connectionTo pid = connections >>> DAC.mapMaybe pid
743+
744+nodeCtrlLinks :: Accessor NodeCtrlState (Map NodeId (Map LocalProcessId (Set ProcessId)))
745+nodeCtrlLinks = accessor _nodeCtrlLinks (\links st -> st { _nodeCtrlLinks = links })
746+
747+nodeCtrlMons :: Accessor NodeCtrlState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))
748+nodeCtrlMons = accessor _nodeCtrlMons (\mons st -> st { _nodeCtrlMons = mons })
749+
750+nodeCtrlConns :: Accessor NodeCtrlState (Map Identifier NT.Connection)
751+nodeCtrlConns = accessor _nodeCtrlConns (\conns st -> st { _nodeCtrlConns = conns })
752+
753+nodeCtrlConnFor :: Identifier -> Accessor NodeCtrlState (Maybe NT.Connection)
754+nodeCtrlConnFor them = nodeCtrlConns >>> DAC.mapMaybe them
755+
756+nodeCtrlLinksForNode :: NodeId -> Accessor NodeCtrlState (Map LocalProcessId (Set ProcessId))
757+nodeCtrlLinksForNode nid = nodeCtrlLinks >>> DAC.mapDefault Map.empty nid
758+
759+nodeCtrlMonsForNode :: NodeId -> Accessor NodeCtrlState (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
760+nodeCtrlMonsForNode nid = nodeCtrlMons >>> DAC.mapDefault Map.empty nid
761+
762+nodeCtrlLinksForProcess :: ProcessId -> Accessor NodeCtrlState (Set ProcessId)
763+nodeCtrlLinksForProcess pid = nodeCtrlLinksForNode (processNodeId pid) >>> DAC.mapDefault Set.empty (processLocalId pid)
764+
765+nodeCtrlMonsForProcess :: ProcessId -> Accessor NodeCtrlState (Map ProcessId (Set MonitorRef))
766+nodeCtrlMonsForProcess pid = nodeCtrlMonsForNode (processNodeId pid) >>> DAC.mapDefault Map.empty (processLocalId pid)
767+
768+nodeCtrlMonsFor :: ProcessId -> ProcessId -> Accessor NodeCtrlState (Set MonitorRef)
769+nodeCtrlMonsFor them us = nodeCtrlMonsForProcess them >>> DAC.mapDefault Set.empty us
770+
771+
772+--------------------------------------------------------------------------------
773+-- Auxiliary --
774+--------------------------------------------------------------------------------
775+
776+fromLeft :: Either a b -> a
777+fromLeft (Left a) = a
778+fromLeft (Right _) = error "fromLeft"
779+