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