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