git.iko.soy/distributed-process
on 1dcdfa5 help sign in
distributed-process/src/Control/Distributed/Process.hs
384 lines 9.9 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 (see
6<http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/>),
7although some of the details are different. The precise message passing
8semantics are based on /A unified semantics for future Erlang/ by Hans
9Svensson, Lars-Åke Fredlund and Clara Benac Earle.
11For a detailed description of the package and other reference materials,
12please see the distributed-process wiki page on github:
13<https://github.com/haskell-distributed/distributed-process/wiki>.
15-}
16module Control.Distributed.Process
17 ( -- * Basic types
18 ProcessId
19 , NodeId
20 , Process
21 , SendPortId
22 , processNodeId
23 , sendPortProcessId
24 , liftIO -- Reexported for convenience
25 -- * Basic messaging
26 , send
27 , expect
28 , expectTimeout
29 -- * Channels
30 , ReceivePort
31 , SendPort
32 , sendPortId
33 , newChan
34 , sendChan
35 , receiveChan
36 , receiveChanTimeout
37 , mergePortsBiased
38 , mergePortsRR
39 -- * Unsafe messaging variants
40 , unsafeSend
41 , unsafeSendChan
42 , unsafeNSend
43 , unsafeWrapMessage
44 -- * Advanced messaging
45 , Match
46 , receiveWait
47 , receiveTimeout
48 , match
49 , matchIf
50 , matchUnknown
51 , matchAny
52 , matchAnyIf
53 , matchChan
54 , Message
55 , matchMessage
56 , matchMessageIf
57 , isEncoded
58 , wrapMessage
59 , unwrapMessage
60 , handleMessage
61 , handleMessageIf
62 , handleMessage_
63 , handleMessageIf_
64 , forward
65 , delegate
66 , relay
67 , proxy
68 -- * Process management
69 , spawn
70 , call
71 , terminate
72 , die
73 , kill
74 , exit
75 , catchExit
76 , catchesExit
77 , ProcessTerminationException(..)
78 , ProcessRegistrationException(..)
79 , SpawnRef
80 , getSelfPid
81 , getSelfNode
82 , ProcessInfo(..)
83 , getProcessInfo
84 -- * Monitoring and linking
85 , link
86 , linkNode
87 , linkPort
88 , unlink
89 , unlinkNode
90 , unlinkPort
91 , monitor
92 , monitorNode
93 , monitorPort
94 , unmonitor
95 , withMonitor
96 , MonitorRef -- opaque
97 , ProcessLinkException(..)
98 , NodeLinkException(..)
99 , PortLinkException(..)
100 , ProcessMonitorNotification(..)
101 , NodeMonitorNotification(..)
102 , PortMonitorNotification(..)
103 , DiedReason(..)
104 -- * Closures
105 , Closure
106 , closure
107 , Static
108 , unStatic
109 , unClosure
110 , RemoteTable
111 -- * Logging
112 , say
113 -- * Registry
114 , register
115 , reregister
116 , unregister
117 , whereis
118 , nsend
119 , registerRemoteAsync
120 , reregisterRemoteAsync
121 , unregisterRemoteAsync
122 , whereisRemoteAsync
123 , nsendRemote
124 , WhereIsReply(..)
125 , RegisterReply(..)
126 -- * Exception handling
127 , catch
128 , Handler(..)
129 , catches
130 , try
131 , mask
132 , onException
133 , bracket
134 , bracket_
135 , finally
136 -- * Auxiliary API
137 , spawnAsync
138 , spawnSupervised
139 , spawnLink
140 , spawnMonitor
141 , spawnChannel
142 , DidSpawn(..)
143 -- * Local versions of 'spawn'
144 , spawnLocal
145 , spawnChannelLocal
146 -- * Reconnecting
147 , reconnect
148 , reconnectPort
149 ) where
151#if ! MIN_VERSION_base(4,6,0)
152import Prelude hiding (catch)
153#endif
155import Control.Monad.IO.Class (liftIO)
156import Control.Applicative ((<$>))
157import Control.Monad.Reader (ask)
158import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
159import Control.Distributed.Static
160 ( Closure
161 , closure
162 , Static
163 , RemoteTable
164 )
165import Control.Distributed.Process.Internal.Types
166 ( NodeId(..)
167 , ProcessId(..)
168 , Process(..)
169 , MonitorRef(..)
170 , ProcessMonitorNotification(..)
171 , NodeMonitorNotification(..)
172 , PortMonitorNotification(..)
173 , ProcessLinkException(..)
174 , NodeLinkException(..)
175 , PortLinkException(..)
176 , ProcessRegistrationException(..)
177 , DiedReason(..)
178 , SpawnRef(..)
179 , DidSpawn(..)
180 , SendPort(..)
181 , ReceivePort(..)
182 , SendPortId(..)
183 , WhereIsReply(..)
184 , RegisterReply(..)
185 , LocalProcess(processNode)
186 , Message
187 )
188import Control.Distributed.Process.Serializable (Serializable)
189import Control.Distributed.Process.Internal.Primitives
190 ( -- Basic messaging
191 send
192 , expect
193 -- Channels
194 , newChan
195 , sendChan
196 , receiveChan
197 , mergePortsBiased
198 , mergePortsRR
199 , unsafeSend
200 , unsafeSendChan
201 , unsafeNSend
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 , isEncoded
215 , wrapMessage
216 , unsafeWrapMessage
217 , unwrapMessage
218 , handleMessage
219 , handleMessageIf
220 , handleMessage_
221 , handleMessageIf_
222 , forward
223 , delegate
224 , relay
225 , proxy
226 -- Process management
227 , terminate
228 , ProcessTerminationException(..)
229 , die
230 , exit
231 , catchExit
232 , catchesExit
233 , kill
234 , getSelfPid
235 , getSelfNode
236 , ProcessInfo(..)
237 , getProcessInfo
238 -- Monitoring and linking
239 , link
240 , linkNode
241 , linkPort
242 , unlink
243 , unlinkNode
244 , unlinkPort
245 , monitor
246 , monitorNode
247 , monitorPort
248 , unmonitor
249 , withMonitor
250 -- Logging
251 , say
252 -- Registry
253 , register
254 , reregister
255 , unregister
256 , whereis
257 , nsend
258 , registerRemoteAsync
259 , reregisterRemoteAsync
260 , unregisterRemoteAsync
261 , whereisRemoteAsync
262 , nsendRemote
263 -- Closures
264 , unStatic
265 , unClosure
266 -- Exception handling
267 , catch
268 , Handler(..)
269 , catches
270 , try
271 , mask
272 , onException
273 , bracket
274 , bracket_
275 , finally
276 -- Auxiliary API
277 , expectTimeout
278 , receiveChanTimeout
279 , spawnAsync
280 -- Reconnecting
281 , reconnect
282 , reconnectPort
283 )
284import Control.Distributed.Process.Node (forkProcess)
285import Control.Distributed.Process.Internal.Spawn
286 ( -- Spawning Processes/Channels
287 spawn
288 , spawnLink
289 , spawnMonitor
290 , spawnChannel
291 , spawnSupervised
292 , call
293 )
295-- INTERNAL NOTES
296--
297-- 1. 'send' never fails. If you want to know that the remote process received
298-- your message, you will need to send an explicit acknowledgement. If you
299-- want to know when the remote process failed, you will need to monitor
300-- that remote process.
301--
302-- 2. 'send' may block (when the system TCP buffers are full, while we are
303-- trying to establish a connection to the remote endpoint, etc.) but its
304-- return does not imply that the remote process received the message (much
305-- less processed it). When 'send' targets a local process, it is dispatched
306-- via the node controller however, in which cases it will /not/ block. This
307-- isn't part of the semantics, so it should be ok. The ordering is maintained
308-- because the ctrlChannel still has FIFO semantics with regards interactions
309-- between two disparate forkIO threads.
310--
311-- 3. Message delivery is reliable and ordered. That means that if process A
312-- sends messages m1, m2, m3 to process B, B will either arrive all three
313-- messages in order (m1, m2, m3) or a prefix thereof; messages will not be
314-- 'missing' (m1, m3) or reordered (m1, m3, m2)
315--
316-- In order to guarantee (3), we stipulate that
317--
318-- 3a. We do not garbage collect connections because Network.Transport provides
319-- ordering guarantees only *per connection*.
320--
321-- 3b. Once a connection breaks, we have no way of knowing which messages
322-- arrived and which did not; hence, once a connection fails, we assume the
323-- remote process to be forever unreachable. Otherwise we might sent m1 and
324-- m2, get notified of the broken connection, reconnect, send m3, but only
325-- m1 and m3 arrive.
326--
327-- 3c. As a consequence of (3b) we should not reuse PIDs. If a process dies,
328-- we consider it forever unreachable. Hence, new processes should get new
329-- IDs or they too would be considered unreachable.
330--
331-- Main reference for Cloud Haskell is
332--
333-- [1] "Towards Haskell in the Cloud", Jeff Epstein, Andrew Black and Simon
334-- Peyton-Jones.
335-- http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/remote.pdf
336--
337-- The precise semantics for message passing is based on
338--
339-- [2] "A Unified Semantics for Future Erlang", Hans Svensson, Lars-Ake Fredlund
340-- and Clara Benac Earle (not freely available online, unfortunately)
341--
342-- Some pointers to related documentation about Erlang, for comparison and
343-- inspiration:
344--
345-- [3] "Programming Distributed Erlang Applications: Pitfalls and Recipes",
346-- Hans Svensson and Lars-Ake Fredlund
347-- http://man.lupaworld.com/content/develop/p37-svensson.pdf
348-- [4] The Erlang manual, sections "Message Sending" and "Send"
349-- http://www.erlang.org/doc/reference_manual/processes.html#id82409
350-- http://www.erlang.org/doc/reference_manual/expressions.html#send
351-- [5] Questions "Is the order of message reception guaranteed?" and
352-- "If I send a message, is it guaranteed to reach the receiver?" of
353-- the Erlang FAQ
354-- http://www.erlang.org/faq/academic.html
355-- [6] "Delivery of Messages", post on erlang-questions
356-- http://erlang.org/pipermail/erlang-questions/2012-February/064767.html
358--------------------------------------------------------------------------------
359-- Local versions of spawn --
360--------------------------------------------------------------------------------
362-- | Spawn a process on the local node
363spawnLocal :: Process () -> Process ProcessId
364spawnLocal proc = do
365 node <- processNode <$> ask
366 liftIO $ forkProcess node proc
368-- | Create a new typed channel, spawn a process on the local node, passing it
369-- the receive port, and return the send port
370spawnChannelLocal :: Serializable a
371 => (ReceivePort a -> Process ())
372 -> Process (SendPort a)
373spawnChannelLocal proc = do
374 node <- processNode <$> ask
375 liftIO $ do
376 mvar <- newEmptyMVar
377 _ <- forkProcess node $ do
378 -- It is important that we allocate the new channel in the new process,
379 -- because otherwise it will be associated with the wrong process ID
380 (sport, rport) <- newChan
381 liftIO $ putMVar mvar sport
382 proc rport
383 takeMVar mvar