4This is an implementation of Cloud Haskell, as described in
5/Towards Haskell in the Cloud/ by Jeff Epstein, Andrew Black, and Simon
7<http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/>),
8although some of the details are different. The precise message passing
9semantics are based on /A unified semantics for future Erlang/ by Hans
10Svensson, Lars-Åke Fredlund and Clara Benac Earle.
12For a detailed description of the package and other reference materials,
13please see the distributed-process wiki page on github:
14<https://github.com/haskell-distributed/distributed-process/wiki>.
17module Control.Distributed.Process
25 , liftIO -- Reexported for convenience
41 -- * Unsafe messaging variants
48 -- * Advanced messaging
74 -- * Process management
83 , ProcessTerminationException(..)
84 , ProcessRegistrationException(..)
93 -- * Monitoring and linking
106 , MonitorRef -- opaque
107 , ProcessLinkException(..)
108 , NodeLinkException(..)
109 , PortLinkException(..)
110 , ProcessMonitorNotification(..)
111 , NodeMonitorNotification(..)
112 , PortMonitorNotification(..)
129 , registerRemoteAsync
130 , reregisterRemoteAsync
131 , unregisterRemoteAsync
136 -- * Exception handling
154 -- * Local versions of 'spawn'
163import Control.Monad.IO.Class (liftIO)
164import Control.Applicative
165import Control.Monad.Reader (ask)
166import Control.Concurrent (killThread)
167import Control.Concurrent.MVar
173import Control.Distributed.Static
179import Control.Distributed.Process.Internal.Types
184 , ProcessMonitorNotification(..)
185 , NodeMonitorNotification(..)
186 , PortMonitorNotification(..)
187 , ProcessLinkException(..)
188 , NodeLinkException(..)
189 , PortLinkException(..)
190 , ProcessRegistrationException(..)
199 , LocalProcess(processNode)
203import Control.Distributed.Process.Serializable (Serializable)
204import Control.Distributed.Process.Internal.Primitives
220 -- Advanced messaging
246 -- Process management
248 , ProcessTerminationException(..)
261 -- Monitoring and linking
282 , registerRemoteAsync
283 , reregisterRemoteAsync
284 , unregisterRemoteAsync
290 -- Exception handling
309import Control.Distributed.Process.Node (forkProcess)
310import Control.Distributed.Process.Internal.Types
312 , withValidLocalState
314import Control.Distributed.Process.Internal.Spawn
315 ( -- Spawning Processes/Channels
323import qualified Control.Monad.Catch as Catch
325#if MIN_VERSION_base(4,6,0)
328import Prelude hiding (catch)
330import Control.Distributed.Process.Internal.StrictMVar (readMVar)
331import qualified Control.Exception as Exception (onException)
332import Data.Accessor ((^.))
333import Data.Foldable (forM_)
338-- 1. 'send' never fails. If you want to know that the remote process received
339-- your message, you will need to send an explicit acknowledgement. If you
340-- want to know when the remote process failed, you will need to monitor
341-- that remote process.
343-- 2. 'send' may block (when the system TCP buffers are full, while we are
344-- trying to establish a connection to the remote endpoint, etc.) but its
345-- return does not imply that the remote process received the message (much
346-- less processed it). When 'send' targets a local process, it is dispatched
347-- via the node controller however, in which cases it will /not/ block. This
348-- isn't part of the semantics, so it should be ok. The ordering is maintained
349-- because the ctrlChannel still has FIFO semantics with regards interactions
350-- between two disparate forkIO threads.
352-- 3. Message delivery is reliable and ordered. That means that if process A
353-- sends messages m1, m2, m3 to process B, B will either arrive all three
354-- messages in order (m1, m2, m3) or a prefix thereof; messages will not be
355-- 'missing' (m1, m3) or reordered (m1, m3, m2)
357-- In order to guarantee (3), we stipulate that
359-- 3a. We do not garbage collect connections because Network.Transport provides
360-- ordering guarantees only *per connection*.
362-- 3b. Once a connection breaks, we have no way of knowing which messages
363-- arrived and which did not; hence, once a connection fails, we assume the
364-- remote process to be forever unreachable. Otherwise we might sent m1 and
365-- m2, get notified of the broken connection, reconnect, send m3, but only
368-- 3c. As a consequence of (3b) we should not reuse PIDs. If a process dies,
369-- we consider it forever unreachable. Hence, new processes should get new
370-- IDs or they too would be considered unreachable.
372-- Main reference for Cloud Haskell is
374-- [1] "Towards Haskell in the Cloud", Jeff Epstein, Andrew Black and Simon
376-- http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/remote.pdf
378-- The precise semantics for message passing is based on
380-- [2] "A Unified Semantics for Future Erlang", Hans Svensson, Lars-Ake Fredlund
381-- and Clara Benac Earle (not freely available online, unfortunately)
383-- Some pointers to related documentation about Erlang, for comparison and
386-- [3] "Programming Distributed Erlang Applications: Pitfalls and Recipes",
387-- Hans Svensson and Lars-Ake Fredlund
388-- http://man.lupaworld.com/content/develop/p37-svensson.pdf
389-- [4] The Erlang manual, sections "Message Sending" and "Send"
390-- http://www.erlang.org/doc/reference_manual/processes.html#id82409
391-- http://www.erlang.org/doc/reference_manual/expressions.html#send
392-- [5] Questions "Is the order of message reception guaranteed?" and
393-- "If I send a message, is it guaranteed to reach the receiver?" of
395-- http://www.erlang.org/faq/academic.html
396-- [6] "Delivery of Messages", post on erlang-questions
397-- http://erlang.org/pipermail/erlang-questions/2012-February/064767.html
399--------------------------------------------------------------------------------
400-- Local versions of spawn --
401--------------------------------------------------------------------------------
403-- | Spawn a process on the local node
404spawnLocal :: Process () -> Process ProcessId
406 node <- processNode <$> ask
407 liftIO $ forkProcess node proc
409-- | Create a new typed channel, spawn a process on the local node, passing it
410-- the receive port, and return the send port
411spawnChannelLocal :: Serializable a
412 => (ReceivePort a -> Process ())
413 -> Process (SendPort a)
414spawnChannelLocal proc = do
415 node <- processNode <$> ask
418 _ <- forkProcess node $ do
419 -- It is important that we allocate the new channel in the new process,
420 -- because otherwise it will be associated with the wrong process ID
421 (sport, rport) <- newChan
422 liftIO $ putMVar mvar sport
426-- | Local version of 'call'. Running a process in this way isolates it from
427-- messages sent to the caller process, and also allows silently dropping late
428-- or duplicate messages sent to the isolated process after it exits.
429-- Silently dropping messages may not always be the best approach.
430callLocal :: Process a -> Process a
431callLocal proc = Catch.mask $ \release -> do
432 mv <- liftIO newEmptyMVar :: Process (MVar (Either Catch.SomeException a))
433 child <- spawnLocal $ Catch.try (release proc) >>= liftIO . putMVar mv
436 rs <- Exception.onException (takeMVar mv) $ Catch.uninterruptibleMask_ $
437 -- Exceptions need to be prevented from interrupting the clean up or
438 -- the original exception which caused entering the handler could be
439 -- forgotten. For instance, this could have a problematic effect
440 -- when the original exception was meant to kill the thread and the
441 -- second exception doesn't (like the exception thrown by
442 -- 'System.Timeout.timeout').
443 do mchildThreadId <- withValidLocalState (processNode lproc) $
444 \vst -> return $ fmap processThread $
445 vst ^. localProcessWithId (processLocalId child)
446 forM_ mchildThreadId killThread
448 either Catch.throwM return rs
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list