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