@@ -1,3 +1,54 @@
1+-- | Cloud Haskell
2+--
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.
7+--
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
11+-- less processed it)
12+--
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)
17+--
18+-- In order to guarantee (3), we stipulate that
19+--
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.
23+--
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.
30+--
31+-- Main reference for Cloud Haskell is
32+--
33+-- [1] "Towards Haskell in the Cloud", Jeff Epstein, Andrew Black and Simon
34+-- Peyton-Jones.
35+-- http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/remote.pdf
36+--
37+-- Some pointers to related documentation about Erlang, for comparison and
38+-- inspiration:
39+--
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
48+-- the Erlang FAQ
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
152module Control.Distributed.Process
253 ( -- * Cloud Haskell API
354 ProcessId
@@ -11,7 +62,7 @@ module Control.Distributed.Process
1162 , runProcess
1263 ) where
1364
14−import qualified Data.ByteString as BSS (ByteString, concat)
65+import qualified Data.ByteString as BSS (ByteString, concat, splitAt)
1566import qualified Data.ByteString.Lazy as BSL ( ByteString
1667 , toChunks
1768 , fromChunks
@@ -19,10 +70,9 @@ import qualified Data.ByteString.Lazy as BSL ( ByteString
1970 )
2071import Data.Binary (Binary, decode, encode, put, get)
2172import Data.Map (Map)
22−import qualified Data.Map as Map (empty)
23−import Data.IntMap (IntMap)
24−import qualified Data.IntMap as IntMap (empty, lookup, insert, delete)
73+import qualified Data.Map as Map (empty, lookup, insert, delete)
2574import qualified Data.List as List (delete)
75+import Data.Int (Int32)
2676import Data.Typeable (Typeable)
2777import Control.Monad (void, liftM2)
2878import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
@@ -68,9 +118,18 @@ import qualified Network.Transport as NT ( Transport
68118 )
69119import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)
70120import 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)
72123
73−type LocalProcessId = Int
124+data LocalProcessId = LocalProcessId
125+ { pidUnique :: Int32
126+ , pidCounter :: Int32
127+ }
128+ deriving (Eq, Ord, Typeable, Show)
129+
130+instance Binary LocalProcessId where
131+ put lpid = put (pidUnique lpid) >> put (pidCounter lpid)
132+ get = liftM2 LocalProcessId get get
74133
75134data ProcessId = ProcessId
76135 { processAddress :: NT.EndPointAddress
@@ -79,9 +138,8 @@ data ProcessId = ProcessId
79138 deriving (Eq, Ord, Typeable, Show)
80139
81140instance Binary ProcessId where
82− put pid = do put (processAddress pid)
83− put (processLocalId pid)
84− get = liftM2 ProcessId get get
141+ put pid = put (processAddress pid) >> put (processLocalId pid)
142+ get = liftM2 ProcessId get get
85143
86144data Message = Message
87145 { messageFingerprint :: Fingerprint
@@ -95,8 +153,9 @@ data LocalNode = LocalNode
95153
96154data LocalNodeState = LocalNodeState
97155 { _localConnections :: Map ProcessId NT.Connection
98− , _localProcesses :: IntMap ProcessState
99− , _localNextProcId :: LocalProcessId
156+ , _localProcesses :: Map LocalProcessId ProcessState
157+ , _localPidCounter :: Int32
158+ , _localPidUnique :: Int32
100159 }
101160
102161data ProcessState = ProcessState
@@ -143,10 +202,12 @@ newLocalNode transport = do
143202 case mEndPoint of
144203 Left ex -> throwIO ex
145204 Right endPoint -> do
205+ unq <- randomIO
146206 state <- newMVar LocalNodeState
147207 { _localConnections = Map.empty
148− , _localProcesses = IntMap.empty
149− , _localNextProcId = firstNonReservedProcessId
208+ , _localProcesses = Map.empty
209+ , _localPidCounter = 0
210+ , _localPidUnique = unq
150211 }
151212 let node = LocalNode { localEndPoint = endPoint
152213 , localState = state
@@ -164,7 +225,9 @@ forkProcess :: LocalNode -> Process () -> IO ProcessId
164225forkProcess node proc = do
165226 queue <- newCQueue
166227 state <- modifyMVar (localState node) $ \st -> do
167− let lpid = st ^. localNextProcId
228+ let lpid = LocalProcessId { pidCounter = st ^. localPidCounter
229+ , pidUnique = st ^. localPidUnique
230+ }
168231 let pid = ProcessId { processAddress = NT.address (localEndPoint node)
169232 , processLocalId = lpid
170233 }
@@ -172,8 +235,9 @@ forkProcess node proc = do
172235 , processNode = node
173236 , processId = pid
174237 }
238+ -- TODO: if the counter overflows we should pick a new unique
175239 return ( (localProcessWithId lpid ^= Just state)
176− . (localNextProcId ^: (+ 1))
240+ . (localPidCounter ^: (+ 1))
177241 $ st
178242 , state
179243 )
@@ -181,7 +245,7 @@ forkProcess node proc = do
181245 return (processId state)
182246
183247handleIncomingMessages :: LocalNode -> IO ()
184−handleIncomingMessages node = go [] IntMap.empty
248+handleIncomingMessages node = go [] Map.empty
185249 where
186250 go halfOpenConns openConns = do
187251 event <- NT.receive endpoint
@@ -189,27 +253,27 @@ handleIncomingMessages node = go [] IntMap.empty
189253 NT.ConnectionOpened cid _rel _theirAddr ->
190254 go (cid : halfOpenConns) openConns
191255 NT.Received cid payload ->
192− case IntMap.lookup cid openConns of
256+ case Map.lookup cid openConns of
193257 Just proc -> do
194258 let msg = payloadToMessage payload
195259 enqueue (processQueue proc) msg
196260 go halfOpenConns openConns
197261 Nothing -> if cid `elem` halfOpenConns
198262 then do
199− let pid = NTI.decodeInt32 . BSS.concat $ payload
200− mProc <- withMVar state $ return . (^. localProcessWithId pid)
263+ let lpid = payloadToLpid payload
264+ mProc <- withMVar state $ return . (^. localProcessWithId lpid)
201265 case mProc of
202266 Just proc ->
203− go (List.delete pid halfOpenConns)
204− (IntMap.insert cid proc openConns)
267+ go (List.delete cid halfOpenConns)
268+ (Map.insert cid proc openConns)
205269 Nothing ->
206− fail "TODO"
270+ fail "handleIncomingMessages: TODO 1"
207271 else
208− fail "TODO"
272+ fail "handleIncomingMessages: TODO 2"
209273 NT.ConnectionClosed cid ->
210− go (List.delete cid halfOpenConns) (IntMap.delete cid openConns)
274+ go (List.delete cid halfOpenConns) (Map.delete cid openConns)
211275 NT.ErrorEvent _ ->
212− fail "TODO"
276+ fail "handleIncomingMessages: TODO 3"
213277 NT.EndPointClosed ->
214278 return ()
215279 NT.ReceivedMulticast _ _ ->
@@ -253,7 +317,7 @@ createConnectionTo node pid = do
253317 NT.close conn
254318 return conn'
255319 Nothing -> do
256− sendBinary node pid conn [NTI.encodeInt32 . processLocalId $ pid]
320+ sendBinary node pid conn $ lpidToPayload (processLocalId pid)
257321 return conn
258322 Left err ->
259323 throwIO err
@@ -279,13 +343,17 @@ payloadToMessage payload = Message fp msg
279343 $ BSL.fromChunks payload
280344 fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp
281345
282−--------------------------------------------------------------------------------
283−-- Constants --
284−--------------------------------------------------------------------------------
285−
286−firstNonReservedProcessId :: Int
287−firstNonReservedProcessId = 1024
288−
346+lpidToPayload :: LocalProcessId -> [BSS.ByteString]
347+lpidToPayload lpid = [ NTI.encodeInt32 (pidCounter lpid)
348+ , NTI.encodeInt32 (pidUnique lpid)
349+ ]
350+
351+payloadToLpid :: [BSS.ByteString] -> LocalProcessId
352+payloadToLpid bss = let (bs1, bs2) = BSS.splitAt 4 . BSS.concat $ bss
353+ in LocalProcessId { pidCounter = NTI.decodeInt32 bs1
354+ , pidUnique = NTI.decodeInt32 bs2
355+ }
356+
289357--------------------------------------------------------------------------------
290358-- Accessors --
291359--------------------------------------------------------------------------------
@@ -293,218 +361,17 @@ firstNonReservedProcessId = 1024
293361localConnections :: Accessor LocalNodeState (Map ProcessId NT.Connection)
294362localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })
295363
296−localProcesses :: Accessor LocalNodeState (IntMap ProcessState)
364+localProcesses :: Accessor LocalNodeState (Map LocalProcessId ProcessState)
297365localProcesses = accessor _localProcesses (\procs st -> st { _localProcesses = procs })
298366
299−localNextProcId :: Accessor LocalNodeState Int
300−localNextProcId = accessor _localNextProcId (\pid st -> st { _localNextProcId = pid })
367+localPidCounter :: Accessor LocalNodeState Int32
368+localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter = ctr })
369+
370+localPidUnique :: Accessor LocalNodeState Int32
371+localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })
301372
302373localConnectionTo :: ProcessId -> Accessor LocalNodeState (Maybe NT.Connection)
303374localConnectionTo pid = localConnections >>> DAC.mapMaybe pid
304375
305376localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe ProcessState)
306−localProcessWithId lpid = localProcesses >>> DAC.intMapMaybe lpid
307−
308−{-
309−{-# LANGUAGE CPP, ExplicitForAll, ScopedTypeVariables #-}
310−
311−module Control.Distributed.Process (
312− -- * Processes
313− Process,
314− liftIO,
315− ProcessId,
316−-- NodeId,
317−
318− -- * Basic messaging
319− send,
320− expect,
321−
322− -- * Process management
323− spawnLocal,
324− getSelfPid,
325−-- getSelfNode,
326−
327− -- * Initialisation
328− Transport,
329− LocalNode,
330− newLocalNode,
331− runProcess,
332−
333− ) where
334−
335−import qualified Network.Transport as Trans
336−import Network.Transport (Transport)
337−
338−import qualified Data.IntMap as IntMap
339−import Data.IntMap (IntMap)
340−import Control.Applicative
341−import Control.Monad
342−import Control.Monad.Trans
343−import Control.Concurrent.MVar
344−import Control.Concurrent.Chan
345−import Control.Concurrent
346−import Data.Typeable
347−
348−import Debug.Trace
349−
350−#ifndef LAZY
351−import qualified Data.ByteString.Char8 as BS
352−import Data.ByteString.Char8 (ByteString)
353−#else
354−import qualified Data.ByteString.Lazy.Char8 as BS
355−import Data.ByteString.Lazy.Char8 (ByteString)
356−#endif
357−
358−------------------------
359−-- Cloud Haskell layer
360−--
361−
362−data NodeId = NodeId -- !Trans.SourceEnd !Trans.SourceEnd
363−
364−data ProcessId = ProcessId !Trans.SourceEnd !NodeId !LocalProcessId
365−type LocalProcessId = Int
366−
367−newtype SourcePort a = SourcePort Trans.SourceEnd
368−
369−newtype Process a = Process { unProcess :: ProcessState -> IO a }
370−
371−instance Functor Process where
372− fmap f m = Process (\ps -> unProcess m ps >>= \x -> return (f x))
373−
374−instance Applicative Process where
375− pure = return
376− (<*>) = ap
377−
378−instance Monad Process where
379− m >>= k = Process (\ps -> unProcess m ps >>= \x -> unProcess (k x) ps)
380− return x = Process (\_ -> return x)
381−
382−instance MonadIO Process where
383− liftIO io = Process (\_ -> io)
384−
385−getProcessState :: Process ProcessState
386−getProcessState = Process (\ps -> return ps)
387−
388−getLocalNode :: Process LocalNode
389−getLocalNode = prNode <$> getProcessState
390−
391−getSelfPid :: Process ProcessId
392−getSelfPid = prPid <$> getProcessState
393−
394−-- State a process carries around and has local access to
395−--
396−data ProcessState = ProcessState {
397− prPid :: !ProcessId,
398− prChan :: !Trans.TargetEnd,
399− prQueue :: !(CQueue Message),
400− prNode :: !LocalNode
401− }
402−
403−data Message = Message String --string rep of TypeRep, sigh.
404− String
405−
406−-- Context for a local node, all processes running on a node have direct access to this
407−--
408−data LocalNode = LocalNode {
409− ndProcessTable :: !(MVar ProcessTable),
410− ndTransport :: !Transport
411− }
412−
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
418− }
419−
420−newLocalNode :: Transport -> IO LocalNode
421−newLocalNode trans = do
422− processTableVar <- newMVar (ProcessTable 0 IntMap.empty)
423− return LocalNode {
424− ndProcessTable = processTableVar,
425− ndTransport = trans
426− }
427−
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
435− takeMVar waitVar
436−
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
445− _ <- forkIO $ do
446− tid <- myThreadId
447− putMVar (ndProcessTable node) (insertProcess tid processTable)
448− queue <- newCQueue
449− _ <- forkIO $ receiverPump chan queue
450− action $ ProcessState pid chan queue node
451− return pid
452−
453− where
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)
458−
459− receiverPump :: Trans.TargetEnd -> CQueue Message -> IO ()
460− receiverPump chan queue = forever $ do
461− msgBlobs <- Trans.receive chan
462− readBlobs (concatMap BS.unpack msgBlobs)
463− where
464− readBlobs :: String -> IO ()
465− readBlobs [] = return ()
466− readBlobs bs = do
467− -- TODO: replace reads with something safer and faster.
468− let [((typerep, body), bs')] = reads bs
469− enqueue queue (Message typerep body)
470− readBlobs bs'
471−
472−send :: (Typeable a, Show a) => ProcessId -> a -> Process ()
473−send (ProcessId chan _ _) msg =
474− liftIO (Trans.send chan msgBlobs)
475− where
476− msgBlobs :: [ByteString]
477− msgBlobs = [BS.pack (show (show (typeOf msg), show msg))]
478−
479−expect :: forall a. (Typeable a, Read a) => Process a
480−expect = do
481− ProcessState { prQueue = queue } <- getProcessState
482− let typerepstr = show (typeOf (undefined :: a))
483− Message _ body <- liftIO $
484− dequeueMatching queue (\(Message typerepstr' _) -> typerepstr' == typerepstr)
485− return (read body)
486−
487−spawnLocal :: Process () -> Process ProcessId
488−spawnLocal proc = do
489− node <- getLocalNode
490− liftIO $ forkProcess node proc
491−
492−{-
493−send :: Serializable a -> ProcessId -> a -> Process ()
494−expect :: Serializable a -> Process a
495−
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
499−
500−spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
501−terminate :: ProcessM a
502−getSelfPid :: ProcessM ProcessId
503−getSelfNode :: ProcessM NodeId
504−
505−monitorProcess :: ProcessId -> ProcessId -> MonitorAction -> Process ()
506−
507−-}
508−
509−--
510−-}
377+localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid