git.iko.soy/distributed-process
on 263cb13 help sign in
distributed-process/src/Control/Distributed/Process.hs
472 lines 13 kB history raw
1{- | [Cloud Haskell]
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>.
14-}
15module Control.Distributed.Process
16 ( -- * Basic types
17 ProcessId
18 , NodeId
19 , Process
20 , SendPortId
21 , processNodeId
22 , sendPortProcessId
23 , liftIO -- Reexported for convenience
24 -- * Basic messaging
25 , send
26 , expect
27 , expectTimeout
28 -- * Channels
29 , ReceivePort
30 , SendPort
31 , sendPortId
32 , newChan
33 , sendChan
34 , receiveChan
35 , receiveChanTimeout
36 , mergePortsBiased
37 , mergePortsRR
38 -- * Advanced messaging
39 , Match
40 , receiveWait
41 , receiveTimeout
42 , match
43 , matchIf
44 , matchUnknown
45 , matchAny
46 , matchAnyIf
47 , matchChan
48 , Message
49 , matchMessage
50 , matchMessageIf
51 , wrapMessage
52 , unwrapMessage
53 , handleMessage
54 , forward
55 -- * Process management
56 , spawn
57 , call
58 , terminate
59 , die
60 , kill
61 , exit
62 , catchExit
63 , catchesExit
64 , ProcessTerminationException(..)
65 , ProcessRegistrationException(..)
66 , SpawnRef
67 , getSelfPid
68 , getSelfNode
69 , ProcessInfo(..)
70 , getProcessInfo
71 -- * Monitoring and linking
72 , link
73 , linkNode
74 , linkPort
75 , unlink
76 , unlinkNode
77 , unlinkPort
78 , monitor
79 , monitorNode
80 , monitorPort
81 , unmonitor
82 , withMonitor
83 , MonitorRef -- opaque
84 , ProcessLinkException(..)
85 , NodeLinkException(..)
86 , PortLinkException(..)
87 , ProcessMonitorNotification(..)
88 , NodeMonitorNotification(..)
89 , PortMonitorNotification(..)
90 , DiedReason(..)
91 -- * Closures
92 , Closure
93 , closure
94 , Static
95 , unStatic
96 , unClosure
97 , RemoteTable
98 -- * Logging
99 , say
100 -- * Registry
101 , register
102 , reregister
103 , unregister
104 , whereis
105 , nsend
106 , registerRemoteAsync
107 , reregisterRemoteAsync
108 , unregisterRemoteAsync
109 , whereisRemoteAsync
110 , nsendRemote
111 , WhereIsReply(..)
112 , RegisterReply(..)
113 -- * Exception handling
114 , catch
115 , Handler(..)
116 , catches
117 , try
118 , mask
119 , onException
120 , bracket
121 , bracket_
122 , finally
123 -- * Auxiliary API
124 , spawnAsync
125 , spawnSupervised
126 , spawnLink
127 , spawnMonitor
128 , spawnChannel
129 , DidSpawn(..)
130 -- * Local versions of 'spawn'
131 , spawnLocal
132 , spawnChannelLocal
133 -- * Reconnecting
134 , reconnect
135 , reconnectPort
136 ) where
138#if ! MIN_VERSION_base(4,6,0)
139import Prelude hiding (catch)
140#endif
142import Data.Typeable (Typeable)
143import Control.Monad.IO.Class (liftIO)
144import Control.Applicative ((<$>))
145import Control.Monad.Reader (ask)
146import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
147import Control.Distributed.Static
148 ( Closure
149 , closure
150 , Static
151 , RemoteTable
152 , closureCompose
153 , staticClosure
154 )
155import Control.Distributed.Process.Internal.Types
156 ( NodeId(..)
157 , ProcessId(..)
158 , Process(..)
159 , MonitorRef(..)
160 , ProcessMonitorNotification(..)
161 , NodeMonitorNotification(..)
162 , PortMonitorNotification(..)
163 , ProcessLinkException(..)
164 , NodeLinkException(..)
165 , PortLinkException(..)
166 , ProcessRegistrationException(..)
167 , DiedReason(..)
168 , SpawnRef(..)
169 , DidSpawn(..)
170 , SendPort(..)
171 , ReceivePort(..)
172 , SendPortId(..)
173 , WhereIsReply(..)
174 , RegisterReply(..)
175 , LocalProcess(processNode)
176 , Message
177 , nullProcessId
178 )
179import Control.Distributed.Process.Serializable (Serializable, SerializableDict)
180import Control.Distributed.Process.Internal.Closure.BuiltIn
181 ( sdictSendPort
182 , sndStatic
183 , idCP
184 , seqCP
185 , bindCP
186 , splitCP
187 , cpLink
188 , cpSend
189 , cpNewChan
190 , cpDelay
191 )
192import Control.Distributed.Process.Internal.Primitives
193 ( -- Basic messaging
194 send
195 , expect
196 -- Channels
197 , newChan
198 , sendChan
199 , receiveChan
200 , mergePortsBiased
201 , mergePortsRR
202 -- Advanced messaging
203 , Match
204 , receiveWait
205 , receiveTimeout
206 , match
207 , matchIf
208 , matchUnknown
209 , matchAny
210 , matchAnyIf
211 , matchChan
212 , matchMessage
213 , matchMessageIf
214 , wrapMessage
215 , unwrapMessage
216 , handleMessage
217 , forward
218 -- Process management
219 , terminate
220 , ProcessTerminationException(..)
221 , die
222 , exit
223 , catchExit
224 , catchesExit
225 , kill
226 , getSelfPid
227 , getSelfNode
228 , ProcessInfo(..)
229 , getProcessInfo
230 -- Monitoring and linking
231 , link
232 , linkNode
233 , linkPort
234 , unlink
235 , unlinkNode
236 , unlinkPort
237 , monitor
238 , monitorNode
239 , monitorPort
240 , unmonitor
241 , withMonitor
242 -- Logging
243 , say
244 -- Registry
245 , register
246 , reregister
247 , unregister
248 , whereis
249 , nsend
250 , registerRemoteAsync
251 , reregisterRemoteAsync
252 , unregisterRemoteAsync
253 , whereisRemoteAsync
254 , nsendRemote
255 -- Closures
256 , unStatic
257 , unClosure
258 -- Exception handling
259 , catch
260 , Handler(..)
261 , catches
262 , try
263 , mask
264 , onException
265 , bracket
266 , bracket_
267 , finally
268 -- Auxiliary API
269 , expectTimeout
270 , receiveChanTimeout
271 , spawnAsync
272 -- Reconnecting
273 , reconnect
274 , reconnectPort
275 )
276import Control.Distributed.Process.Node (forkProcess)
278-- INTERNAL NOTES
279--
280-- 1. 'send' never fails. If you want to know that the remote process received
281-- your message, you will need to send an explicit acknowledgement. If you
282-- want to know when the remote process failed, you will need to monitor
283-- that remote process.
284--
285-- 2. 'send' may block (when the system TCP buffers are full, while we are
286-- trying to establish a connection to the remote endpoint, etc.) but its
287-- return does not imply that the remote process received the message (much
288-- less processed it)
289--
290-- 3. Message delivery is reliable and ordered. That means that if process A
291-- sends messages m1, m2, m3 to process B, B will either arrive all three
292-- messages in order (m1, m2, m3) or a prefix thereof; messages will not be
293-- 'missing' (m1, m3) or reordered (m1, m3, m2)
294--
295-- In order to guarantee (3), we stipulate that
296--
297-- 3a. We do not garbage collect connections because Network.Transport provides
298-- ordering guarantees only *per connection*.
299--
300-- 3b. Once a connection breaks, we have no way of knowing which messages
301-- arrived and which did not; hence, once a connection fails, we assume the
302-- remote process to be forever unreachable. Otherwise we might sent m1 and
303-- m2, get notified of the broken connection, reconnect, send m3, but only
304-- m1 and m3 arrive.
305--
306-- 3c. As a consequence of (3b) we should not reuse PIDs. If a process dies,
307-- we consider it forever unreachable. Hence, new processes should get new
308-- IDs or they too would be considered unreachable.
309--
310-- Main reference for Cloud Haskell is
311--
312-- [1] "Towards Haskell in the Cloud", Jeff Epstein, Andrew Black and Simon
313-- Peyton-Jones.
314-- http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/remote.pdf
315--
316-- The precise semantics for message passing is based on
317--
318-- [2] "A Unified Semantics for Future Erlang", Hans Svensson, Lars-Ake Fredlund
319-- and Clara Benac Earle (not freely available online, unfortunately)
320--
321-- Some pointers to related documentation about Erlang, for comparison and
322-- inspiration:
323--
324-- [3] "Programming Distributed Erlang Applications: Pitfalls and Recipes",
325-- Hans Svensson and Lars-Ake Fredlund
326-- http://man.lupaworld.com/content/develop/p37-svensson.pdf
327-- [4] The Erlang manual, sections "Message Sending" and "Send"
328-- http://www.erlang.org/doc/reference_manual/processes.html#id82409
329-- http://www.erlang.org/doc/reference_manual/expressions.html#send
330-- [5] Questions "Is the order of message reception guaranteed?" and
331-- "If I send a message, is it guaranteed to reach the receiver?" of
332-- the Erlang FAQ
333-- http://www.erlang.org/faq/academic.html
334-- [6] "Delivery of Messages", post on erlang-questions
335-- http://erlang.org/pipermail/erlang-questions/2012-February/064767.html
337--------------------------------------------------------------------------------
338-- Primitives are defined in a separate module; here we only define derived --
339-- constructs --
340--------------------------------------------------------------------------------
342-- | Spawn a process
343--
344-- For more information about 'Closure', see
345-- "Control.Distributed.Process.Closure".
346--
347-- See also 'call'.
348spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
349spawn nid proc = do
350 us <- getSelfPid
351 mRef <- monitorNode nid
352 sRef <- spawnAsync nid (cpDelay us proc)
353 receiveWait [
354 matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do
355 unmonitor mRef
356 send pid ()
357 return pid
358 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ ->
359 return (nullProcessId nid)
360 ]
362-- | Spawn a process and link to it
363--
364-- Note that this is just the sequential composition of 'spawn' and 'link'.
365-- (The "Unified" semantics that underlies Cloud Haskell does not even support
366-- a synchronous link operation)
367spawnLink :: NodeId -> Closure (Process ()) -> Process ProcessId
368spawnLink nid proc = do
369 pid <- spawn nid proc
370 link pid
371 return pid
373-- | Like 'spawnLink', but monitor the spawned process
374spawnMonitor :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef)
375spawnMonitor nid proc = do
376 pid <- spawn nid proc
377 ref <- monitor pid
378 return (pid, ref)
380-- | Run a process remotely and wait for it to reply
381--
382-- We monitor the remote process: if it dies before it can send a reply, we die
383-- too.
384--
385-- For more information about 'Static', 'SerializableDict', and 'Closure', see
386-- "Control.Distributed.Process.Closure".
387--
388-- See also 'spawn'.
389call :: Serializable a
390 => Static (SerializableDict a)
391 -> NodeId
392 -> Closure (Process a)
393 -> Process a
394call dict nid proc = do
395 us <- getSelfPid
396 (pid, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)
397 -- We are guaranteed to receive the reply before the monitor notification
398 -- (if a reply is sent at all)
399 -- NOTE: This might not be true if we switch to unreliable delivery.
400 mResult <- receiveWait
401 [ match (return . Right)
402 , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)
403 (\(ProcessMonitorNotification _ _ reason) -> return (Left reason))
404 ]
405 case mResult of
406 Right a -> do
407 -- Wait for the monitor message so that we the mailbox doesn't grow
408 receiveWait
409 [ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)
410 (\(ProcessMonitorNotification {}) -> return ())
411 ]
412 -- Clean up connection to pid
413 reconnect pid
414 return a
415 Left err ->
416 fail $ "call: remote process died: " ++ show err
418-- | Spawn a child process, have the child link to the parent and the parent
419-- monitor the child
420spawnSupervised :: NodeId
421 -> Closure (Process ())
422 -> Process (ProcessId, MonitorRef)
423spawnSupervised nid proc = do
424 us <- getSelfPid
425 them <- spawn nid (cpLink us `seqCP` proc)
426 ref <- monitor them
427 return (them, ref)
429-- | Spawn a new process, supplying it with a new 'ReceivePort' and return
430-- the corresponding 'SendPort'.
431spawnChannel :: forall a. Typeable a => Static (SerializableDict a)
432 -> NodeId
433 -> Closure (ReceivePort a -> Process ())
434 -> Process (SendPort a)
435spawnChannel dict nid proc = do
436 us <- getSelfPid
437 _ <- spawn nid (go us)
438 expect
439 where
440 go :: ProcessId -> Closure (Process ())
441 go pid = cpNewChan dict
442 `bindCP`
443 (cpSend (sdictSendPort dict) pid `splitCP` proc)
444 `bindCP`
445 (idCP `closureCompose` staticClosure sndStatic)
447--------------------------------------------------------------------------------
448-- Local versions of spawn --
449--------------------------------------------------------------------------------
451-- | Spawn a process on the local node
452spawnLocal :: Process () -> Process ProcessId
453spawnLocal proc = do
454 node <- processNode <$> ask
455 liftIO $ forkProcess node proc
457-- | Create a new typed channel, spawn a process on the local node, passing it
458-- the receive port, and return the send port
459spawnChannelLocal :: Serializable a
460 => (ReceivePort a -> Process ())
461 -> Process (SendPort a)
462spawnChannelLocal proc = do
463 node <- processNode <$> ask
464 liftIO $ do
465 mvar <- newEmptyMVar
466 _ <- forkProcess node $ do
467 -- It is important that we allocate the new channel in the new process,
468 -- because otherwise it will be associated with the wrong process ID
469 (sport, rport) <- newChan
470 liftIO $ putMVar mvar sport
471 proc rport
472 takeMVar mvar