@@ -1,64 +1,16 @@
1−-- | Cloud Haskell
1+-- | [Cloud Haskell]
3−-- 1. 'send' never fails. If you want to know that the remote process received
3+-- This is an implementation of Cloud Haskell, as described in
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. We do not garbage collect connections because Network.Transport provides
21−-- ordering guarantees only *per connection*.
23−-- 3b. Once a connection breaks, we have no way of knowing which messages
24−-- arrived and which did not; hence, once a connection fails, we assume the
25−-- remote process to be forever unreachable. Otherwise we might sent m1 and
26−-- m2, get notified of the broken connection, reconnect, send m3, but only
29−-- 3c. As a consequence of (3b) we should not reuse PIDs. If a process dies,
30−-- we consider it forever unreachable. Hence, new processes should get new
31−-- IDs or they too would be considered unreachable.
33−-- Main reference for Cloud Haskell is
35−-- [1] "Towards Haskell in the Cloud", Jeff Epstein, Andrew Black and Simon
4+-- /Towards Haskell in the Cloud/ by Jeff Epstein, Andrew Black, and Simon
36−-- Peyton-Jones.
5+-- Peyton Jones
37−-- http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/remote.pdf
6+-- (<http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/>),
39−-- The precise semantics for message passing is based on
7+-- although some of the details are different. The precise message passing
40−--
8+-- semantics are based on /A unified semantics for future Erlang/ by Hans
41−-- [2] "A Unified Semantics for Future Erlang", Hans Svensson, Lars-Ake Fredlund
9+-- Svensson, Lars-Åke Fredlund and Clara Benac Earle.
42−-- and Clara Benac Earle (not freely available online, unfortunately)
44−-- Some pointers to related documentation about Erlang, for comparison and
47−-- [3] "Programming Distributed Erlang Applications: Pitfalls and Recipes",
48−-- Hans Svensson and Lars-Ake Fredlund
49−-- http://man.lupaworld.com/content/develop/p37-svensson.pdf
50−-- [4] The Erlang manual, sections "Message Sending" and "Send"
51−-- http://www.erlang.org/doc/reference_manual/processes.html#id82409
52−-- http://www.erlang.org/doc/reference_manual/expressions.html#send
53−-- [5] Questions "Is the order of message reception guaranteed?" and
54−-- "If I send a message, is it guaranteed to reach the receiver?" of
56−-- http://www.erlang.org/faq/academic.html
57−-- [6] "Delivery of Messages", post on erlang-questions
58−-- http://erlang.org/pipermail/erlang-questions/2012-February/064767.html
59module Control.Distributed.Process
10module Control.Distributed.Process
60 ( -- * Basic cloud Haskell API
11 ( -- * Basic cloud Haskell API
@@ -104,7 +56,6 @@ import qualified Data.List as List (delete)
104import Data.Set (Set)
56import Data.Set (Set)
105import qualified Data.Set as Set (empty, insert, delete, member)
57import qualified Data.Set as Set (empty, insert, delete, member)
106import Data.Foldable (forM_)
58import Data.Foldable (forM_)
107−import Data.Dynamic (Dynamic)
108import Data.Typeable (Typeable)
59import Data.Typeable (Typeable)
109import Control.Monad (void)
60import Control.Monad (void)
110import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
61import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
@@ -174,14 +125,73 @@ import Control.Distributed.Process.Internal ( RemoteCallMetaData
174 , connectionTo
125 , connectionTo
177−-- The Cloud Haskell 'Process' type
128+-- INTERNAL NOTES
178−newtype Process a = Process { unProcess :: ReaderT LocalProcess IO a }
129+--
179− deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable)
130+-- 1. 'send' never fails. If you want to know that the remote process received
131+-- your message, you will need to send an explicit acknowledgement. If you
132+-- want to know when the remote process failed, you will need to monitor
133+-- that remote process.
135+-- 2. 'send' may block (when the system TCP buffers are full, while we are
136+-- trying to establish a connection to the remote endpoint, etc.) but its
137+-- return does not imply that the remote process received the message (much
138+-- less processed it)
140+-- 3. Message delivery is reliable and ordered. That means that if process A
141+-- sends messages m1, m2, m3 to process B, B will either arrive all three
142+-- messages in order (m1, m2, m3) or a prefix thereof; messages will not be
143+-- 'missing' (m1, m3) or reordered (m1, m3, m2)
145+-- In order to guarantee (3), we stipulate that
147+-- 3a. We do not garbage collect connections because Network.Transport provides
148+-- ordering guarantees only *per connection*.
150+-- 3b. Once a connection breaks, we have no way of knowing which messages
151+-- arrived and which did not; hence, once a connection fails, we assume the
152+-- remote process to be forever unreachable. Otherwise we might sent m1 and
153+-- m2, get notified of the broken connection, reconnect, send m3, but only
156+-- 3c. As a consequence of (3b) we should not reuse PIDs. If a process dies,
157+-- we consider it forever unreachable. Hence, new processes should get new
158+-- IDs or they too would be considered unreachable.
160+-- Main reference for Cloud Haskell is
162+-- [1] "Towards Haskell in the Cloud", Jeff Epstein, Andrew Black and Simon
164+-- http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/remote.pdf
166+-- The precise semantics for message passing is based on
168+-- [2] "A Unified Semantics for Future Erlang", Hans Svensson, Lars-Ake Fredlund
169+-- and Clara Benac Earle (not freely available online, unfortunately)
171+-- Some pointers to related documentation about Erlang, for comparison and
174+-- [3] "Programming Distributed Erlang Applications: Pitfalls and Recipes",
175+-- Hans Svensson and Lars-Ake Fredlund
176+-- http://man.lupaworld.com/content/develop/p37-svensson.pdf
177+-- [4] The Erlang manual, sections "Message Sending" and "Send"
178+-- http://www.erlang.org/doc/reference_manual/processes.html#id82409
179+-- http://www.erlang.org/doc/reference_manual/expressions.html#send
180+-- [5] Questions "Is the order of message reception guaranteed?" and
181+-- "If I send a message, is it guaranteed to reach the receiver?" of
183+-- http://www.erlang.org/faq/academic.html
184+-- [6] "Delivery of Messages", post on erlang-questions
185+-- http://erlang.org/pipermail/erlang-questions/2012-February/064767.html
181--------------------------------------------------------------------------------
187--------------------------------------------------------------------------------
182-- Basic Cloud Haskell API --
188-- Basic Cloud Haskell API --
183--------------------------------------------------------------------------------
189--------------------------------------------------------------------------------
191+-- | The Cloud Haskell 'Process' type
192+newtype Process a = Process { unProcess :: ReaderT LocalProcess IO a }
193+ deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable)
185-- | Wait for a message of a specific type
195-- | Wait for a message of a specific type
186expect :: forall a. Serializable a => Process a
196expect :: forall a. Serializable a => Process a
187expect = receiveWait [match return]
197expect = receiveWait [match return]
@@ -261,19 +271,20 @@ receiveTimeout t ms = do
261-- Auxiliary API --
271-- Auxiliary API --
262--------------------------------------------------------------------------------
272--------------------------------------------------------------------------------
264−-- Force-close a local node
274+-- | Force-close a local node
266-- TODO: for now we just close the associated endpoint
276-- TODO: for now we just close the associated endpoint
267closeLocalNode :: LocalNode -> IO ()
277closeLocalNode :: LocalNode -> IO ()
268closeLocalNode node =
278closeLocalNode node =
269 NT.closeEndPoint (localEndPoint node)
279 NT.closeEndPoint (localEndPoint node)
271−-- Catch exceptions within a process
281+-- | Catch exceptions within a process
272catch :: Exception e => Process a -> (e -> Process a) -> Process a
282catch :: Exception e => Process a -> (e -> Process a) -> Process a
273catch p h = do
283catch p h = do
274 run <- runLocalProcess <$> ask
284 run <- runLocalProcess <$> ask
275 liftIO $ Exception.catch (run p) (run . h)
285 liftIO $ Exception.catch (run p) (run . h)
287+-- | Like 'expect' but with a timeout
277expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
288expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
278expectTimeout timeout = receiveTimeout timeout [match return]
289expectTimeout timeout = receiveTimeout timeout [match return]
@@ -281,7 +292,11 @@ expectTimeout timeout = receiveTimeout timeout [match return]
281-- Initialization --
292-- Initialization --
282--------------------------------------------------------------------------------
293--------------------------------------------------------------------------------
295+-- | Initialize a new local node.
297+-- Note that proper Cloud Haskell initialization and configuration is still
284−newLocalNode :: NT.Transport -> Map String Dynamic -> IO LocalNode
299+newLocalNode :: NT.Transport -> RemoteCallMetaData -> IO LocalNode
285newLocalNode transport metaData = do
300newLocalNode transport metaData = do
286 mEndPoint <- NT.newEndPoint transport
301 mEndPoint <- NT.newEndPoint transport
287 case mEndPoint of
302 case mEndPoint of
@@ -305,12 +320,14 @@ newLocalNode transport metaData = do
305 void . forkIO $ handleIncomingMessages node
320 void . forkIO $ handleIncomingMessages node
306 return node
321 return node
323+-- | Run a process on a local node and wait for it to finish
308runProcess :: LocalNode -> Process () -> IO ()
324runProcess :: LocalNode -> Process () -> IO ()
309runProcess node proc = do
325runProcess node proc = do
310 done <- newEmptyMVar
326 done <- newEmptyMVar
311 void $ forkProcess node (proc >> liftIO (putMVar done ()))
327 void $ forkProcess node (proc >> liftIO (putMVar done ()))
312 takeMVar done
328 takeMVar done
330+-- | Spawn a new process on a local node
314forkProcess :: LocalNode -> Process () -> IO ProcessId
331forkProcess :: LocalNode -> Process () -> IO ProcessId
315forkProcess node proc = modifyMVar (localState node) $ \st -> do
332forkProcess node proc = modifyMVar (localState node) $ \st -> do
316 let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter
333 let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter