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