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