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
105 , MonitorRef -- opaque
106 , ProcessLinkException(..)
107 , NodeLinkException(..)
108 , PortLinkException(..)
109 , ProcessMonitorNotification(..)
110 , NodeMonitorNotification(..)
111 , PortMonitorNotification(..)
128 , registerRemoteAsync
129 , reregisterRemoteAsync
130 , unregisterRemoteAsync
135 -- * Exception handling
153 -- * Local versions of 'spawn'
162#if ! MIN_VERSION_base(4,6,0)
163import Prelude hiding (catch)
166import Control.Monad.IO.Class (liftIO)
167import Control.Applicative ((<$>))
168import Control.Monad.Reader (ask)
169import Control.Concurrent (forkIO)
170import Control.Concurrent.MVar
176import Control.Distributed.Static
182import Control.Distributed.Process.Internal.Types
187 , ProcessMonitorNotification(..)
188 , NodeMonitorNotification(..)
189 , PortMonitorNotification(..)
190 , ProcessLinkException(..)
191 , NodeLinkException(..)
192 , PortLinkException(..)
193 , ProcessRegistrationException(..)
202 , LocalProcess(processNode)
206import Control.Distributed.Process.Serializable (Serializable)
207import Control.Distributed.Process.Internal.Primitives
223 -- Advanced messaging
249 -- Process management
251 , ProcessTerminationException(..)
264 -- Monitoring and linking
284 , registerRemoteAsync
285 , reregisterRemoteAsync
286 , unregisterRemoteAsync
292 -- Exception handling
311import Control.Distributed.Process.Node (forkProcess)
312import Control.Distributed.Process.Internal.Types
316 , withValidLocalState
318import Control.Distributed.Process.Internal.Spawn
319 ( -- Spawning Processes/Channels
327import Control.Distributed.Process.Internal.StrictMVar (readMVar)
328import Control.Exception (SomeException, throw, throwTo, throwIO)
329import Control.Monad (forM_)
334-- 1. 'send' never fails. If you want to know that the remote process received
335-- your message, you will need to send an explicit acknowledgement. If you
336-- want to know when the remote process failed, you will need to monitor
337-- that remote process.
339-- 2. 'send' may block (when the system TCP buffers are full, while we are
340-- trying to establish a connection to the remote endpoint, etc.) but its
341-- return does not imply that the remote process received the message (much
342-- less processed it). When 'send' targets a local process, it is dispatched
343-- via the node controller however, in which cases it will /not/ block. This
344-- isn't part of the semantics, so it should be ok. The ordering is maintained
345-- because the ctrlChannel still has FIFO semantics with regards interactions
346-- between two disparate forkIO threads.
348-- 3. Message delivery is reliable and ordered. That means that if process A
349-- sends messages m1, m2, m3 to process B, B will either arrive all three
350-- messages in order (m1, m2, m3) or a prefix thereof; messages will not be
351-- 'missing' (m1, m3) or reordered (m1, m3, m2)
353-- In order to guarantee (3), we stipulate that
355-- 3a. We do not garbage collect connections because Network.Transport provides
356-- ordering guarantees only *per connection*.
358-- 3b. Once a connection breaks, we have no way of knowing which messages
359-- arrived and which did not; hence, once a connection fails, we assume the
360-- remote process to be forever unreachable. Otherwise we might sent m1 and
361-- m2, get notified of the broken connection, reconnect, send m3, but only
364-- 3c. As a consequence of (3b) we should not reuse PIDs. If a process dies,
365-- we consider it forever unreachable. Hence, new processes should get new
366-- IDs or they too would be considered unreachable.
368-- Main reference for Cloud Haskell is
370-- [1] "Towards Haskell in the Cloud", Jeff Epstein, Andrew Black and Simon
372-- http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/remote.pdf
374-- The precise semantics for message passing is based on
376-- [2] "A Unified Semantics for Future Erlang", Hans Svensson, Lars-Ake Fredlund
377-- and Clara Benac Earle (not freely available online, unfortunately)
379-- Some pointers to related documentation about Erlang, for comparison and
382-- [3] "Programming Distributed Erlang Applications: Pitfalls and Recipes",
383-- Hans Svensson and Lars-Ake Fredlund
384-- http://man.lupaworld.com/content/develop/p37-svensson.pdf
385-- [4] The Erlang manual, sections "Message Sending" and "Send"
386-- http://www.erlang.org/doc/reference_manual/processes.html#id82409
387-- http://www.erlang.org/doc/reference_manual/expressions.html#send
388-- [5] Questions "Is the order of message reception guaranteed?" and
389-- "If I send a message, is it guaranteed to reach the receiver?" of
391-- http://www.erlang.org/faq/academic.html
392-- [6] "Delivery of Messages", post on erlang-questions
393-- http://erlang.org/pipermail/erlang-questions/2012-February/064767.html
395--------------------------------------------------------------------------------
396-- Local versions of spawn --
397--------------------------------------------------------------------------------
399-- | Spawn a process on the local node
400spawnLocal :: Process () -> Process ProcessId
402 node <- processNode <$> ask
403 liftIO $ forkProcess node proc
405-- | Create a new typed channel, spawn a process on the local node, passing it
406-- the receive port, and return the send port
407spawnChannelLocal :: Serializable a
408 => (ReceivePort a -> Process ())
409 -> Process (SendPort a)
410spawnChannelLocal proc = do
411 node <- processNode <$> ask
414 _ <- forkProcess node $ do
415 -- It is important that we allocate the new channel in the new process,
416 -- because otherwise it will be associated with the wrong process ID
417 (sport, rport) <- newChan
418 liftIO $ putMVar mvar sport
422-- | Local version of 'call'. Running a process in this way isolates it from
423-- messages sent to the caller process, and also allows silently dropping late
424-- or duplicate messages sent to the isolated process after it exits.
425-- Silently dropping messages may not always be the best approach.
426callLocal :: Process a -> Process a
427callLocal proc = mask $ \release -> do
428 mv <- liftIO newEmptyMVar :: Process (MVar (Either SomeException a))
429 (spInit, rpInit) <- newChan -- TODO: Remove when spawnLocal inherits the
431 child <- spawnLocal $ mask_ $
432 try (sendChan spInit () >> release proc) >>= liftIO . putMVar mv
433 rs <- liftIO (takeMVar mv)
435 -- Exceptions need to be prevented from interrupting the clean up or
436 -- the original exception which caused entering the handler could be
437 -- forgotten. For instance, this could have a problematic effect
438 -- when the original exception was meant to kill the thread and the
439 -- second exception doesn't (like the exception thrown by
440 -- 'System.Timeout.timeout').
442 -- Using 'uninterruptibleMask_' would be problematic as there is no
443 -- way to be sure the node isn't closing when the cleanup executes.
444 -- Successfully running the cleanup is necessary to ensure
445 -- termination. If exceptions are masked uninterruptibly,
446 -- 'receiveChan' might block indefinitely and 'kill' might have no
449 do (exs0, _) <- collectExceptions [] $
450 whenNotClosed $ receiveChan rpInit
451 (exs1, _) <- collectExceptions exs0 $
452 whenNotClosed $ kill child "callLocal was interrupted"
453 (exs, mr) <- collectExceptions exs1 $
454 whenNotClosed $ liftIO (takeMVar mv)
455 -- Forward exceptions asynchronously.
457 liftIO $ forkIO $ forM_ (reverse exs) $
458 throwTo (processThread lproc)
461 maybe (throwIO $ userError $ "Node closed " ++ show here)
463 either throw return rs
465 -- Retries a given action until it succeeds.
466 collectExceptions :: [SomeException]
468 -> Process ([SomeException], a)
469 collectExceptions exs p =
470 try p >>= either (\e -> collectExceptions (e : exs) p) (return . (,) exs)
472 -- Runs the given closure if the node is not closed.
473 whenNotClosed :: Process a -> Process (Maybe a)
476 st <- liftIO $ readMVar (localState $ processNode lproc)
478 LocalNodeValid _ -> fmap Just p
479 LocalNodeClosed -> return Nothing
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