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