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