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