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