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