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