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