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