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