git.iko.soy/distributed-process
on master help sign in
Split into multiple files
806e468Edsko de Vries
distributed-process/src/Control/Distributed/Process.hs+37−423
compare with master
unifiedsplit
@@ -78,32 +78,22 @@ module Control.Distributed.Process
78 ) where
78 ) where
79
79
80import Prelude hiding (catch)
80import Prelude hiding (catch)
81−import qualified Data.ByteString as BSS (ByteString, concat, splitAt)
81+import qualified Data.ByteString as BSS (ByteString)
82−import qualified Data.ByteString.Lazy as BSL ( ByteString
82+import qualified Data.ByteString.Lazy as BSL (fromChunks)
83− , toChunks
84− , fromChunks
85− , splitAt
86− )
87−import Data.Binary (Binary, decode, encode, put, get, putWord8, getWord8)
83+import Data.Binary (decode)
88import Data.Map (Map)
84import Data.Map (Map)
89−import qualified Data.Map as Map (empty, lookup, insert, delete, toList)
85+import qualified Data.Map as Map (empty, lookup, insert, delete)
90import qualified Data.List as List (delete)
86import qualified Data.List as List (delete)
91import Data.Set (Set)
87import Data.Set (Set)
92import qualified Data.Set as Set (empty, insert, delete, member)
88import qualified Data.Set as Set (empty, insert, delete, member)
93−import Data.Int (Int32)
94−import Data.Typeable (Typeable)
95import Data.Foldable (forM_)
89import Data.Foldable (forM_)
96−import Data.Maybe (isJust)
97−import Control.Monad (void, when, unless, forever)
90+import Control.Monad (void)
98import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
91import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
99import Control.Monad.IO.Class (MonadIO, liftIO)
92import Control.Monad.IO.Class (MonadIO, liftIO)
100−import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)
101−import Control.Applicative ((<$>), (<*>))
93+import Control.Applicative ((<$>))
102−import Control.Category ((>>>))
103import Control.Exception (Exception, throwIO, catch, SomeException)
94import Control.Exception (Exception, throwIO, catch, SomeException)
104import Control.Concurrent (forkIO)
95import Control.Concurrent (forkIO)
105−import Control.Concurrent.MVar ( MVar
96+import Control.Concurrent.MVar ( newMVar
106− , newMVar
107 , withMVar
97 , withMVar
108 , modifyMVar
98 , modifyMVar
109 , modifyMVar_
99 , modifyMVar_
@@ -111,22 +101,14 @@ import Control.Concurrent.MVar ( MVar
111 , putMVar
101 , putMVar
112 , takeMVar
102 , takeMVar
113 )
103 )
114−import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
104+import Control.Concurrent.Chan (newChan, writeChan)
115−import Control.Distributed.Process.Internal.CQueue ( CQueue
105+import Control.Distributed.Process.Internal.CQueue ( dequeueMatching
116− , dequeueMatching
117 , enqueue
106 , enqueue
118 , newCQueue
107 , newCQueue
119 )
108 )
120−import Control.Distributed.Process.Serializable ( Serializable
109+import Control.Distributed.Process.Serializable (Serializable, fingerprint)
121− , Fingerprint
122− , encodeFingerprint
123− , decodeFingerprint
124− , fingerprint
125− , sizeOfFingerprint
126− )
127import qualified Network.Transport as NT ( Transport
110import qualified Network.Transport as NT ( Transport
128 , EndPoint
111 , EndPoint
129− , EndPointAddress
130 , Connection
112 , Connection
131 , Reliability(ReliableOrdered)
113 , Reliability(ReliableOrdered)
132 , defaultConnectHints
114 , defaultConnectHints
@@ -142,65 +124,33 @@ import qualified Network.Transport as NT ( Transport
142 , closeEndPoint
124 , closeEndPoint
143 , ConnectionId
125 , ConnectionId
144 )
126 )
145−import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)
146−import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
127+import Data.Accessor ((^.), (^=), (^:))
147−import qualified Data.Accessor.Container as DAC (mapMaybe, mapDefault)
148import System.Random (randomIO)
128import System.Random (randomIO)
149−
129+import Control.Distributed.Process.Internal.NodeController (runNodeController)
150−-- | A local process ID consists of a seed which distinguishes processes from
130+import Control.Distributed.Process.Internal ( NodeId(..)
151−-- different instances of the same local node and a counter
152−data LocalProcessId = LocalProcessId
131+ , LocalProcessId(..)
153− { lpidUnique :: Int32
132+ , ProcessId(..)
154− , lpidCounter :: Int32
133+ , LocalNode(..)
155− }
134+ , LocalNodeState(..)
156− deriving (Eq, Ord, Typeable, Show)
135+ , LocalProcess(..)
157−
136+ , LocalProcessState(..)
158−-- We identify node IDs and endpoint IDs
159−newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress }
160− deriving (Show, Eq, Ord, Binary)
161−
162−-- | A process ID combines a local process with with an endpoint address
163−-- (in other words, we identify nodes and endpoints)
164−data ProcessId = ProcessId
165− { processNodeId :: NodeId
166− , processLocalId :: LocalProcessId
167− }
168− deriving (Eq, Ord, Typeable, Show)
169−
170−-- | Messages consist of their typeRep fingerprint and their encoding
171−data Message = Message
137+ , Message(..)
172− { messageFingerprint :: Fingerprint
138+ , MonitorRef
173− , messageEncoding :: BSL.ByteString
139+ , MonitorReply(..)
174− }
140+ , DiedReason(..)
175−
141+ , NCMsg(..)
176−-- | Local nodes
142+ , ProcessSignal(..)
177−data LocalNode = LocalNode
143+ , createMessage
178− { localNodeId :: NodeId
144+ , messageToPayload
179− , localEndPoint :: NT.EndPoint
145+ , payloadToMessage
180− , localState :: MVar LocalNodeState
146+ , payloadToId
181− , localCtrlChan :: Chan NCMsg
147+ , idToPayload
182− }
148+ , localPidCounter
183−
149+ , localPidUnique
184−-- | Local node state
150+ , localProcessWithId
185−data LocalNodeState = LocalNodeState
151+ , localMonitorCounter
186− { _localProcesses :: Map LocalProcessId LocalProcess
152+ , connectionTo
187− , _localPidCounter :: Int32
153+ )
188− , _localPidUnique :: Int32
189− , _localMonitorCounter :: Int32
190− }
191−
192−-- | Processes running on our local node
193−data LocalProcess = LocalProcess
194− { processQueue :: CQueue Message
195− , processNode :: LocalNode
196− , processId :: ProcessId
197− , processState :: MVar LocalProcessState
198− }
199−
200−-- | Local process state
201−data LocalProcessState = LocalProcessState
202− { _connections :: Map ProcessId NT.Connection
203− }
204
154
205-- The Cloud Haskell 'Process' type
155-- The Cloud Haskell 'Process' type
206newtype Process a = Process { unProcess :: ReaderT LocalProcess IO a }
156newtype Process a = Process { unProcess :: ReaderT LocalProcess IO a }
@@ -269,209 +219,6 @@ expectTimeout timeout = do
269 dequeueMatching queue (Just timeout) ((== fp) . messageFingerprint)
219 dequeueMatching queue (Just timeout) ((== fp) . messageFingerprint)
270 return $ fmap (decode . messageEncoding) msg
220 return $ fmap (decode . messageEncoding) msg
271
221
272−--------------------------------------------------------------------------------
273−-- The node controller --
274−--------------------------------------------------------------------------------
275−
276−data ProcessSignal =
277− Monitor ProcessId MonitorRef
278− | Died Identifier DiedReason
279−
280−type MonitorRef = Int32
281−
282−data MonitorReply = ProcessDied MonitorRef ProcessId DiedReason
283− deriving (Typeable)
284−
285−data DiedReason =
286− DiedNormal
287− | DiedException String -- TODO: would prefer SomeException instead of String, but exceptions don't implement Binary
288− | DiedDisconnect
289− | DiedNodeDown
290− | DiedNoProc
291− deriving Show
292−
293−type Identifier = Either ProcessId NodeId
294−
295−data NCMsg = NCMsg
296− { ctrlMsgSender :: Identifier
297− , ctrlMsgSignal :: ProcessSignal
298− }
299−
300−data NCState = NCState
301− { -- Mapping from remote processes to linked local processes
302− _ncLinks :: Map NodeId (Map LocalProcessId (Set ProcessId))
303− -- Mapping from remote processes to monitoring local processes
304− , _ncMons :: Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
305− -- The node controller maintains its own set of connections
306− -- TODO: still not convinced that this is correct
307− , _ncConns :: Map Identifier NT.Connection
308− }
309−
310−destNid :: ProcessSignal -> Maybe NodeId
311−destNid (Monitor pid _) = Just $ processNodeId pid
312−destNid (Died (Right _) _) = Nothing
313−destNid (Died (Left _pid) _) = fail "destNid: TODO"
314−
315−initNCState :: NCState
316−initNCState = NCState
317− { _ncLinks = Map.empty
318− , _ncMons = Map.empty
319− , _ncConns = Map.empty
320− }
321−
322−newtype NC a = NC { unNC :: ReaderT LocalNode (StateT NCState IO) a }
323− deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)
324−
325−nodeController :: NC ()
326−nodeController = forever $ do
327− node <- ask
328− msg <- liftIO $ readChan (localCtrlChan node)
329−
330− -- Forward the message if appropriate
331− case destNid (ctrlMsgSignal msg) of
332− Just nid' | nid' /= localNodeId node -> ncSendCtrlMsg nid' msg
333− _ -> return ()
334−
335− ncEffect msg
336−
337−ctrlSendTo :: Identifier -> [BSS.ByteString] -> NC ()
338−ctrlSendTo them payload = do
339− mConn <- ncConnTo them
340− didSend <- case mConn of
341− Just conn -> do
342− didSend <- liftIO $ NT.send conn payload
343− case didSend of
344− Left _ -> return False
345− Right () -> return True
346− Nothing ->
347− return False
348− unless didSend $ do
349− -- [Unified: Table 9, rule node_disconnect]
350− node <- ask
351− liftIO . writeChan (localCtrlChan node) $ NCMsg
352− { ctrlMsgSender = them
353− , ctrlMsgSignal = Died them DiedDisconnect
354− }
355−
356−ncSendCtrlMsg :: NodeId -> NCMsg -> NC ()
357−ncSendCtrlMsg dest = ctrlSendTo (Right dest) . BSL.toChunks . encode
358−
359−ncSendLocal :: LocalProcessId -> Message -> NC ()
360−ncSendLocal lpid msg = do
361− node <- ask
362− liftIO $ do
363− mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
364− -- By [Unified: table 6, rule missing_process] messages to dead processes
365− -- can silently be dropped
366− forM_ mProc $ \proc -> enqueue (processQueue proc) msg
367−
368−ncConnTo :: Identifier -> NC (Maybe NT.Connection)
369−ncConnTo them = do
370− mConn <- gets (^. ncConnFor them)
371− case mConn of
372− Just conn -> return (Just conn)
373− Nothing -> ncCreateConnTo them
374−
375−ncCreateConnTo :: Identifier -> NC (Maybe NT.Connection)
376−ncCreateConnTo them = do
377− node <- ask
378− mConn <- liftIO $ NT.connect (localEndPoint node)
379− addr
380− NT.ReliableOrdered
381− NT.defaultConnectHints
382− case mConn of
383− Right conn -> do
384− didSend <- liftIO $ NT.send conn firstMsg
385− case didSend of
386− Left _ ->
387− return Nothing
388− Right () -> do
389− modify $ ncConnFor them ^= Just conn
390− return $ Just conn
391− Left _ ->
392− return Nothing
393− where
394− (addr, firstMsg) = case them of
395− Left pid -> ( nodeAddress (processNodeId pid)
396− , idToPayload (Just $ processLocalId pid)
397− )
398− Right nid -> ( nodeAddress nid
399− , idToPayload Nothing
400− )
401−
402−
403−-- [Unified: ncEffect]
404−ncEffect :: NCMsg -> NC ()
405−
406−-- [Unified: Table 10]
407−ncEffect (NCMsg (Left from) (Monitor them ref)) = do
408− node <- ask
409− shouldLink <-
410− if processNodeId them /= localNodeId node
411− then return True
412− else liftIO . withMVar (localState node) $
413− return . isJust . (^. localProcessWithId (processLocalId them))
414− let localFrom = processNodeId from == localNodeId node
415− case (shouldLink, localFrom) of
416− (True, _) -> -- [Unified: first rule]
417− modify $ ncMonsFor them from ^: Set.insert ref
418− (False, True) -> -- [Unified: second rule]
419− ncSendLocal (processLocalId from) . createMessage $
420− ProcessDied ref them DiedNoProc
421− (False, False) -> -- [Unified: third rule]
422− ncSendCtrlMsg (processNodeId from) NCMsg
423− { ctrlMsgSender = Right (localNodeId node)
424− , ctrlMsgSignal = Died (Left them) DiedNoProc
425− }
426−
427−ncEffect (NCMsg (Right _) (Monitor _ _)) =
428− error "Monitor message from a node?"
429−
430−-- [Unified: Table 12, bottom rule]
431−ncEffect (NCMsg _from (Died (Right nid) reason)) = do
432− node <- ask
433− links <- gets (^. ncLinksForNode nid)
434− mons <- gets (^. ncMonsForNode nid)
435−
436− forM_ (Map.toList links) $ \(_them, _uss) ->
437− fail "ncEffect: linking not implemented"
438−
439− forM_ (Map.toList mons) $ \(them, uss) ->
440− forM_ (Map.toList uss) $ \(us, refs) ->
441− -- We only need to notify local processes
442− when (processNodeId us == localNodeId node) $ do
443− let lpid = processLocalId us
444− rpid = ProcessId nid them
445− forM_ refs $ \ref ->
446− ncSendLocal lpid . createMessage $ ProcessDied ref rpid reason
447−
448− modify $ (ncLinksForNode nid ^= Map.empty)
449− . (ncMonsForNode nid ^= Map.empty)
450−
451−-- [Unified: Table 12, top rule]
452−ncEffect (NCMsg _from (Died (Left pid) reason)) = do
453− node <- ask
454− links <- gets (^. ncLinksForProcess pid)
455− mons <- gets (^. ncMonsForProcess pid)
456−
457− forM_ links $ \_us ->
458− fail "ncEffect: linking not implemented"
459−
460− forM_ (Map.toList mons) $ \(us, refs) ->
461− forM_ refs $ \ref -> do
462− let msg = createMessage $ ProcessDied ref pid reason
463− if processNodeId us == localNodeId node
464− then
465− ncSendLocal (processLocalId us) msg
466− else
467− ncSendCtrlMsg (processNodeId us) NCMsg
468− { ctrlMsgSender = Right (localNodeId node) -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
469− , ctrlMsgSignal = Died (Left pid) reason
470− }
471−
472− modify $ (ncLinksForProcess pid ^= Set.empty)
473− . (ncMonsForProcess pid ^= Map.empty)
474−
475--------------------------------------------------------------------------------
222--------------------------------------------------------------------------------
476-- Initialization --
223-- Initialization --
477--------------------------------------------------------------------------------
224--------------------------------------------------------------------------------
@@ -495,8 +242,7 @@ newLocalNode transport = do
495 , localState = state
242 , localState = state
496 , localCtrlChan = ctrlChan
243 , localCtrlChan = ctrlChan
497 }
244 }
498− void . forkIO $ evalStateT (runReaderT (unNC nodeController) node)
245+ void . forkIO $ runNodeController node
499− initNCState
500 void . forkIO $ handleIncomingMessages node
246 void . forkIO $ handleIncomingMessages node
501 return node
247 return node
502
248
@@ -658,37 +404,6 @@ sendBinary them conn payload = do
658sendMessage :: ProcessId -> NT.Connection -> Message -> Process ()
404sendMessage :: ProcessId -> NT.Connection -> Message -> Process ()
659sendMessage them conn = sendBinary them conn . messageToPayload
405sendMessage them conn = sendBinary them conn . messageToPayload
660
406
661−messageToPayload :: Message -> [BSS.ByteString]
662−messageToPayload (Message fp enc) = encodeFingerprint fp : BSL.toChunks enc
663−
664−payloadToMessage :: [BSS.ByteString] -> Message
665−payloadToMessage payload = Message fp msg
666− where
667− (encFp, msg) = BSL.splitAt (fromIntegral sizeOfFingerprint)
668− $ BSL.fromChunks payload
669− fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp
670−
671−-- | The first message we send across a connection to indicate the intended
672−-- recipient. Pass Nothing for the remote node controller
673−idToPayload :: Maybe LocalProcessId -> [BSS.ByteString]
674−idToPayload Nothing = [ NTI.encodeInt32 (0 :: Int) ]
675−idToPayload (Just lpid) = [ NTI.encodeInt32 (1 :: Int)
676− , NTI.encodeInt32 (lpidCounter lpid)
677− , NTI.encodeInt32 (lpidUnique lpid)
678− ]
679−
680−-- | Inverse of 'idToPayload'
681−payloadToId :: [BSS.ByteString] -> Maybe LocalProcessId
682−payloadToId bss = let (bs1, bss') = BSS.splitAt 4 . BSS.concat $ bss
683− (bs2, bs3) = BSS.splitAt 4 bss' in
684− case NTI.decodeInt32 bs1 :: Int of
685− 0 -> Nothing
686− 1 -> Just LocalProcessId
687− { lpidCounter = NTI.decodeInt32 bs2
688− , lpidUnique = NTI.decodeInt32 bs3
689− }
690− _ -> fail "payloadToId"
691−
692remoteProcessFailed :: LocalNode -> ProcessId -> IO ()
407remoteProcessFailed :: LocalNode -> ProcessId -> IO ()
693remoteProcessFailed node them = do
408remoteProcessFailed node them = do
694 -- [Unified: Table 9 rule node_disconnect]
409 -- [Unified: Table 9 rule node_disconnect]
@@ -698,9 +413,6 @@ remoteProcessFailed node them = do
698 , ctrlMsgSignal = Died nid DiedDisconnect
413 , ctrlMsgSignal = Died nid DiedDisconnect
699 }
414 }
700
415
701−createMessage :: Serializable a => a -> Message
702−createMessage a = Message (fingerprint a) (encode a)
703−
704getMonitorRef :: Process MonitorRef
416getMonitorRef :: Process MonitorRef
705getMonitorRef = do
417getMonitorRef = do
706 node <- processNode <$> ask
418 node <- processNode <$> ask
@@ -712,101 +424,3 @@ getMonitorRef = do
712-- This is most definitely NOT exported
424-- This is most definitely NOT exported
713runLocalProcess :: LocalProcess -> Process a -> IO a
425runLocalProcess :: LocalProcess -> Process a -> IO a
714runLocalProcess lproc proc = runReaderT (unProcess proc) lproc
426runLocalProcess lproc proc = runReaderT (unProcess proc) lproc
715−
716−--------------------------------------------------------------------------------
717−-- Binary instances --
718−--------------------------------------------------------------------------------
719−
720−instance Binary LocalProcessId where
721− put lpid = put (lpidUnique lpid) >> put (lpidCounter lpid)
722− get = LocalProcessId <$> get <*> get
723−
724−instance Binary ProcessId where
725− put pid = put (processNodeId pid) >> put (processLocalId pid)
726− get = ProcessId <$> get <*> get
727−
728−instance Binary MonitorReply where
729− put (ProcessDied ref pid reason) = put ref >> put pid >> put reason
730− get = ProcessDied <$> get <*> get <*> get
731−
732−instance Binary NCMsg where
733− put msg = put (ctrlMsgSender msg) >> put (ctrlMsgSignal msg)
734− get = NCMsg <$> get <*> get
735−
736−instance Binary ProcessSignal where
737− put (Monitor pid ref) = putWord8 0 >> put pid >> put ref
738− put (Died who reason) = putWord8 1 >> put who >> put reason
739− get = do
740− header <- getWord8
741− case header of
742− 0 -> Monitor <$> get <*> get
743− 1 -> Died <$> get <*> get
744− _ -> fail "ProcessSignal.get: invalid"
745−
746−instance Binary DiedReason where
747− put DiedNormal = putWord8 0
748− put (DiedException e) = putWord8 1 >> put e
749− put DiedDisconnect = putWord8 2
750− put DiedNodeDown = putWord8 3
751− put DiedNoProc = putWord8 4
752− get = do
753− header <- getWord8
754− case header of
755− 0 -> return DiedNormal
756− 1 -> DiedException <$> get
757− 2 -> return DiedDisconnect
758− 3 -> return DiedNodeDown
759− 4 -> return DiedNoProc
760− _ -> fail "DiedReason.get: invalid"
761−
762−--------------------------------------------------------------------------------
763−-- Accessors --
764−--------------------------------------------------------------------------------
765−
766−localProcesses :: Accessor LocalNodeState (Map LocalProcessId LocalProcess)
767−localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })
768−
769−localPidCounter :: Accessor LocalNodeState Int32
770−localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter = ctr })
771−
772−localPidUnique :: Accessor LocalNodeState Int32
773−localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })
774−
775−localMonitorCounter :: Accessor LocalNodeState Int32
776−localMonitorCounter = accessor _localMonitorCounter (\ctr st -> st { _localMonitorCounter = ctr })
777−
778−localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)
779−localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid
780−
781−connections :: Accessor LocalProcessState (Map ProcessId NT.Connection)
782−connections = accessor _connections (\conns st -> st { _connections = conns })
783−
784−connectionTo :: ProcessId -> Accessor LocalProcessState (Maybe NT.Connection)
785−connectionTo pid = connections >>> DAC.mapMaybe pid
786−
787−ncLinks :: Accessor NCState (Map NodeId (Map LocalProcessId (Set ProcessId)))
788−ncLinks = accessor _ncLinks (\links st -> st { _ncLinks = links })
789−
790−ncMons :: Accessor NCState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))
791−ncMons = accessor _ncMons (\mons st -> st { _ncMons = mons })
792−
793−ncConns :: Accessor NCState (Map Identifier NT.Connection)
794−ncConns = accessor _ncConns (\conns st -> st { _ncConns = conns })
795−
796−ncConnFor :: Identifier -> Accessor NCState (Maybe NT.Connection)
797−ncConnFor them = ncConns >>> DAC.mapMaybe them
798−
799−ncLinksForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Set ProcessId))
800−ncLinksForNode nid = ncLinks >>> DAC.mapDefault Map.empty nid
801−
802−ncMonsForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
803−ncMonsForNode nid = ncMons >>> DAC.mapDefault Map.empty nid
804−
805−ncLinksForProcess :: ProcessId -> Accessor NCState (Set ProcessId)
806−ncLinksForProcess pid = ncLinksForNode (processNodeId pid) >>> DAC.mapDefault Set.empty (processLocalId pid)
807−
808−ncMonsForProcess :: ProcessId -> Accessor NCState (Map ProcessId (Set MonitorRef))
809−ncMonsForProcess pid = ncMonsForNode (processNodeId pid) >>> DAC.mapDefault Map.empty (processLocalId pid)
810−
811−ncMonsFor :: ProcessId -> ProcessId -> Accessor NCState (Set MonitorRef)
812−ncMonsFor them us = ncMonsForProcess them >>> DAC.mapDefault Set.empty us
distributed-process/src/Control/Distributed/Process/Internal.hsnew file+252
@@ -0,0 +1,252 @@
1+module Control.Distributed.Process.Internal
2+ ( -- * Node and process identifiers
3+ NodeId(..)
4+ , LocalProcessId(..)
5+ , ProcessId(..)
6+ -- * Local nodes and processes
7+ , LocalNode(..)
8+ , LocalNodeState(..)
9+ , LocalProcess(..)
10+ , LocalProcessState(..)
11+ -- * Messages
12+ , Message(..)
13+ -- * Monitoring
14+ , MonitorRef
15+ , MonitorReply(..)
16+ , DiedReason(..)
17+ -- * Node controller data types
18+ , Identifier
19+ , NCMsg(..)
20+ , ProcessSignal(..)
21+ -- * Serialization/deserialization
22+ , createMessage
23+ , messageToPayload
24+ , payloadToMessage
25+ , idToPayload
26+ , payloadToId
27+ -- * Accessors
28+ , localProcesses
29+ , localPidCounter
30+ , localPidUnique
31+ , localMonitorCounter
32+ , localProcessWithId
33+ , connections
34+ , connectionTo
35+ ) where
36+
37+import Control.Concurrent.MVar (MVar)
38+import Control.Concurrent.Chan (Chan)
39+import Control.Category ((>>>))
40+import Data.Map (Map)
41+import Data.Int (Int32)
42+import Data.Typeable (Typeable)
43+import Data.Binary (Binary, encode, put, get, putWord8, getWord8)
44+import Data.Accessor (Accessor, accessor)
45+import qualified Data.Accessor.Container as DAC (mapMaybe)
46+import qualified Data.ByteString.Lazy as BSL (ByteString, toChunks, fromChunks, splitAt)
47+import qualified Data.ByteString as BSS (ByteString, concat, splitAt)
48+import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)
49+import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)
50+import Control.Applicative ((<$>), (<*>))
51+import Control.Distributed.Process.Serializable ( Fingerprint
52+ , Serializable
53+ , encodeFingerprint
54+ , decodeFingerprint
55+ , fingerprint
56+ , sizeOfFingerprint
57+ )
58+import Control.Distributed.Process.Internal.CQueue (CQueue)
59+
60+-- We identify node IDs and endpoint IDs
61+newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress }
62+ deriving (Show, Eq, Ord, Binary)
63+
64+-- | A local process ID consists of a seed which distinguishes processes from
65+-- different instances of the same local node and a counter
66+data LocalProcessId = LocalProcessId
67+ { lpidUnique :: Int32
68+ , lpidCounter :: Int32
69+ }
70+ deriving (Eq, Ord, Typeable, Show)
71+
72+-- | A process ID combines a local process with with an endpoint address
73+-- (in other words, we identify nodes and endpoints)
74+data ProcessId = ProcessId
75+ { processNodeId :: NodeId
76+ , processLocalId :: LocalProcessId
77+ }
78+ deriving (Eq, Ord, Typeable, Show)
79+
80+-- | Local nodes
81+data LocalNode = LocalNode
82+ { localNodeId :: NodeId
83+ , localEndPoint :: NT.EndPoint
84+ , localState :: MVar LocalNodeState
85+ , localCtrlChan :: Chan NCMsg
86+ }
87+
88+-- | Local node state
89+data LocalNodeState = LocalNodeState
90+ { _localProcesses :: Map LocalProcessId LocalProcess
91+ , _localPidCounter :: Int32
92+ , _localPidUnique :: Int32
93+ , _localMonitorCounter :: Int32
94+ }
95+
96+-- | Processes running on our local node
97+data LocalProcess = LocalProcess
98+ { processQueue :: CQueue Message
99+ , processNode :: LocalNode
100+ , processId :: ProcessId
101+ , processState :: MVar LocalProcessState
102+ }
103+
104+-- | Local process state
105+data LocalProcessState = LocalProcessState
106+ { _connections :: Map ProcessId NT.Connection
107+ }
108+
109+-- | Messages consist of their typeRep fingerprint and their encoding
110+data Message = Message
111+ { messageFingerprint :: Fingerprint
112+ , messageEncoding :: BSL.ByteString
113+ }
114+
115+-- | MonitorRef should be an opaque type
116+type MonitorRef = Int32
117+
118+-- | Messages sent by monitors
119+data MonitorReply = ProcessDied MonitorRef ProcessId DiedReason
120+ deriving (Typeable)
121+
122+-- | Why did a process die?
123+data DiedReason =
124+ DiedNormal
125+ | DiedException String -- TODO: would prefer SomeException instead of String, but exceptions don't implement Binary
126+ | DiedDisconnect
127+ | DiedNodeDown
128+ | DiedNoProc
129+ deriving Show
130+
131+-- | Node or process identifier
132+type Identifier = Either ProcessId NodeId
133+
134+-- | Messages to the node controller
135+data NCMsg = NCMsg
136+ { ctrlMsgSender :: Identifier
137+ , ctrlMsgSignal :: ProcessSignal
138+ }
139+
140+-- | Signals to the node controller (see 'NCMsg')
141+data ProcessSignal =
142+ Monitor ProcessId MonitorRef
143+ | Died Identifier DiedReason
144+
145+--------------------------------------------------------------------------------
146+-- Serialization/deserialization --
147+--------------------------------------------------------------------------------
148+
149+-- | Turn any serialiable term into a message
150+createMessage :: Serializable a => a -> Message
151+createMessage a = Message (fingerprint a) (encode a)
152+
153+-- | Serialize a message
154+messageToPayload :: Message -> [BSS.ByteString]
155+messageToPayload (Message fp enc) = encodeFingerprint fp : BSL.toChunks enc
156+
157+-- | Deserialize a message
158+payloadToMessage :: [BSS.ByteString] -> Message
159+payloadToMessage payload = Message fp msg
160+ where
161+ (encFp, msg) = BSL.splitAt (fromIntegral sizeOfFingerprint)
162+ $ BSL.fromChunks payload
163+ fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp
164+
165+
166+-- | The first message we send across a connection to indicate the intended
167+-- recipient. Pass Nothing for the remote node controller
168+idToPayload :: Maybe LocalProcessId -> [BSS.ByteString]
169+idToPayload Nothing = [ NTI.encodeInt32 (0 :: Int) ]
170+idToPayload (Just lpid) = [ NTI.encodeInt32 (1 :: Int)
171+ , NTI.encodeInt32 (lpidCounter lpid)
172+ , NTI.encodeInt32 (lpidUnique lpid)
173+ ]
174+
175+-- | Inverse of 'idToPayload'
176+payloadToId :: [BSS.ByteString] -> Maybe LocalProcessId
177+payloadToId bss = let (bs1, bss') = BSS.splitAt 4 . BSS.concat $ bss
178+ (bs2, bs3) = BSS.splitAt 4 bss' in
179+ case NTI.decodeInt32 bs1 :: Int of
180+ 0 -> Nothing
181+ 1 -> Just LocalProcessId
182+ { lpidCounter = NTI.decodeInt32 bs2
183+ , lpidUnique = NTI.decodeInt32 bs3
184+ }
185+ _ -> fail "payloadToId"
186+
187+instance Binary LocalProcessId where
188+ put lpid = put (lpidUnique lpid) >> put (lpidCounter lpid)
189+ get = LocalProcessId <$> get <*> get
190+
191+instance Binary ProcessId where
192+ put pid = put (processNodeId pid) >> put (processLocalId pid)
193+ get = ProcessId <$> get <*> get
194+
195+instance Binary MonitorReply where
196+ put (ProcessDied ref pid reason) = put ref >> put pid >> put reason
197+ get = ProcessDied <$> get <*> get <*> get
198+
199+instance Binary NCMsg where
200+ put msg = put (ctrlMsgSender msg) >> put (ctrlMsgSignal msg)
201+ get = NCMsg <$> get <*> get
202+
203+instance Binary ProcessSignal where
204+ put (Monitor pid ref) = putWord8 0 >> put pid >> put ref
205+ put (Died who reason) = putWord8 1 >> put who >> put reason
206+ get = do
207+ header <- getWord8
208+ case header of
209+ 0 -> Monitor <$> get <*> get
210+ 1 -> Died <$> get <*> get
211+ _ -> fail "ProcessSignal.get: invalid"
212+
213+instance Binary DiedReason where
214+ put DiedNormal = putWord8 0
215+ put (DiedException e) = putWord8 1 >> put e
216+ put DiedDisconnect = putWord8 2
217+ put DiedNodeDown = putWord8 3
218+ put DiedNoProc = putWord8 4
219+ get = do
220+ header <- getWord8
221+ case header of
222+ 0 -> return DiedNormal
223+ 1 -> DiedException <$> get
224+ 2 -> return DiedDisconnect
225+ 3 -> return DiedNodeDown
226+ 4 -> return DiedNoProc
227+ _ -> fail "DiedReason.get: invalid"
228+
229+--------------------------------------------------------------------------------
230+-- Accessors --
231+--------------------------------------------------------------------------------
232+
233+localProcesses :: Accessor LocalNodeState (Map LocalProcessId LocalProcess)
234+localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })
235+
236+localPidCounter :: Accessor LocalNodeState Int32
237+localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter = ctr })
238+
239+localPidUnique :: Accessor LocalNodeState Int32
240+localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })
241+
242+localMonitorCounter :: Accessor LocalNodeState Int32
243+localMonitorCounter = accessor _localMonitorCounter (\ctr st -> st { _localMonitorCounter = ctr })
244+
245+localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)
246+localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid
247+
248+connections :: Accessor LocalProcessState (Map ProcessId NT.Connection)
249+connections = accessor _connections (\conns st -> st { _connections = conns })
250+
251+connectionTo :: ProcessId -> Accessor LocalProcessState (Maybe NT.Connection)
252+connectionTo pid = connections >>> DAC.mapMaybe pid
distributed-process/src/Control/Distributed/Process/Internal/NodeController.hsnew file+262
@@ -0,0 +1,262 @@
1+module Control.Distributed.Process.Internal.NodeController
2+ ( runNodeController
3+ , NCMsg(..)
4+ ) where
5+
6+import Data.Map (Map)
7+import qualified Data.Map as Map (empty, toList)
8+import Data.Set (Set)
9+import qualified Data.Set as Set (empty, insert)
10+import qualified Data.ByteString as BSS (ByteString)
11+import qualified Data.ByteString.Lazy as BSL (toChunks)
12+import Data.Binary (encode)
13+import Data.Foldable (forM_)
14+import Data.Maybe (isJust)
15+import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
16+import qualified Data.Accessor.Container as DAC (mapMaybe, mapDefault)
17+import Control.Category ((>>>))
18+import Control.Monad (when, unless, forever)
19+import Control.Monad.IO.Class (MonadIO, liftIO)
20+import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)
21+import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
22+import Control.Concurrent.MVar (withMVar)
23+import Control.Concurrent.Chan (readChan, writeChan)
24+import qualified Network.Transport as NT ( EndPoint
25+ , Connection
26+ , Reliability(ReliableOrdered)
27+ , defaultConnectHints
28+ , send
29+ , connect
30+ )
31+import Control.Distributed.Process.Internal.CQueue (enqueue)
32+import Control.Distributed.Process.Internal ( NodeId(..)
33+ , LocalProcessId(..)
34+ , ProcessId(..)
35+ , LocalNode(..)
36+ , LocalProcess(..)
37+ , Message(..)
38+ , MonitorRef
39+ , MonitorReply(..)
40+ , DiedReason(..)
41+ , Identifier
42+ , NCMsg(..)
43+ , ProcessSignal(..)
44+ , createMessage
45+ , idToPayload
46+ , localProcessWithId
47+ )
48+
49+--------------------------------------------------------------------------------
50+-- Top-level access to the node controller --
51+--------------------------------------------------------------------------------
52+
53+runNodeController :: LocalNode -> IO ()
54+runNodeController node =
55+ evalStateT (runReaderT (unNC nodeController) node) initNCState
56+
57+
58+data NCState = NCState
59+ { -- Mapping from remote processes to linked local processes
60+ _ncLinks :: Map NodeId (Map LocalProcessId (Set ProcessId))
61+ -- Mapping from remote processes to monitoring local processes
62+ , _ncMons :: Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
63+ -- The node controller maintains its own set of connections
64+ -- TODO: still not convinced that this is correct
65+ , _ncConns :: Map Identifier NT.Connection
66+ }
67+
68+destNid :: ProcessSignal -> Maybe NodeId
69+destNid (Monitor pid _) = Just $ processNodeId pid
70+destNid (Died (Right _) _) = Nothing
71+destNid (Died (Left _pid) _) = fail "destNid: TODO"
72+
73+initNCState :: NCState
74+initNCState = NCState
75+ { _ncLinks = Map.empty
76+ , _ncMons = Map.empty
77+ , _ncConns = Map.empty
78+ }
79+
80+newtype NC a = NC { unNC :: ReaderT LocalNode (StateT NCState IO) a }
81+ deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)
82+
83+nodeController :: NC ()
84+nodeController = forever $ do
85+ node <- ask
86+ msg <- liftIO $ readChan (localCtrlChan node)
87+
88+ -- Forward the message if appropriate
89+ case destNid (ctrlMsgSignal msg) of
90+ Just nid' | nid' /= localNodeId node -> ncSendCtrlMsg nid' msg
91+ _ -> return ()
92+
93+ ncEffect msg
94+
95+ctrlSendTo :: Identifier -> [BSS.ByteString] -> NC ()
96+ctrlSendTo them payload = do
97+ mConn <- ncConnTo them
98+ didSend <- case mConn of
99+ Just conn -> do
100+ didSend <- liftIO $ NT.send conn payload
101+ case didSend of
102+ Left _ -> return False
103+ Right () -> return True
104+ Nothing ->
105+ return False
106+ unless didSend $ do
107+ -- [Unified: Table 9, rule node_disconnect]
108+ node <- ask
109+ liftIO . writeChan (localCtrlChan node) $ NCMsg
110+ { ctrlMsgSender = them
111+ , ctrlMsgSignal = Died them DiedDisconnect
112+ }
113+
114+ncSendCtrlMsg :: NodeId -> NCMsg -> NC ()
115+ncSendCtrlMsg dest = ctrlSendTo (Right dest) . BSL.toChunks . encode
116+
117+ncSendLocal :: LocalProcessId -> Message -> NC ()
118+ncSendLocal lpid msg = do
119+ node <- ask
120+ liftIO $ do
121+ mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
122+ -- By [Unified: table 6, rule missing_process] messages to dead processes
123+ -- can silently be dropped
124+ forM_ mProc $ \proc -> enqueue (processQueue proc) msg
125+
126+ncConnTo :: Identifier -> NC (Maybe NT.Connection)
127+ncConnTo them = do
128+ mConn <- gets (^. ncConnFor them)
129+ case mConn of
130+ Just conn -> return (Just conn)
131+ Nothing -> ncCreateConnTo them
132+
133+ncCreateConnTo :: Identifier -> NC (Maybe NT.Connection)
134+ncCreateConnTo them = do
135+ node <- ask
136+ mConn <- liftIO $ NT.connect (localEndPoint node)
137+ addr
138+ NT.ReliableOrdered
139+ NT.defaultConnectHints
140+ case mConn of
141+ Right conn -> do
142+ didSend <- liftIO $ NT.send conn firstMsg
143+ case didSend of
144+ Left _ ->
145+ return Nothing
146+ Right () -> do
147+ modify $ ncConnFor them ^= Just conn
148+ return $ Just conn
149+ Left _ ->
150+ return Nothing
151+ where
152+ (addr, firstMsg) = case them of
153+ Left pid -> ( nodeAddress (processNodeId pid)
154+ , idToPayload (Just $ processLocalId pid)
155+ )
156+ Right nid -> ( nodeAddress nid
157+ , idToPayload Nothing
158+ )
159+
160+
161+-- [Unified: ncEffect]
162+ncEffect :: NCMsg -> NC ()
163+
164+-- [Unified: Table 10]
165+ncEffect (NCMsg (Left from) (Monitor them ref)) = do
166+ node <- ask
167+ shouldLink <-
168+ if processNodeId them /= localNodeId node
169+ then return True
170+ else liftIO . withMVar (localState node) $
171+ return . isJust . (^. localProcessWithId (processLocalId them))
172+ let localFrom = processNodeId from == localNodeId node
173+ case (shouldLink, localFrom) of
174+ (True, _) -> -- [Unified: first rule]
175+ modify $ ncMonsFor them from ^: Set.insert ref
176+ (False, True) -> -- [Unified: second rule]
177+ ncSendLocal (processLocalId from) . createMessage $
178+ ProcessDied ref them DiedNoProc
179+ (False, False) -> -- [Unified: third rule]
180+ ncSendCtrlMsg (processNodeId from) NCMsg
181+ { ctrlMsgSender = Right (localNodeId node)
182+ , ctrlMsgSignal = Died (Left them) DiedNoProc
183+ }
184+
185+ncEffect (NCMsg (Right _) (Monitor _ _)) =
186+ error "Monitor message from a node?"
187+
188+-- [Unified: Table 12, bottom rule]
189+ncEffect (NCMsg _from (Died (Right nid) reason)) = do
190+ node <- ask
191+ links <- gets (^. ncLinksForNode nid)
192+ mons <- gets (^. ncMonsForNode nid)
193+
194+ forM_ (Map.toList links) $ \(_them, _uss) ->
195+ fail "ncEffect: linking not implemented"
196+
197+ forM_ (Map.toList mons) $ \(them, uss) ->
198+ forM_ (Map.toList uss) $ \(us, refs) ->
199+ -- We only need to notify local processes
200+ when (processNodeId us == localNodeId node) $ do
201+ let lpid = processLocalId us
202+ rpid = ProcessId nid them
203+ forM_ refs $ \ref ->
204+ ncSendLocal lpid . createMessage $ ProcessDied ref rpid reason
205+
206+ modify $ (ncLinksForNode nid ^= Map.empty)
207+ . (ncMonsForNode nid ^= Map.empty)
208+
209+-- [Unified: Table 12, top rule]
210+ncEffect (NCMsg _from (Died (Left pid) reason)) = do
211+ node <- ask
212+ links <- gets (^. ncLinksForProcess pid)
213+ mons <- gets (^. ncMonsForProcess pid)
214+
215+ forM_ links $ \_us ->
216+ fail "ncEffect: linking not implemented"
217+
218+ forM_ (Map.toList mons) $ \(us, refs) ->
219+ forM_ refs $ \ref -> do
220+ let msg = createMessage $ ProcessDied ref pid reason
221+ if processNodeId us == localNodeId node
222+ then
223+ ncSendLocal (processLocalId us) msg
224+ else
225+ ncSendCtrlMsg (processNodeId us) NCMsg
226+ { ctrlMsgSender = Right (localNodeId node) -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
227+ , ctrlMsgSignal = Died (Left pid) reason
228+ }
229+
230+ modify $ (ncLinksForProcess pid ^= Set.empty)
231+ . (ncMonsForProcess pid ^= Map.empty)
232+
233+--------------------------------------------------------------------------------
234+-- Accessors --
235+--------------------------------------------------------------------------------
236+
237+ncLinks :: Accessor NCState (Map NodeId (Map LocalProcessId (Set ProcessId)))
238+ncLinks = accessor _ncLinks (\links st -> st { _ncLinks = links })
239+
240+ncMons :: Accessor NCState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))
241+ncMons = accessor _ncMons (\mons st -> st { _ncMons = mons })
242+
243+ncConns :: Accessor NCState (Map Identifier NT.Connection)
244+ncConns = accessor _ncConns (\conns st -> st { _ncConns = conns })
245+
246+ncConnFor :: Identifier -> Accessor NCState (Maybe NT.Connection)
247+ncConnFor them = ncConns >>> DAC.mapMaybe them
248+
249+ncLinksForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Set ProcessId))
250+ncLinksForNode nid = ncLinks >>> DAC.mapDefault Map.empty nid
251+
252+ncMonsForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
253+ncMonsForNode nid = ncMons >>> DAC.mapDefault Map.empty nid
254+
255+ncLinksForProcess :: ProcessId -> Accessor NCState (Set ProcessId)
256+ncLinksForProcess pid = ncLinksForNode (processNodeId pid) >>> DAC.mapDefault Set.empty (processLocalId pid)
257+
258+ncMonsForProcess :: ProcessId -> Accessor NCState (Map ProcessId (Set MonitorRef))
259+ncMonsForProcess pid = ncMonsForNode (processNodeId pid) >>> DAC.mapDefault Map.empty (processLocalId pid)
260+
261+ncMonsFor :: ProcessId -> ProcessId -> Accessor NCState (Set MonitorRef)
262+ncMonsFor them us = ncMonsForProcess them >>> DAC.mapDefault Set.empty us