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