@@ -17,16 +17,18 @@
18-- In order to guarantee (3), we stipulate that
18-- In order to guarantee (3), we stipulate that
20−-- 3a. Once a connection to a remote process fails, that process is considered
20+-- 3a. We do not garbage collect connections because Network.Transport provides
21−-- forever unreachable. When the remote process restarts, it will receive a
21+-- ordering guarantees only *per connection*.
22−-- brand new ProcessId.
24−-- 3b. We do not garbage collect (lightweight) connections, because we have
23+-- 3b. Once a connection breaks, we have no way of knowing which messages
25−-- ordering guarantees from Network.Transport only per lightweight
24+-- arrived and which did not; hence, once a connection fails, we assume the
26−-- connection. We could lift this restriction later, if we wish, by adding
25+-- remote process to be forever unreachable. Otherwise we might sent m1 and
27−-- some acknowledgement control messages: we can drop one lightweight
26+-- m2, get notified of the broken connection, reconnect, send m3, but only
28−-- connection and open another once we know that all messages sent on the
27+-- m1 and m3 arrive.
29−-- former have been received.
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.
31-- Main reference for Cloud Haskell is
33-- Main reference for Cloud Haskell is
@@ -70,13 +72,10 @@ module Control.Distributed.Process
72 , getSelfPid
74 , getSelfPid
75+ -- * Monitoring and linking
74− , monitorProcess
77+ , MonitorReply(..)
75− , linkProcess
78+ , DiedReason(..)
78− , ProcessMonitorException(..)
80 -- * Initialization
79 -- * Initialization
81 , newLocalNode
80 , newLocalNode
82 , forkProcess
81 , forkProcess
@@ -94,20 +93,23 @@ import qualified Data.ByteString.Lazy as BSL ( ByteString
94 , fromChunks
93 , fromChunks
97−import Data.Binary (Binary, decode, encode, put, get, getWord8, putWord8)
96+import Data.Binary (Binary, decode, encode, put, get, putWord8, getWord8)
98import Data.Map (Map)
97import 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)
100import qualified Data.List as List (delete)
99import qualified Data.List as List (delete)
100+import Data.Set (Set)
101+import qualified Data.Set as Set (empty, insert)
101import Data.Int (Int32)
102import Data.Int (Int32)
102import Data.Typeable (Typeable)
103import Data.Typeable (Typeable)
103import Data.Foldable (forM_)
104import Data.Foldable (forM_)
105+import Data.Maybe (isNothing)
104−import Control.Monad (void, liftM, liftM2)
106+import Control.Monad (void, liftM2, liftM3, when)
105import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
107import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
106import Control.Monad.IO.Class (MonadIO, liftIO)
108import Control.Monad.IO.Class (MonadIO, liftIO)
107import Control.Applicative ((<$>))
109import Control.Applicative ((<$>))
108import Control.Category ((>>>))
110import Control.Category ((>>>))
109−import Control.Exception (Exception, throwIO, catch, throwTo)
111+import Control.Exception (Exception, throwIO, catch)
110−import Control.Concurrent (forkIO, myThreadId)
112+import Control.Concurrent (forkIO)
111import Control.Concurrent.MVar ( MVar
113import Control.Concurrent.MVar ( MVar
112 , newMVar
114 , newMVar
113 , withMVar
115 , withMVar
@@ -117,6 +119,7 @@ import Control.Concurrent.MVar ( MVar
117 , putMVar
119 , putMVar
118 , takeMVar
120 , takeMVar
122+import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
120import Control.Distributed.Process.Internal.CQueue ( CQueue
123import Control.Distributed.Process.Internal.CQueue ( CQueue
121 , dequeueMatching
124 , dequeueMatching
122 , enqueue
125 , enqueue
@@ -145,6 +148,7 @@ import qualified Network.Transport as NT ( Transport
145 , TransportError(..)
148 , TransportError(..)
146 , address
149 , address
147 , closeEndPoint
150 , closeEndPoint
149import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)
153import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)
150import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
154import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
@@ -159,10 +163,14 @@ data LocalProcessId = LocalProcessId
160 deriving (Eq, Ord, Typeable, Show)
164 deriving (Eq, Ord, Typeable, Show)
166+-- We identify node IDs and endpoint IDs
167+newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress }
168+ deriving (Show, Eq, Ord, Binary)
162-- | A process ID combines a local process with with an endpoint address
170-- | A process ID combines a local process with with an endpoint address
163-- (in other words, we identify nodes and endpoints)
171-- (in other words, we identify nodes and endpoints)
164data ProcessId = ProcessId
172data ProcessId = ProcessId
165− { processAddress :: NT.EndPointAddress
173+ { processNodeId :: NodeId
166 , processLocalId :: LocalProcessId
174 , processLocalId :: LocalProcessId
168 deriving (Eq, Ord, Typeable, Show)
176 deriving (Eq, Ord, Typeable, Show)
@@ -175,16 +183,19 @@ data Message = Message
176-- | Local nodes
184-- | Local nodes
177data LocalNode = LocalNode
185data LocalNode = LocalNode
186+ { localNodeId :: NodeId
178− { localEndPoint :: NT.EndPoint
187+ , localEndPoint :: NT.EndPoint
179− , localState :: MVar LocalNodeState
188+ , localState :: MVar LocalNodeState
189+ , localCtrlChan :: Chan NodeCtrlMsg
182-- | Local node state
192-- | Local node state
183data LocalNodeState = LocalNodeState
193data LocalNodeState = LocalNodeState
184− { _localProcesses :: Map LocalProcessId LocalProcess
194+ { _localProcesses :: Map LocalProcessId LocalProcess
185− , _localPidCounter :: Int32
195+ , _localPidCounter :: Int32
186− , _localPidUnique :: Int32
196+ , _localPidUnique :: Int32
187− , _monitorActions :: Map ProcessId [IO ()]
197+ , _monitorActions :: Map ProcessId [IO ()]
198+ , _localMonitorCounter :: Int32
190-- | Processes running on our local node
201-- | Processes running on our local node
@@ -229,40 +240,16 @@ send them msg = do
229getSelfPid :: Process ProcessId
240getSelfPid :: Process ProcessId
230getSelfPid = processId <$> ask
241getSelfPid = processId <$> ask
232−-- | Variation on 'monitorProcess' where the monitor is the current process
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 ()
243+monitor :: ProcessId -> Process MonitorRef
238−monitor them action = do
244+monitor them = do
239− proc <- ask
245+ us <- getSelfPid
240− liftIO $ do
246+ monitorRef <- getMonitorRef
241− tid <- myThreadId
247+ ourCtrlChan <- localCtrlChan . processNode <$> ask
242− let err = ProcessMonitorException them SrNoPing
248+ liftIO . writeChan ourCtrlChan $ NodeCtrlMsg
243− act = case action of
249+ { ctrlMsgSender = Left us
244− MaMonitor -> enqueue (processQueue proc) $ createMessage err
250+ , ctrlMsgSignal = Monitor them monitorRef
245− MaLink -> throwTo tid err
246− MaLinkError -> fail "monitor: MaLinkError not implemented"
247− modifyMVar_ (localState . processNode $ proc) $
248− return . (monitorActionsFor them ^: (act :))
251− ourState <- processState <$> ask
252− liftIO $ modifyMVar_ ourState $ \st ->
253− return . (monitorActionFor them ^= Just action) $ st
255−
252+ return monitorRef
256−-- | Have one process monitor another
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
263−-- | Link failure in two processes
264−linkProcess :: ProcessId -> Process ()
265−linkProcess = undefined
267--------------------------------------------------------------------------------
254--------------------------------------------------------------------------------
268-- Auxiliary API --
255-- Auxiliary API --
@@ -290,93 +277,166 @@ expectTimeout timeout = do
290 dequeueMatching queue (Just timeout) ((== fp) . messageFingerprint)
277 dequeueMatching queue (Just timeout) ((== fp) . messageFingerprint)
291 return $ fmap (decode . messageEncoding) msg
278 return $ fmap (decode . messageEncoding) msg
293--------------------------------------------------------------------------------
281--------------------------------------------------------------------------------
294−-- Monitoring --
282+-- The node controller --
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? --
298--------------------------------------------------------------------------------
283--------------------------------------------------------------------------------
300−-- | The different kinds of monitoring available between processes.
285+data ProcessSignal =
301−data MonitorAction =
286+ Monitor ProcessId MonitorRef
302− -- MaMonitor means that the monitor process will be sent a
287+ | Died Identifier DiedReason
303− -- ProcessMonitorException message when the monitee terminates for any
288+
304− -- reason.
289+type MonitorRef = Int32
306− -- MaLink means that the monitor process will receive an asynchronous
291+data MonitorReply = ProcessDied MonitorRef ProcessId DiedReason
307− -- exception of type ProcessMonitorException when the monitee terminates
310− -- MaLinkError means that the monitor process will receive an asynchronous
311− -- exception of type ProcessMonitorException when the monitee terminates
314− deriving (Typeable, Show, Ord, Eq)
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
323 deriving (Typeable)
292 deriving (Typeable)
325−-- | Part of the notification system of process monitoring, indicating why the
294+data DiedReason =
326−-- monitor is being notified.
295+ DiedNormal
327−data SignalReason =
296+ | DiedDisconnect
328− -- the monitee terminated normally
297+ | DiedNodeDown
330− -- the monitee terminated with an uncaught exception, which is given as a
299+type Identifier = Either ProcessId NodeId
332− | SrException String
301+data NodeCtrlMsg = NodeCtrlMsg
333− -- the monitee is believed to have ended or be inaccessible, as the node on
302+ { ctrlMsgSender :: Identifier
334− -- which its running is not responding to pings. This may indicate a
303+ , ctrlMsgSignal :: ProcessSignal
335− -- network bisection or that the remote node has crashed.
304+ }
337− -- the monitee was not running at the time of the attempt to establish
306+data NodeCtrlState = NodeCtrlState
338− -- monitoring
307+ { -- Mapping from remote processes to linked local processes
339− | SrInvalid
308+ _nodeCtrlLinks :: Map NodeId (Map LocalProcessId (Set ProcessId))
340− deriving (Typeable,Show)
309+ -- Mapping from remote processes to monitoring local processes
341−
310+ , _nodeCtrlMons :: Map NodeId (Map LocalProcessId (Map ProcessId (Set (MonitorRef))))
342−instance Binary MonitorAction where
311+ -- The node controller maintains its own set of connections
343− put MaMonitor = putWord8 0
312+ -- TODO: still not convinced that this is correct
344− put MaLink = putWord8 1
313+ , _nodeCtrlConns :: Map Identifier NT.Connection
345− put MaLinkError = putWord8 2
314+ }
347− get = do x <- getWord8
316+destNid :: ProcessSignal -> Maybe NodeId
348− case x of
317+destNid (Monitor pid _) = Just $ processNodeId pid
349− 0 -> return MaMonitor
318+destNid (Died (Right _) _) = Nothing
350− 1 -> return MaLink
319+destNid (Died (Left _pid) _) = fail "destNid: TODO"
351− 2 -> return MaLinkError
320+
352− _ -> fail "Invalid MonitorAction"
321+initNodeCtrlState :: NodeCtrlState
353−
322+initNodeCtrlState = NodeCtrlState
354−instance Binary ProcessMonitorException where
323+ { _nodeCtrlLinks = Map.empty
355− put (ProcessMonitorException pid sr) = put pid >> put sr
324+ , _nodeCtrlMons = Map.empty
356− get = liftM2 ProcessMonitorException get get
325+ , _nodeCtrlConns = Map.empty
358−instance Binary SignalReason where
327+
359− put SrNormal = putWord8 0
328+nodeController :: LocalNode -> IO ()
360− put (SrException s) = putWord8 1 >> put s
329+nodeController node = go initNodeCtrlState
361− put SrNoPing = putWord8 2
330+ where
362− put SrInvalid = putWord8 3
331+ -- [Unified: Table 7]
363−
332+ go :: NodeCtrlState -> IO ()
364− get = do a <- getWord8
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
340+ didSend <- NT.send conn (BSL.toChunks . encode $ msg)
365− case a of
341+ case didSend of
366− 0 -> return SrNormal
342+ Left _ -> return False
367− 1 -> liftM SrException get
343+ Right () -> return True
368− 2 -> return SrNoPing
344+ Nothing ->
369− 3 -> return SrInvalid
345+ return False
370− _ -> fail "Invalid SignalReason"
346+ if didSend
372−instance Exception ProcessMonitorException
348+ return st'
374−instance Show ProcessMonitorException where
350+ writeChan ourChan $ NodeCtrlMsg
375− show (ProcessMonitorException pid why) =
351+ { ctrlMsgSender = Right nid'
376− "ProcessMonitorException: "
352+ , ctrlMsgSignal = Died (Right nid') DiedDisconnect
378− ++ " has terminated because "
354+ return st'
379− ++ show why
355+ _ ->
356+ -- No need to forward
358+ st'' <- processCtrlMsg node st' msg
361+ ourChan = localCtrlChan node
362+ ourEndPoint = localEndPoint node
364+ connTo :: NodeCtrlState
366+ -> IO (NodeCtrlState, Maybe NT.Connection)
368+ case st ^. nodeCtrlConnFor them of
370+ (addr, firstMsg) <- return $ case them of
371+ Left pid -> ( nodeAddress (processNodeId pid)
372+ , idToPayload (Just $ processLocalId pid)
374+ Right nid -> ( nodeAddress nid
375+ , idToPayload Nothing
377+ mConn <- NT.connect ourEndPoint
380+ NT.defaultConnectHints
383+ didSend <- NT.send conn firstMsg
386+ return (st, Nothing)
388+ return (nodeCtrlConnFor them ^= Just conn $ st, Just conn)
390+ return (st, Nothing)
392+ return (st, Just conn)
394+-- [Unified: ncEffect]
395+processCtrlMsg :: LocalNode
398+ -> IO (NodeCtrlState)
399+-- [Unified: Table 10]
400+processCtrlMsg node st (NodeCtrlMsg from (Monitor them ref)) = do
402+ if processNodeId them /= localNodeId node
404+ else withMVar (localState node) $
405+ return . isNothing . (^. localProcessWithId (processLocalId them))
408+ return (nodeCtrlMonsFor them (fromLeft from) ^: Set.insert ref $ st)
410+ fail "processCtrlMsg: TODO"
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
417+ localMons :: Map LocalProcessId (Map ProcessId (Set MonitorRef))
418+ localMons = st ^. nodeCtrlMonsForNode nid
420+ forM_ (Map.toList localLinks) $ \(_them, _uss) -> do
421+ fail "disconnected: linking not implemented"
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
436+ return . (nodeCtrlLinksForNode nid ^= Map.empty)
437+ . (nodeCtrlMonsForNode nid ^= Map.empty)
381--------------------------------------------------------------------------------
441--------------------------------------------------------------------------------
382-- Initialization --
442-- Initialization --
@@ -390,14 +450,19 @@ newLocalNode transport = do
390 Right endPoint -> do
450 Right endPoint -> do
391 unq <- randomIO
451 unq <- randomIO
392 state <- newMVar LocalNodeState
452 state <- newMVar LocalNodeState
393− { _localProcesses = Map.empty
453+ { _localProcesses = Map.empty
394− , _localPidCounter = 0
454+ , _localPidCounter = 0
395− , _localPidUnique = unq
455+ , _localPidUnique = unq
396− , _monitorActions = Map.empty
456+ , _monitorActions = Map.empty
457+ , _localMonitorCounter = 0
460+ let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
398− let node = LocalNode { localEndPoint = endPoint
461+ , localEndPoint = endPoint
399 , localState = state
462 , localState = state
463+ , localCtrlChan = ctrlChan
465+ void . forkIO $ nodeController node
401 void . forkIO $ handleIncomingMessages node
466 void . forkIO $ handleIncomingMessages node
402 return node
467 return node
@@ -414,7 +479,7 @@ forkProcess node proc = do
414 let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
479 let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
415 , lpidUnique = st ^. localPidUnique
480 , lpidUnique = st ^. localPidUnique
417− let pid = ProcessId { processAddress = NT.address (localEndPoint node)
482+ let pid = ProcessId { processNodeId = localNodeId node
418 , processLocalId = lpid
483 , processLocalId = lpid
420 pst <- newMVar LocalProcessState { _connections = Map.empty
485 pst <- newMVar LocalProcessState { _connections = Map.empty
@@ -436,6 +501,7 @@ forkProcess node proc = do
436handleIncomingMessages :: LocalNode -> IO ()
501handleIncomingMessages :: LocalNode -> IO ()
437handleIncomingMessages node = go [] Map.empty
502handleIncomingMessages node = go [] Map.empty
504+ go :: [NT.ConnectionId] -> Map NT.ConnectionId LocalProcess -> IO ()
439 go halfOpenConns openConns = do
505 go halfOpenConns openConns = do
440 event <- NT.receive endpoint
506 event <- NT.receive endpoint
441 case event of
507 case event of
@@ -449,7 +515,7 @@ handleIncomingMessages node = go [] Map.empty
449 go halfOpenConns openConns
515 go halfOpenConns openConns
450 Nothing -> if cid `elem` halfOpenConns
516 Nothing -> if cid `elem` halfOpenConns
452− let lpid = payloadToLpid payload
518+ let Just lpid = payloadToId payload
453 mProc <- withMVar state $ return . (^. localProcessWithId lpid)
519 mProc <- withMVar state $ return . (^. localProcessWithId lpid)
454 case mProc of
520 case mProc of
455 Just proc ->
521 Just proc ->
@@ -464,7 +530,8 @@ handleIncomingMessages node = go [] Map.empty
464 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr _) _) -> do
530 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr _) _) -> do
465 -- TODO: we could cache this information
531 -- TODO: we could cache this information
466 pids <- withMVar state $ \st ->
532 pids <- withMVar state $ \st ->
467− return $ filter ((== theirAddr) . processAddress) . Map.keys
533+ return $ filter ((== theirAddr) . nodeAddress . processNodeId)
468 $ st ^. monitorActions
535 $ st ^. monitorActions
469 forM_ pids $ remoteProcessFailed node
536 forM_ pids $ remoteProcessFailed node
470 NT.ErrorEvent _ ->
537 NT.ErrorEvent _ ->
@@ -493,7 +560,7 @@ createConnectionTo :: ProcessId -> Process (Maybe NT.Connection)
493createConnectionTo them = do
560createConnectionTo them = do
494 proc <- ask
561 proc <- ask
495 mConn <- liftIO $ NT.connect (localEndPoint . processNode $ proc)
562 mConn <- liftIO $ NT.connect (localEndPoint . processNode $ proc)
496− (processAddress them)
563+ (nodeAddress . processNodeId $ them)
497 NT.ReliableOrdered
564 NT.ReliableOrdered
498 NT.defaultConnectHints
565 NT.defaultConnectHints
499 case mConn of
566 case mConn of
@@ -512,7 +579,7 @@ createConnectionTo them = do
512 liftIO $ NT.close conn
579 liftIO $ NT.close conn
513 return . Just $ conn'
580 return . Just $ conn'
514 Nothing -> do
581 Nothing -> do
515− sendBinary them conn $ lpidToPayload (processLocalId them)
582+ sendBinary them conn . idToPayload . Just . processLocalId $ them
516 return . Just $ conn
583 return . Just $ conn
517 Left _err -> do
584 Left _err -> do
518 -- TODO: should probably pass this error to remoteProcessFailed
585 -- TODO: should probably pass this error to remoteProcessFailed
@@ -541,16 +608,26 @@ payloadToMessage payload = Message fp msg
541 $ BSL.fromChunks payload
608 $ BSL.fromChunks payload
542 fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp
609 fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp
611+-- | The first message we send across a connection to indicate the intended
612+-- recipient. Pass Nothing for the remote node controller
544−lpidToPayload :: LocalProcessId -> [BSS.ByteString]
613+idToPayload :: Maybe LocalProcessId -> [BSS.ByteString]
614+idToPayload Nothing = [ NTI.encodeInt32 (0 :: Int) ]
545−lpidToPayload lpid = [ NTI.encodeInt32 (lpidCounter lpid)
615+idToPayload (Just lpid) = [ NTI.encodeInt32 (1 :: Int)
616+ , NTI.encodeInt32 (lpidCounter lpid)
546− , NTI.encodeInt32 (lpidUnique lpid)
617+ , NTI.encodeInt32 (lpidUnique lpid)
620+-- | Inverse of 'idToPayload'
549−payloadToLpid :: [BSS.ByteString] -> LocalProcessId
621+payloadToId :: [BSS.ByteString] -> Maybe LocalProcessId
550−payloadToLpid bss = let (bs1, bs2) = BSS.splitAt 4 . BSS.concat $ bss
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
626+ 1 -> Just $ LocalProcessId
551− in LocalProcessId { lpidCounter = NTI.decodeInt32 bs1
627+ { lpidCounter = NTI.decodeInt32 bs2
552− , lpidUnique = NTI.decodeInt32 bs2
628+ , lpidUnique = NTI.decodeInt32 bs3
630+ _ -> fail "payloadToId"
555remoteProcessFailed :: LocalNode -> ProcessId -> IO ()
632remoteProcessFailed :: LocalNode -> ProcessId -> IO ()
556remoteProcessFailed node them = do
633remoteProcessFailed node them = do
@@ -580,6 +657,14 @@ remoteProcessFailed node them = do
580createMessage :: Serializable a => a -> Message
657createMessage :: Serializable a => a -> Message
581createMessage a = Message (fingerprint a) (encode a)
658createMessage a = Message (fingerprint a) (encode a)
660+getMonitorRef :: Process MonitorRef
662+ node <- processNode <$> ask
663+ liftIO $ modifyMVar (localState node) $ \st ->
664+ return ( localMonitorCounter ^: (+ 1) $ st
665+ , st ^. localMonitorCounter
583-- This is most definitely NOT exported
668-- This is most definitely NOT exported
584runLocalProcess :: LocalProcess -> Process a -> IO a
669runLocalProcess :: LocalProcess -> Process a -> IO a
585runLocalProcess lproc proc = runReaderT (unProcess proc) lproc
670runLocalProcess lproc proc = runReaderT (unProcess proc) lproc
@@ -593,9 +678,38 @@ instance Binary LocalProcessId where
593 get = liftM2 LocalProcessId get get
678 get = liftM2 LocalProcessId get get
595instance Binary ProcessId where
680instance Binary ProcessId where
596− put pid = put (processAddress pid) >> put (processLocalId pid)
681+ put pid = put (processNodeId pid) >> put (processLocalId pid)
597 get = liftM2 ProcessId get get
682 get = liftM2 ProcessId get get
684+instance Binary MonitorReply where
685+ put (ProcessDied ref pid reason) = put ref >> put pid >> put reason
686+ get = liftM3 ProcessDied get get get
688+instance Binary NodeCtrlMsg where
689+ put msg = put (ctrlMsgSender msg) >> put (ctrlMsgSignal msg)
690+ get = liftM2 NodeCtrlMsg get get
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
698+ 0 -> liftM2 Monitor get get
699+ _ -> fail "ProcessSignal.get: invalid"
701+instance Binary DiedReason where
702+ put DiedNormal = putWord8 0
703+ put DiedDisconnect = putWord8 1
704+ put DiedNodeDown = putWord8 2
708+ 0 -> return DiedNormal
709+ 1 -> return DiedDisconnect
710+ 2 -> return DiedNodeDown
711+ _ -> fail "DiedReason.get: invalid"
599--------------------------------------------------------------------------------
713--------------------------------------------------------------------------------
600-- Accessors --
714-- Accessors --
601--------------------------------------------------------------------------------
715--------------------------------------------------------------------------------
@@ -612,6 +726,9 @@ localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq
612monitorActions :: Accessor LocalNodeState (Map ProcessId [IO ()])
726monitorActions :: Accessor LocalNodeState (Map ProcessId [IO ()])
613monitorActions = accessor _monitorActions (\as st -> st { _monitorActions = as })
727monitorActions = accessor _monitorActions (\as st -> st { _monitorActions = as })
729+localMonitorCounter :: Accessor LocalNodeState Int32
730+localMonitorCounter = accessor _localMonitorCounter (\ctr st -> st { _localMonitorCounter = ctr })
615localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)
732localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)
616localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid
733localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid
@@ -623,3 +740,40 @@ connections = accessor _connections (\conns st -> st { _connections = conns })
624connectionTo :: ProcessId -> Accessor LocalProcessState (Maybe NT.Connection)
741connectionTo :: ProcessId -> Accessor LocalProcessState (Maybe NT.Connection)
625connectionTo pid = connections >>> DAC.mapMaybe pid
742connectionTo pid = connections >>> DAC.mapMaybe pid
744+nodeCtrlLinks :: Accessor NodeCtrlState (Map NodeId (Map LocalProcessId (Set ProcessId)))
745+nodeCtrlLinks = accessor _nodeCtrlLinks (\links st -> st { _nodeCtrlLinks = links })
747+nodeCtrlMons :: Accessor NodeCtrlState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))
748+nodeCtrlMons = accessor _nodeCtrlMons (\mons st -> st { _nodeCtrlMons = mons })
750+nodeCtrlConns :: Accessor NodeCtrlState (Map Identifier NT.Connection)
751+nodeCtrlConns = accessor _nodeCtrlConns (\conns st -> st { _nodeCtrlConns = conns })
753+nodeCtrlConnFor :: Identifier -> Accessor NodeCtrlState (Maybe NT.Connection)
754+nodeCtrlConnFor them = nodeCtrlConns >>> DAC.mapMaybe them
756+nodeCtrlLinksForNode :: NodeId -> Accessor NodeCtrlState (Map LocalProcessId (Set ProcessId))
757+nodeCtrlLinksForNode nid = nodeCtrlLinks >>> DAC.mapDefault Map.empty nid
759+nodeCtrlMonsForNode :: NodeId -> Accessor NodeCtrlState (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
760+nodeCtrlMonsForNode nid = nodeCtrlMons >>> DAC.mapDefault Map.empty nid
762+nodeCtrlLinksForProcess :: ProcessId -> Accessor NodeCtrlState (Set ProcessId)
763+nodeCtrlLinksForProcess pid = nodeCtrlLinksForNode (processNodeId pid) >>> DAC.mapDefault Set.empty (processLocalId pid)
765+nodeCtrlMonsForProcess :: ProcessId -> Accessor NodeCtrlState (Map ProcessId (Set MonitorRef))
766+nodeCtrlMonsForProcess pid = nodeCtrlMonsForNode (processNodeId pid) >>> DAC.mapDefault Map.empty (processLocalId pid)
768+nodeCtrlMonsFor :: ProcessId -> ProcessId -> Accessor NodeCtrlState (Set MonitorRef)
769+nodeCtrlMonsFor them us = nodeCtrlMonsForProcess them >>> DAC.mapDefault Set.empty us
772+--------------------------------------------------------------------------------
774+--------------------------------------------------------------------------------
776+fromLeft :: Either a b -> a
777+fromLeft (Left a) = a
778+fromLeft (Right _) = error "fromLeft"