@@ -78,32 +78,22 @@ module Control.Distributed.Process
7878 ) where
7979
8080import Prelude hiding (catch)
81−import qualified Data.ByteString as BSS (ByteString, concat, splitAt)
82−import qualified Data.ByteString.Lazy as BSL ( ByteString
83− , toChunks
84− , fromChunks
85− , splitAt
86− )
87−import Data.Binary (Binary, decode, encode, put, get, putWord8, getWord8)
81+import qualified Data.ByteString as BSS (ByteString)
82+import qualified Data.ByteString.Lazy as BSL (fromChunks)
83+import Data.Binary (decode)
8884import 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)
9086import qualified Data.List as List (delete)
9187import Data.Set (Set)
9288import qualified Data.Set as Set (empty, insert, delete, member)
93−import Data.Int (Int32)
94−import Data.Typeable (Typeable)
9589import Data.Foldable (forM_)
96−import Data.Maybe (isJust)
97−import Control.Monad (void, when, unless, forever)
90+import Control.Monad (void)
9891import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
9992import Control.Monad.IO.Class (MonadIO, liftIO)
100−import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)
101−import Control.Applicative ((<$>), (<*>))
102−import Control.Category ((>>>))
93+import Control.Applicative ((<$>))
10394import Control.Exception (Exception, throwIO, catch, SomeException)
10495import Control.Concurrent (forkIO)
105−import Control.Concurrent.MVar ( MVar
106− , newMVar
96+import Control.Concurrent.MVar ( newMVar
10797 , withMVar
10898 , modifyMVar
10999 , modifyMVar_
@@ -111,22 +101,14 @@ import Control.Concurrent.MVar ( MVar
111101 , putMVar
112102 , takeMVar
113103 )
114−import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
115−import Control.Distributed.Process.Internal.CQueue ( CQueue
116− , dequeueMatching
104+import Control.Concurrent.Chan (newChan, writeChan)
105+import Control.Distributed.Process.Internal.CQueue ( dequeueMatching
117106 , enqueue
118107 , newCQueue
119108 )
120−import Control.Distributed.Process.Serializable ( Serializable
121− , Fingerprint
122− , encodeFingerprint
123− , decodeFingerprint
124− , fingerprint
125− , sizeOfFingerprint
126− )
109+import Control.Distributed.Process.Serializable (Serializable, fingerprint)
127110import qualified Network.Transport as NT ( Transport
128111 , EndPoint
129− , EndPointAddress
130112 , Connection
131113 , Reliability(ReliableOrdered)
132114 , defaultConnectHints
@@ -142,65 +124,33 @@ import qualified Network.Transport as NT ( Transport
142124 , closeEndPoint
143125 , ConnectionId
144126 )
145−import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)
146−import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
147−import qualified Data.Accessor.Container as DAC (mapMaybe, mapDefault)
127+import Data.Accessor ((^.), (^=), (^:))
148128import System.Random (randomIO)
149−
150−-- | A local process ID consists of a seed which distinguishes processes from
151−-- different instances of the same local node and a counter
152−data LocalProcessId = LocalProcessId
153− { lpidUnique :: Int32
154− , lpidCounter :: Int32
155− }
156− deriving (Eq, Ord, Typeable, Show)
157−
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
172− { messageFingerprint :: Fingerprint
173− , messageEncoding :: BSL.ByteString
174− }
175−
176−-- | Local nodes
177−data LocalNode = LocalNode
178− { localNodeId :: NodeId
179− , localEndPoint :: NT.EndPoint
180− , localState :: MVar LocalNodeState
181− , localCtrlChan :: Chan NCMsg
182− }
183−
184−-- | Local node state
185−data LocalNodeState = LocalNodeState
186− { _localProcesses :: Map LocalProcessId LocalProcess
187− , _localPidCounter :: Int32
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− }
129+import Control.Distributed.Process.Internal.NodeController (runNodeController)
130+import Control.Distributed.Process.Internal ( NodeId(..)
131+ , LocalProcessId(..)
132+ , ProcessId(..)
133+ , LocalNode(..)
134+ , LocalNodeState(..)
135+ , LocalProcess(..)
136+ , LocalProcessState(..)
137+ , Message(..)
138+ , MonitorRef
139+ , MonitorReply(..)
140+ , DiedReason(..)
141+ , NCMsg(..)
142+ , ProcessSignal(..)
143+ , createMessage
144+ , messageToPayload
145+ , payloadToMessage
146+ , payloadToId
147+ , idToPayload
148+ , localPidCounter
149+ , localPidUnique
150+ , localProcessWithId
151+ , localMonitorCounter
152+ , connectionTo
153+ )
204154
205155-- The Cloud Haskell 'Process' type
206156newtype Process a = Process { unProcess :: ReaderT LocalProcess IO a }
@@ -269,209 +219,6 @@ expectTimeout timeout = do
269219 dequeueMatching queue (Just timeout) ((== fp) . messageFingerprint)
270220 return $ fmap (decode . messageEncoding) msg
271221
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−
475222--------------------------------------------------------------------------------
476223-- Initialization --
477224--------------------------------------------------------------------------------
@@ -495,8 +242,7 @@ newLocalNode transport = do
495242 , localState = state
496243 , localCtrlChan = ctrlChan
497244 }
498− void . forkIO $ evalStateT (runReaderT (unNC nodeController) node)
499− initNCState
245+ void . forkIO $ runNodeController node
500246 void . forkIO $ handleIncomingMessages node
501247 return node
502248
@@ -658,37 +404,6 @@ sendBinary them conn payload = do
658404sendMessage :: ProcessId -> NT.Connection -> Message -> Process ()
659405sendMessage them conn = sendBinary them conn . messageToPayload
660406
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−
692407remoteProcessFailed :: LocalNode -> ProcessId -> IO ()
693408remoteProcessFailed node them = do
694409 -- [Unified: Table 9 rule node_disconnect]
@@ -698,9 +413,6 @@ remoteProcessFailed node them = do
698413 , ctrlMsgSignal = Died nid DiedDisconnect
699414 }
700415
701−createMessage :: Serializable a => a -> Message
702−createMessage a = Message (fingerprint a) (encode a)
703−
704416getMonitorRef :: Process MonitorRef
705417getMonitorRef = do
706418 node <- processNode <$> ask
@@ -712,101 +424,3 @@ getMonitorRef = do
712424-- This is most definitely NOT exported
713425runLocalProcess :: LocalProcess -> Process a -> IO a
714426runLocalProcess 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