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