3This is an implementation of Cloud Haskell, as described in
4/Towards Haskell in the Cloud/ by Jeff Epstein, Andrew Black, and Simon
5Peyton Jones (<http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/>),
6although some of the details are different. The precise message passing
7semantics are based on /A unified semantics for future Erlang/ by Hans
8Svensson, Lars-Åke Fredlund and Clara Benac Earle.
10For a detailed description of the package and other reference materials,
11please see the distributed-process wiki page on github:
12<https://github.com/haskell-distributed/distributed-process/wiki>.
15module Control.Distributed.Process
23 , liftIO -- Reexported for convenience
38 -- * Advanced messaging
58 -- * Process management
67 , ProcessTerminationException(..)
68 , ProcessRegistrationException(..)
74 -- * Monitoring and linking
86 , MonitorRef -- opaque
87 , ProcessLinkException(..)
88 , NodeLinkException(..)
89 , PortLinkException(..)
90 , ProcessMonitorNotification(..)
91 , NodeMonitorNotification(..)
92 , PortMonitorNotification(..)
109 , registerRemoteAsync
110 , reregisterRemoteAsync
111 , unregisterRemoteAsync
116 -- * Exception handling
133 -- * Local versions of 'spawn'
141#if ! MIN_VERSION_base(4,6,0)
142import Prelude hiding (catch)
145import Data.Typeable (Typeable)
146import Control.Monad.IO.Class (liftIO)
147import Control.Applicative ((<$>))
148import Control.Monad.Reader (ask)
149import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
150import Control.Distributed.Static
158import Control.Distributed.Process.Internal.Types
163 , ProcessMonitorNotification(..)
164 , NodeMonitorNotification(..)
165 , PortMonitorNotification(..)
166 , ProcessLinkException(..)
167 , NodeLinkException(..)
168 , PortLinkException(..)
169 , ProcessRegistrationException(..)
178 , LocalProcess(processNode)
182import Control.Distributed.Process.Serializable (Serializable, SerializableDict)
183import Control.Distributed.Process.Internal.Closure.BuiltIn
195import Control.Distributed.Process.Internal.Primitives
205 -- Advanced messaging
224 -- Process management
226 , ProcessTerminationException(..)
236 -- Monitoring and linking
256 , registerRemoteAsync
257 , reregisterRemoteAsync
258 , unregisterRemoteAsync
264 -- Exception handling
282import Control.Distributed.Process.Node (forkProcess)
286-- 1. 'send' never fails. If you want to know that the remote process received
287-- your message, you will need to send an explicit acknowledgement. If you
288-- want to know when the remote process failed, you will need to monitor
289-- that remote process.
291-- 2. 'send' may block (when the system TCP buffers are full, while we are
292-- trying to establish a connection to the remote endpoint, etc.) but its
293-- return does not imply that the remote process received the message (much
294-- less processed it). When 'send' targets a local process, it is dispatched
295-- via the node controller however, in which cases it will /not/ block. This
296-- isn't part of the semantics, so it should be ok. The ordering is maintained
297-- because the ctrlChannel still has FIFO semantics with regards interactions
298-- between two disparate forkIO threads.
300-- 3. Message delivery is reliable and ordered. That means that if process A
301-- sends messages m1, m2, m3 to process B, B will either arrive all three
302-- messages in order (m1, m2, m3) or a prefix thereof; messages will not be
303-- 'missing' (m1, m3) or reordered (m1, m3, m2)
305-- In order to guarantee (3), we stipulate that
307-- 3a. We do not garbage collect connections because Network.Transport provides
308-- ordering guarantees only *per connection*.
310-- 3b. Once a connection breaks, we have no way of knowing which messages
311-- arrived and which did not; hence, once a connection fails, we assume the
312-- remote process to be forever unreachable. Otherwise we might sent m1 and
313-- m2, get notified of the broken connection, reconnect, send m3, but only
316-- 3c. As a consequence of (3b) we should not reuse PIDs. If a process dies,
317-- we consider it forever unreachable. Hence, new processes should get new
318-- IDs or they too would be considered unreachable.
320-- Main reference for Cloud Haskell is
322-- [1] "Towards Haskell in the Cloud", Jeff Epstein, Andrew Black and Simon
324-- http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/remote.pdf
326-- The precise semantics for message passing is based on
328-- [2] "A Unified Semantics for Future Erlang", Hans Svensson, Lars-Ake Fredlund
329-- and Clara Benac Earle (not freely available online, unfortunately)
331-- Some pointers to related documentation about Erlang, for comparison and
334-- [3] "Programming Distributed Erlang Applications: Pitfalls and Recipes",
335-- Hans Svensson and Lars-Ake Fredlund
336-- http://man.lupaworld.com/content/develop/p37-svensson.pdf
337-- [4] The Erlang manual, sections "Message Sending" and "Send"
338-- http://www.erlang.org/doc/reference_manual/processes.html#id82409
339-- http://www.erlang.org/doc/reference_manual/expressions.html#send
340-- [5] Questions "Is the order of message reception guaranteed?" and
341-- "If I send a message, is it guaranteed to reach the receiver?" of
343-- http://www.erlang.org/faq/academic.html
344-- [6] "Delivery of Messages", post on erlang-questions
345-- http://erlang.org/pipermail/erlang-questions/2012-February/064767.html
347--------------------------------------------------------------------------------
348-- Primitives are defined in a separate module; here we only define derived --
350--------------------------------------------------------------------------------
354-- For more information about 'Closure', see
355-- "Control.Distributed.Process.Closure".
358spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
361 mRef <- monitorNode nid
362 sRef <- spawnAsync nid (cpDelay us proc)
364 matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do
368 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ ->
369 return (nullProcessId nid)
372-- | Spawn a process and link to it
374-- Note that this is just the sequential composition of 'spawn' and 'link'.
375-- (The "Unified" semantics that underlies Cloud Haskell does not even support
376-- a synchronous link operation)
377spawnLink :: NodeId -> Closure (Process ()) -> Process ProcessId
378spawnLink nid proc = do
379 pid <- spawn nid proc
383-- | Like 'spawnLink', but monitor the spawned process
384spawnMonitor :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef)
385spawnMonitor nid proc = do
386 pid <- spawn nid proc
390-- | Run a process remotely and wait for it to reply
392-- We monitor the remote process: if it dies before it can send a reply, we die
395-- For more information about 'Static', 'SerializableDict', and 'Closure', see
396-- "Control.Distributed.Process.Closure".
399call :: Serializable a
400 => Static (SerializableDict a)
402 -> Closure (Process a)
404call dict nid proc = do
406 (pid, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)
407 -- We are guaranteed to receive the reply before the monitor notification
408 -- (if a reply is sent at all)
409 -- NOTE: This might not be true if we switch to unreliable delivery.
410 mResult <- receiveWait
411 [ match (return . Right)
412 , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)
413 (\(ProcessMonitorNotification _ _ reason) -> return (Left reason))
417 -- Wait for the monitor message so that we the mailbox doesn't grow
419 [ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)
420 (\(ProcessMonitorNotification {}) -> return ())
422 -- Clean up connection to pid
426 fail $ "call: remote process died: " ++ show err
428-- | Spawn a child process, have the child link to the parent and the parent
430spawnSupervised :: NodeId
431 -> Closure (Process ())
432 -> Process (ProcessId, MonitorRef)
433spawnSupervised nid proc = do
435 them <- spawn nid (cpLink us `seqCP` proc)
439-- | Spawn a new process, supplying it with a new 'ReceivePort' and return
440-- the corresponding 'SendPort'.
441spawnChannel :: forall a. Typeable a => Static (SerializableDict a)
443 -> Closure (ReceivePort a -> Process ())
444 -> Process (SendPort a)
445spawnChannel dict nid proc = do
447 _ <- spawn nid (go us)
450 go :: ProcessId -> Closure (Process ())
451 go pid = cpNewChan dict
453 (cpSend (sdictSendPort dict) pid `splitCP` proc)
455 (idCP `closureCompose` staticClosure sndStatic)
457--------------------------------------------------------------------------------
458-- Local versions of spawn --
459--------------------------------------------------------------------------------
461-- | Spawn a process on the local node
462spawnLocal :: Process () -> Process ProcessId
464 node <- processNode <$> ask
465 liftIO $ forkProcess node proc
467-- | Create a new typed channel, spawn a process on the local node, passing it
468-- the receive port, and return the send port
469spawnChannelLocal :: Serializable a
470 => (ReceivePort a -> Process ())
471 -> Process (SendPort a)
472spawnChannelLocal proc = do
473 node <- processNode <$> ask
476 _ <- forkProcess node $ do
477 -- It is important that we allocate the new channel in the new process,
478 -- because otherwise it will be associated with the wrong process ID
479 (sport, rport) <- newChan
480 liftIO $ putMVar mvar sport
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