@@ -1,3 +1,54 @@
3+-- 1. 'send' never fails. If you want to know that the remote process received
4+-- your message, you will need to send an explicit acknowledgement. If you
5+-- want to know when the remote process failed, you will need to monitor
6+-- that remote process.
8+-- 2. 'send' may block (when the system TCP buffers are full, while we are
9+-- trying to establish a connection to the remote endpoint, etc.) but its
10+-- return does not imply that the remote process received the message (much
13+-- 3. Message delivery is reliable and ordered. That means that if process A
14+-- sends messages m1, m2, m3 to process B, B will either arrive all three
15+-- messages in order (m1, m2, m3) or a prefix thereof; messages will not be
16+-- 'missing' (m1, m3) or reordered (m1, m3, m2)
18+-- In order to guarantee (3), we stipulate that
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.
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.
31+-- Main reference for Cloud Haskell is
33+-- [1] "Towards Haskell in the Cloud", Jeff Epstein, Andrew Black and Simon
35+-- http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/remote.pdf
37+-- Some pointers to related documentation about Erlang, for comparison and
40+-- [1] "Programming Distributed Erlang Applications: Pitfalls and Recipes",
41+-- Hans Svensson and Lars-Ake Fredlund
42+-- http://man.lupaworld.com/content/develop/p37-svensson.pdf
43+-- [2] The Erlang manual, sections "Message Sending" and "Send"
44+-- http://www.erlang.org/doc/reference_manual/processes.html#id82409
45+-- http://www.erlang.org/doc/reference_manual/expressions.html#send
46+-- [3] Questions "Is the order of message reception guaranteed?" and
47+-- "If I send a message, is it guaranteed to reach the receiver?" of
49+-- http://www.erlang.org/faq/academic.html
50+-- [4] "Delivery of Messages", post on erlang-questions
51+-- http://erlang.org/pipermail/erlang-questions/2012-February/064767.html
1module Control.Distributed.Process
52module Control.Distributed.Process
2 ( -- * Cloud Haskell API
53 ( -- * Cloud Haskell API
@@ -11,7 +62,7 @@ module Control.Distributed.Process
11 , runProcess
62 , runProcess
14−import qualified Data.ByteString as BSS (ByteString, concat)
65+import qualified Data.ByteString as BSS (ByteString, concat, splitAt)
15import qualified Data.ByteString.Lazy as BSL ( ByteString
66import qualified Data.ByteString.Lazy as BSL ( ByteString
16 , toChunks
67 , toChunks
17 , fromChunks
68 , fromChunks
@@ -19,10 +70,9 @@ import qualified Data.ByteString.Lazy as BSL ( ByteString
20import Data.Binary (Binary, decode, encode, put, get)
71import Data.Binary (Binary, decode, encode, put, get)
21import Data.Map (Map)
72import Data.Map (Map)
22−import qualified Data.Map as Map (empty)
73+import qualified Data.Map as Map (empty, lookup, insert, delete)
23−import Data.IntMap (IntMap)
24−import qualified Data.IntMap as IntMap (empty, lookup, insert, delete)
25import qualified Data.List as List (delete)
74import qualified Data.List as List (delete)
75+import Data.Int (Int32)
26import Data.Typeable (Typeable)
76import Data.Typeable (Typeable)
27import Control.Monad (void, liftM2)
77import Control.Monad (void, liftM2)
28import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
78import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
@@ -68,9 +118,18 @@ import qualified Network.Transport as NT ( Transport
69import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)
119import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)
70import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
120import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
71−import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe)
121+import qualified Data.Accessor.Container as DAC (mapMaybe)
122+import System.Random (randomIO)
73−type LocalProcessId = Int
124+data LocalProcessId = LocalProcessId
125+ { pidUnique :: Int32
126+ , pidCounter :: Int32
128+ deriving (Eq, Ord, Typeable, Show)
130+instance Binary LocalProcessId where
131+ put lpid = put (pidUnique lpid) >> put (pidCounter lpid)
132+ get = liftM2 LocalProcessId get get
75data ProcessId = ProcessId
134data ProcessId = ProcessId
76 { processAddress :: NT.EndPointAddress
135 { processAddress :: NT.EndPointAddress
@@ -79,9 +138,8 @@ data ProcessId = ProcessId
79 deriving (Eq, Ord, Typeable, Show)
138 deriving (Eq, Ord, Typeable, Show)
81instance Binary ProcessId where
140instance Binary ProcessId where
82− put pid = do put (processAddress pid)
141+ put pid = put (processAddress pid) >> put (processLocalId pid)
83− put (processLocalId pid)
84− get = liftM2 ProcessId get get
142+ get = liftM2 ProcessId get get
86data Message = Message
144data Message = Message
87 { messageFingerprint :: Fingerprint
145 { messageFingerprint :: Fingerprint
@@ -95,8 +153,9 @@ data LocalNode = LocalNode
96data LocalNodeState = LocalNodeState
154data LocalNodeState = LocalNodeState
97 { _localConnections :: Map ProcessId NT.Connection
155 { _localConnections :: Map ProcessId NT.Connection
98− , _localProcesses :: IntMap ProcessState
156+ , _localProcesses :: Map LocalProcessId ProcessState
99− , _localNextProcId :: LocalProcessId
157+ , _localPidCounter :: Int32
158+ , _localPidUnique :: Int32
102data ProcessState = ProcessState
161data ProcessState = ProcessState
@@ -143,10 +202,12 @@ newLocalNode transport = do
143 case mEndPoint of
202 case mEndPoint of
144 Left ex -> throwIO ex
203 Left ex -> throwIO ex
145 Right endPoint -> do
204 Right endPoint -> do
146 state <- newMVar LocalNodeState
206 state <- newMVar LocalNodeState
147 { _localConnections = Map.empty
207 { _localConnections = Map.empty
148− , _localProcesses = IntMap.empty
208+ , _localProcesses = Map.empty
149− , _localNextProcId = firstNonReservedProcessId
209+ , _localPidCounter = 0
210+ , _localPidUnique = unq
151 let node = LocalNode { localEndPoint = endPoint
212 let node = LocalNode { localEndPoint = endPoint
152 , localState = state
213 , localState = state
@@ -164,7 +225,9 @@ forkProcess :: LocalNode -> Process () -> IO ProcessId
164forkProcess node proc = do
225forkProcess node proc = do
165 queue <- newCQueue
226 queue <- newCQueue
166 state <- modifyMVar (localState node) $ \st -> do
227 state <- modifyMVar (localState node) $ \st -> do
167− let lpid = st ^. localNextProcId
228+ let lpid = LocalProcessId { pidCounter = st ^. localPidCounter
229+ , pidUnique = st ^. localPidUnique
168 let pid = ProcessId { processAddress = NT.address (localEndPoint node)
231 let pid = ProcessId { processAddress = NT.address (localEndPoint node)
169 , processLocalId = lpid
232 , processLocalId = lpid
@@ -172,8 +235,9 @@ forkProcess node proc = do
172 , processNode = node
235 , processNode = node
173 , processId = pid
236 , processId = pid
238+ -- TODO: if the counter overflows we should pick a new unique
175 return ( (localProcessWithId lpid ^= Just state)
239 return ( (localProcessWithId lpid ^= Just state)
176− . (localNextProcId ^: (+ 1))
240+ . (localPidCounter ^: (+ 1))
@@ -181,7 +245,7 @@ forkProcess node proc = do
181 return (processId state)
245 return (processId state)
183handleIncomingMessages :: LocalNode -> IO ()
247handleIncomingMessages :: LocalNode -> IO ()
184−handleIncomingMessages node = go [] IntMap.empty
248+handleIncomingMessages node = go [] Map.empty
186 go halfOpenConns openConns = do
250 go halfOpenConns openConns = do
187 event <- NT.receive endpoint
251 event <- NT.receive endpoint
@@ -189,27 +253,27 @@ handleIncomingMessages node = go [] IntMap.empty
189 NT.ConnectionOpened cid _rel _theirAddr ->
253 NT.ConnectionOpened cid _rel _theirAddr ->
190 go (cid : halfOpenConns) openConns
254 go (cid : halfOpenConns) openConns
191 NT.Received cid payload ->
255 NT.Received cid payload ->
192− case IntMap.lookup cid openConns of
256+ case Map.lookup cid openConns of
193 Just proc -> do
257 Just proc -> do
194 let msg = payloadToMessage payload
258 let msg = payloadToMessage payload
195 enqueue (processQueue proc) msg
259 enqueue (processQueue proc) msg
196 go halfOpenConns openConns
260 go halfOpenConns openConns
197 Nothing -> if cid `elem` halfOpenConns
261 Nothing -> if cid `elem` halfOpenConns
199− let pid = NTI.decodeInt32 . BSS.concat $ payload
263+ let lpid = payloadToLpid payload
200− mProc <- withMVar state $ return . (^. localProcessWithId pid)
264+ mProc <- withMVar state $ return . (^. localProcessWithId lpid)
201 case mProc of
265 case mProc of
202 Just proc ->
266 Just proc ->
203− go (List.delete pid halfOpenConns)
267+ go (List.delete cid halfOpenConns)
204− (IntMap.insert cid proc openConns)
268+ (Map.insert cid proc openConns)
205 Nothing ->
269 Nothing ->
206− fail "TODO"
270+ fail "handleIncomingMessages: TODO 1"
208− fail "TODO"
272+ fail "handleIncomingMessages: TODO 2"
209 NT.ConnectionClosed cid ->
273 NT.ConnectionClosed cid ->
210− go (List.delete cid halfOpenConns) (IntMap.delete cid openConns)
274+ go (List.delete cid halfOpenConns) (Map.delete cid openConns)
211 NT.ErrorEvent _ ->
275 NT.ErrorEvent _ ->
212− fail "TODO"
276+ fail "handleIncomingMessages: TODO 3"
213 NT.EndPointClosed ->
277 NT.EndPointClosed ->
214 return ()
278 return ()
215 NT.ReceivedMulticast _ _ ->
279 NT.ReceivedMulticast _ _ ->
@@ -253,7 +317,7 @@ createConnectionTo node pid = do
253 NT.close conn
317 NT.close conn
254 return conn'
318 return conn'
255 Nothing -> do
319 Nothing -> do
256− sendBinary node pid conn [NTI.encodeInt32 . processLocalId $ pid]
320+ sendBinary node pid conn $ lpidToPayload (processLocalId pid)
257 return conn
321 return conn
258 Left err ->
322 Left err ->
259 throwIO err
323 throwIO err
@@ -279,13 +343,17 @@ payloadToMessage payload = Message fp msg
279 $ BSL.fromChunks payload
343 $ BSL.fromChunks payload
280 fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp
344 fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp
282−--------------------------------------------------------------------------------
346+lpidToPayload :: LocalProcessId -> [BSS.ByteString]
283−-- Constants --
347+lpidToPayload lpid = [ NTI.encodeInt32 (pidCounter lpid)
284−--------------------------------------------------------------------------------
348+ , NTI.encodeInt32 (pidUnique lpid)
286−firstNonReservedProcessId :: Int
350+
287−firstNonReservedProcessId = 1024
351+payloadToLpid :: [BSS.ByteString] -> LocalProcessId
288−
352+payloadToLpid bss = let (bs1, bs2) = BSS.splitAt 4 . BSS.concat $ bss
353+ in LocalProcessId { pidCounter = NTI.decodeInt32 bs1
354+ , pidUnique = NTI.decodeInt32 bs2
289--------------------------------------------------------------------------------
357--------------------------------------------------------------------------------
290-- Accessors --
358-- Accessors --
291--------------------------------------------------------------------------------
359--------------------------------------------------------------------------------
@@ -293,218 +361,17 @@ firstNonReservedProcessId = 1024
293localConnections :: Accessor LocalNodeState (Map ProcessId NT.Connection)
361localConnections :: Accessor LocalNodeState (Map ProcessId NT.Connection)
294localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })
362localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })
296−localProcesses :: Accessor LocalNodeState (IntMap ProcessState)
364+localProcesses :: Accessor LocalNodeState (Map LocalProcessId ProcessState)
297localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })
365localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })
367+localPidCounter :: Accessor LocalNodeState Int32
368+localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter = ctr })
299−localNextProcId :: Accessor LocalNodeState Int
370+localPidUnique :: Accessor LocalNodeState Int32
300−localNextProcId = accessor _localNextProcId (\pid st -> st { _localNextProcId = pid })
371+localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })
302localConnectionTo :: ProcessId -> Accessor LocalNodeState (Maybe NT.Connection)
373localConnectionTo :: ProcessId -> Accessor LocalNodeState (Maybe NT.Connection)
303localConnectionTo pid = localConnections >>> DAC.mapMaybe pid
374localConnectionTo pid = localConnections >>> DAC.mapMaybe pid
305localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe ProcessState)
376localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe ProcessState)
306−localProcessWithId lpid = localProcesses >>> DAC.intMapMaybe lpid
377+localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid
309−{-# LANGUAGE CPP, ExplicitForAll, ScopedTypeVariables #-}
311−module Control.Distributed.Process (
318− -- * Basic messaging
322− -- * Process management
335−import qualified Network.Transport as Trans
336−import Network.Transport (Transport)
338−import qualified Data.IntMap as IntMap
339−import Data.IntMap (IntMap)
340−import Control.Applicative
342−import Control.Monad.Trans
343−import Control.Concurrent.MVar
344−import Control.Concurrent.Chan
345−import Control.Concurrent
351−import qualified Data.ByteString.Char8 as BS
352−import Data.ByteString.Char8 (ByteString)
354−import qualified Data.ByteString.Lazy.Char8 as BS
355−import Data.ByteString.Lazy.Char8 (ByteString)
358−------------------------
359−-- Cloud Haskell layer
362−data NodeId = NodeId -- !Trans.SourceEnd !Trans.SourceEnd
364−data ProcessId = ProcessId !Trans.SourceEnd !NodeId !LocalProcessId
365−type LocalProcessId = Int
367−newtype SourcePort a = SourcePort Trans.SourceEnd
369−newtype Process a = Process { unProcess :: ProcessState -> IO a }
371−instance Functor Process where
372− fmap f m = Process (\ps -> unProcess m ps >>= \x -> return (f x))
374−instance Applicative Process where
378−instance Monad Process where
379− m >>= k = Process (\ps -> unProcess m ps >>= \x -> unProcess (k x) ps)
380− return x = Process (\_ -> return x)
382−instance MonadIO Process where
383− liftIO io = Process (\_ -> io)
385−getProcessState :: Process ProcessState
386−getProcessState = Process (\ps -> return ps)
388−getLocalNode :: Process LocalNode
389−getLocalNode = prNode <$> getProcessState
391−getSelfPid :: Process ProcessId
392−getSelfPid = prPid <$> getProcessState
394−-- State a process carries around and has local access to
396−data ProcessState = ProcessState {
397− prPid :: !ProcessId,
398− prChan :: !Trans.TargetEnd,
399− prQueue :: !(CQueue Message),
400− prNode :: !LocalNode
403−data Message = Message String --string rep of TypeRep, sigh.
406−-- Context for a local node, all processes running on a node have direct access to this
408−data LocalNode = LocalNode {
409− ndProcessTable :: !(MVar ProcessTable),
410− ndTransport :: !Transport
413−data ProcessTable = ProcessTable
414− !LocalProcessId -- ^ Value of next ProcessTableEntry index
415− !(IntMap ProcessTableEntry) -- ^ Index from LocalProcessIds to ProcessTableEntry
416−data ProcessTableEntry = ProcessTableEntry {
417− pteThread :: !ThreadId
420−newLocalNode :: Transport -> IO LocalNode
421−newLocalNode trans = do
422− processTableVar <- newMVar (ProcessTable 0 IntMap.empty)
424− ndProcessTable = processTableVar,
428−-- | `runProcess` executes a process on a given node, and waits for the process
429−-- to finish before returning.
430−runProcess :: LocalNode -> Process () -> IO ()
431−runProcess node proc = do
432− waitVar <- newEmptyMVar
433− _ <- forkProcess node (proc >> liftIO (putMVar waitVar ()))
434− --TODO: should use linking for waiting for the end
437−-- | `forkProcess` forks and executes process on a given node. This
438−-- returns the ProcessId when the new process has been created.
439−forkProcess :: LocalNode -> Process () -> IO ProcessId
440−forkProcess node (Process action) = do
441− (sourceAddr, chan) <- Trans.newConnection (ndTransport node)
442− sourceEnd <- Trans.connect sourceAddr
443− processTable@(ProcessTable lpid _) <- takeMVar (ndProcessTable node)
444− let pid = ProcessId sourceEnd NodeId lpid
447− putMVar (ndProcessTable node) (insertProcess tid processTable)
449− _ <- forkIO $ receiverPump chan queue
450− action $ ProcessState pid chan queue node
454− insertProcess :: ThreadId -> ProcessTable -> ProcessTable
455− insertProcess tid (ProcessTable nextPid table) =
456− let pte = ProcessTableEntry tid
457− in ProcessTable (nextPid+1) (IntMap.insert nextPid pte table)
459− receiverPump :: Trans.TargetEnd -> CQueue Message -> IO ()
460− receiverPump chan queue = forever $ do
461− msgBlobs <- Trans.receive chan
462− readBlobs (concatMap BS.unpack msgBlobs)
464− readBlobs :: String -> IO ()
465− readBlobs [] = return ()
467− -- TODO: replace reads with something safer and faster.
468− let [((typerep, body), bs')] = reads bs
469− enqueue queue (Message typerep body)
472−send :: (Typeable a, Show a) => ProcessId -> a -> Process ()
473−send (ProcessId chan _ _) msg =
474− liftIO (Trans.send chan msgBlobs)
476− msgBlobs :: [ByteString]
477− msgBlobs = [BS.pack (show (show (typeOf msg), show msg))]
479−expect :: forall a. (Typeable a, Read a) => Process a
481− ProcessState { prQueue = queue } <- getProcessState
482− let typerepstr = show (typeOf (undefined :: a))
483− Message _ body <- liftIO $
484− dequeueMatching queue (\(Message typerepstr' _) -> typerepstr' == typerepstr)
487−spawnLocal :: Process () -> Process ProcessId
489− node <- getLocalNode
490− liftIO $ forkProcess node proc
493−send :: Serializable a -> ProcessId -> a -> Process ()
494−expect :: Serializable a -> Process a
496−newChan :: Serializable a => Process (SourcePort a, TargetPort a)
497−sendChan :: Serializable a => SourcePort a -> a -> Process ()
498−receiveChan :: Serializable a => TargetPort a -> Process a
500−spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
501−terminate :: ProcessM a
502−getSelfPid :: ProcessM ProcessId
503−getSelfNode :: ProcessM NodeId
505−monitorProcess :: ProcessId -> ProcessId -> MonitorAction -> Process ()