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