git.iko.soy/distributed-process
on fix-spawnChan help sign in
distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs
991 lines 39 kB history raw
1-- | Cloud Haskell primitives
2--
3-- We define these in a separate module so that we don't have to rely on
4-- the closure combinators
5module Control.Distributed.Process.Internal.Primitives
6 ( -- * Basic messaging
7 send
8 , expect
9 -- * Channels
10 , newChan
11 , sendChan
12 , receiveChan
13 , mergePortsBiased
14 , mergePortsRR
15 -- * Advanced messaging
16 , Match
17 , receiveWait
18 , receiveTimeout
19 , match
20 , matchIf
21 , matchUnknown
22 , AbstractMessage(..)
23 , matchAny
24 , matchAnyIf
25 , matchChan
26 -- * Process management
27 , terminate
28 , ProcessTerminationException(..)
29 , die
30 , kill
31 , exit
32 , catchExit
33 , catchesExit
34 -- keep the exception constructor hidden, so that handling exit
35 -- reasons /must/ take place via the 'catchExit' family of primitives
36 , ProcessExitException()
37 , getSelfPid
38 , getSelfNode
39 , ProcessInfo(..)
40 , getProcessInfo
41 -- * Monitoring and linking
42 , link
43 , unlink
44 , monitor
45 , unmonitor
46 , withMonitor
47 -- * Logging
48 , say
49 -- * Registry
50 , register
51 , reregister
52 , unregister
53 , whereis
54 , nsend
55 , registerRemoteAsync
56 , reregisterRemoteAsync
57 , unregisterRemoteAsync
58 , whereisRemoteAsync
59 , nsendRemote
60 -- * Closures
61 , unClosure
62 , unStatic
63 -- * Exception handling
64 , catch
65 , Handler(..)
66 , catches
67 , try
68 , mask
69 , onException
70 , bracket
71 , bracket_
72 , finally
73 -- * Auxiliary API
74 , expectTimeout
75 , receiveChanTimeout
76 , spawnAsync
77 , linkNode
78 , linkPort
79 , unlinkNode
80 , unlinkPort
81 , monitorNode
82 , monitorPort
83 -- * Reconnecting
84 , reconnect
85 , reconnectPort
86 -- * Tracing/Debugging
87 , trace
88 ) where
90#if ! MIN_VERSION_base(4,6,0)
91import Prelude hiding (catch)
92#endif
94import Data.Binary (decode)
95import Data.Time.Clock (getCurrentTime)
96import Data.Time.Format (formatTime)
97import System.Locale (defaultTimeLocale)
98import System.Timeout (timeout)
99import Control.Monad (when)
100import Control.Monad.Reader (ask)
101import Control.Monad.IO.Class (MonadIO, liftIO)
102import Control.Applicative ((<$>))
103import Control.Exception (Exception(..), throw, throwIO, SomeException)
104import qualified Control.Exception as Ex (catch, mask, try)
105import Control.Distributed.Process.Internal.StrictMVar
106 ( StrictMVar
107 , modifyMVar
108 , modifyMVar_
109 )
110import Control.Concurrent.Chan (writeChan)
111import Control.Concurrent.STM
112 ( STM
113 , TVar
114 , atomically
115 , orElse
116 , newTVar
117 , readTVar
118 , writeTVar
119 )
120import Control.Distributed.Process.Internal.CQueue
121 ( dequeue
122 , BlockSpec(..)
123 , MatchOn(..)
124 )
125import Control.Distributed.Process.Serializable (Serializable, fingerprint)
126import Data.Accessor ((^.), (^:), (^=))
127import Control.Distributed.Static (Closure, Static)
128import Data.Rank1Typeable (Typeable)
129import qualified Control.Distributed.Static as Static (unstatic, unclosure)
130import Control.Distributed.Process.Internal.Types
131 ( NodeId(..)
132 , ProcessId(..)
133 , LocalNode(..)
134 , LocalProcess(..)
135 , Process(..)
136 , Message(..)
137 , MonitorRef(..)
138 , SpawnRef(..)
139 , NCMsg(..)
140 , ProcessSignal(..)
141 , monitorCounter
142 , spawnCounter
143 , SendPort(..)
144 , ReceivePort(..)
145 , channelCounter
146 , typedChannelWithId
147 , TypedChannel(..)
148 , SendPortId(..)
149 , Identifier(..)
150 , ProcessExitException(..)
151 , DidUnmonitor(..)
152 , DidUnlinkProcess(..)
153 , DidUnlinkNode(..)
154 , DidUnlinkPort(..)
155 , WhereIsReply(..)
156 , RegisterReply(..)
157 , ProcessRegistrationException(..)
158 , ProcessInfo(..)
159 , ProcessInfoNone(..)
160 , createMessage
161 , runLocalProcess
162 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
163 , LocalProcessState
164 , LocalSendPortId
165 , messageToPayload
166 )
167import Control.Distributed.Process.Internal.Messaging
168 ( sendMessage
169 , sendBinary
170 , sendPayload
171 , disconnect
172 )
173import qualified Control.Distributed.Process.Internal.Trace as Trace
174import Control.Distributed.Process.Internal.WeakTQueue
175 ( newTQueueIO
176 , readTQueue
177 , mkWeakTQueue
178 )
180--------------------------------------------------------------------------------
181-- Basic messaging --
182--------------------------------------------------------------------------------
184-- | Send a message
185send :: Serializable a => ProcessId -> a -> Process ()
186-- This requires a lookup on every send. If we want to avoid that we need to
187-- modify serializable to allow for stateful (IO) deserialization
188send them msg = do
189 proc <- ask
190 liftIO $ sendMessage (processNode proc)
191 (ProcessIdentifier (processId proc))
192 (ProcessIdentifier them)
193 NoImplicitReconnect
194 msg
196-- | Wait for a message of a specific type
197expect :: forall a. Serializable a => Process a
198expect = receiveWait [match return]
200--------------------------------------------------------------------------------
201-- Channels --
202--------------------------------------------------------------------------------
204-- | Create a new typed channel
205newChan :: Serializable a => Process (SendPort a, ReceivePort a)
206newChan = do
207 proc <- ask
208 liftIO . modifyMVar (processState proc) $ \st -> do
209 let lcid = st ^. channelCounter
210 let cid = SendPortId { sendPortProcessId = processId proc
211 , sendPortLocalId = lcid
212 }
213 let sport = SendPort cid
214 chan <- liftIO newTQueueIO
215 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
216 let rport = ReceivePort $ readTQueue chan
217 let tch = TypedChannel chan'
218 return ( (channelCounter ^: (+ 1))
219 . (typedChannelWithId lcid ^= Just tch)
220 $ st
221 , (sport, rport)
222 )
223 where
224 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
225 finalizer st lcid = modifyMVar_ st $
226 return . (typedChannelWithId lcid ^= Nothing)
228-- | Send a message on a typed channel
229sendChan :: Serializable a => SendPort a -> a -> Process ()
230sendChan (SendPort cid) msg = do
231 proc <- ask
232 liftIO $ sendBinary (processNode proc)
233 (ProcessIdentifier (processId proc))
234 (SendPortIdentifier cid)
235 NoImplicitReconnect
236 msg
238-- | Wait for a message on a typed channel
239receiveChan :: Serializable a => ReceivePort a -> Process a
240receiveChan = liftIO . atomically . receiveSTM
242-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
243-- non-blocking check for a message.
244receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
245receiveChanTimeout 0 ch = liftIO . atomically $
246 (Just <$> receiveSTM ch) `orElse` return Nothing
247receiveChanTimeout n ch = liftIO . timeout n . atomically $
248 receiveSTM ch
250-- | Merge a list of typed channels.
251--
252-- The result port is left-biased: if there are messages available on more
253-- than one port, the first available message is returned.
254mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
255mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
257-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
258-- left-biased)
259mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
260mergePortsRR = \ps -> do
261 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
262 return $ ReceivePort (rr psVar)
263 where
264 rotate :: [a] -> [a]
265 rotate [] = []
266 rotate (x:xs) = xs ++ [x]
268 rr :: TVar [STM a] -> STM a
269 rr psVar = do
270 ps <- readTVar psVar
271 a <- foldr1 orElse ps
272 writeTVar psVar (rotate ps)
273 return a
275--------------------------------------------------------------------------------
276-- Advanced messaging --
277--------------------------------------------------------------------------------
279-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
280newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
282-- | Test the matches in order against each message in the queue
283receiveWait :: [Match b] -> Process b
284receiveWait ms = do
285 queue <- processQueue <$> ask
286 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
287 proc
289-- | Like 'receiveWait' but with a timeout.
290--
291-- If the timeout is zero do a non-blocking check for matching messages. A
292-- non-zero timeout is applied only when waiting for incoming messages (that is,
293-- /after/ we have checked the messages that are already in the mailbox).
294receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
295receiveTimeout t ms = do
296 queue <- processQueue <$> ask
297 let blockSpec = if t == 0 then NonBlocking else Timeout t
298 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
299 case mProc of
300 Nothing -> return Nothing
301 Just proc -> Just <$> proc
303matchChan :: ReceivePort a -> (a -> Process b) -> Match b
304matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
306-- | Match against any message of the right type
307match :: forall a b. Serializable a => (a -> Process b) -> Match b
308match = matchIf (const True)
310-- | Match against any message of the right type that satisfies a predicate
311matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
312matchIf c p = Match $ MatchMsg $ \msg ->
313 case messageFingerprint msg == fingerprint (undefined :: a) of
314 True | c decoded -> Just (p decoded)
315 where
316 decoded :: a
317 -- Make sure the value is fully decoded so that we don't hang to
318 -- bytestrings when the process calling 'matchIf' doesn't process
319 -- the values immediately
320 !decoded = decode (messageEncoding msg)
321 _ -> Nothing
323-- | Represents a received message and provides two basic operations on it.
324data AbstractMessage = AbstractMessage {
325 forward :: ProcessId -> Process () -- ^ forward the message to @ProcessId@
326 , maybeHandleMessage :: forall a b. (Serializable a)
327 => (a -> Process b) -> Process (Maybe b) {- ^ Handle the message.
328 If the type of the message matches the type of the first argument to
329 the supplied expression, then the expression will be evaluated against
330 it. If this runtime type checking fails, then @Nothing@ will be returned
331 to indicate the fact. If the check succeeds and evaluation proceeds
332 however, the resulting value with be wrapped with @Just@.
333 -}
334 }
336-- | Match against an arbitrary message. 'matchAny' removes the first available
337-- message from the process mailbox, and via the 'AbstractMessage' type,
338-- supports forwarding /or/ handling the message /if/ it is of the correct
339-- type. If /not/ of the right type, then the 'AbstractMessage'
340-- @maybeHandleMessage@ function will not evaluate the supplied expression,
341-- /but/ the message will still have been removed from the process mailbox!
342--
343matchAny :: forall b. (AbstractMessage -> Process b) -> Match b
344matchAny p = Match $ MatchMsg $ Just . p . abstract
346-- | Match against an arbitrary message. 'matchAnyIf' will /only/ remove the
347-- message from the process mailbox, /if/ the supplied condition matches. The
348-- success (or failure) of runtime type checks in @maybeHandleMessage@ does not
349-- count here, i.e., if the condition evaluates to @True@ then the message will
350-- be removed from the process mailbox and decoded, but that does /not/
351-- guarantee that an expression passed to @maybeHandleMessage@ will pass the
352-- runtime type checks and therefore be evaluated. If the types do not match
353-- up, then @maybeHandleMessage@ returns 'Nothing'.
354matchAnyIf :: forall a b. (Serializable a)
355 => (a -> Bool)
356 -> (AbstractMessage -> Process b)
357 -> Match b
358matchAnyIf c p = Match $ MatchMsg $ \msg ->
359 case messageFingerprint msg == fingerprint (undefined :: a) of
360 True | c decoded -> Just (p (abstract msg))
361 where
362 decoded :: a
363 -- Make sure the value is fully decoded so that we don't hang to
364 -- bytestrings when the calling process doesn't evaluate immediately
365 !decoded = decode (messageEncoding msg)
366 _ -> Nothing
368abstract :: Message -> AbstractMessage
369abstract msg = AbstractMessage {
370 forward = \them -> do
371 proc <- ask
372 liftIO $ sendPayload (processNode proc)
373 (ProcessIdentifier (processId proc))
374 (ProcessIdentifier them)
375 NoImplicitReconnect
376 (messageToPayload msg)
377 , maybeHandleMessage = \(proc :: (a -> Process b)) -> do
378 case messageFingerprint msg == fingerprint (undefined :: a) of
379 True -> do { r <- proc (decoded :: a); return (Just r) }
380 where
381 decoded :: a
382 !decoded = decode (messageEncoding msg)
383 _ -> return Nothing
384 }
386-- | Remove any message from the queue
387matchUnknown :: Process b -> Match b
388matchUnknown p = Match $ MatchMsg (const (Just p))
390--------------------------------------------------------------------------------
391-- Process management --
392--------------------------------------------------------------------------------
394-- | Thrown by 'terminate'
395data ProcessTerminationException = ProcessTerminationException
396 deriving (Show, Typeable)
398instance Exception ProcessTerminationException
400-- | Terminate immediately (throws a ProcessTerminationException)
401terminate :: Process a
402terminate = liftIO $ throwIO ProcessTerminationException
404-- [Issue #110]
405-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
406die :: Serializable a => a -> Process b
407die reason = do
408 pid <- getSelfPid
409 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
411-- | Forceful request to kill a process. Where 'exit' provides an exception
412-- that can be caught and handled, 'kill' throws an unexposed exception type
413-- which cannot be handled explicitly (by type).
414kill :: ProcessId -> String -> Process ()
415-- NOTE: We send the message to our local node controller, which will then
416-- forward it to a remote node controller (if applicable). Sending it directly
417-- to a remote node controller means that that the message may overtake a
418-- 'monitor' or 'link' request.
419kill them reason = sendCtrlMsg Nothing (Kill them reason)
421-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
422-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
423-- manner can be handled using the 'catchExit' family of functions.
424exit :: Serializable a => ProcessId -> a -> Process ()
425-- NOTE: We send the message to our local node controller, which will then
426-- forward it to a remote node controller (if applicable). Sending it directly
427-- to a remote node controller means that that the message may overtake a
428-- 'monitor' or 'link' request.
429exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
431-- | Catches 'ProcessExitException'. The handler will not be applied unless its
432-- type matches the encoded data stored in the exception (see the /reason/
433-- argument given to the 'exit' primitive). If the handler cannot be applied,
434-- the exception will be re-thrown.
435--
436-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
437-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
438catchExit :: forall a b . (Show a, Serializable a)
439 => Process b
440 -> (ProcessId -> a -> Process b)
441 -> Process b
442catchExit act exitHandler = catch act handleExit
443 where
444 handleExit ex@(ProcessExitException from msg) =
445 if messageFingerprint msg == fingerprint (undefined :: a)
446 then exitHandler from decoded
447 else liftIO $ throwIO ex
448 where
449 decoded :: a
450 -- Make sure the value is fully decoded so that we don't hang to
451 -- bytestrings if the caller doesn't use the value immediately
452 !decoded = decode (messageEncoding msg)
454-- | Lift 'Control.Exception.catches' (almost).
455--
456-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
457-- message, a handler must accept an input of the expected type. In order to
458-- handle a list of potentially different handlers (and therefore input types),
459-- a handler passed to 'catchesExit' must accept 'AbstractMessage' and return
460-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
461--
462-- See 'maybeHandleMessage' and 'AsbtractMessage' for more details.
463catchesExit :: Process b
464 -> [(ProcessId -> AbstractMessage -> (Process (Maybe b)))]
465 -> Process b
466catchesExit act handlers = catch act ((flip handleExit) handlers)
467 where
468 handleExit :: ProcessExitException
469 -> [(ProcessId -> AbstractMessage -> Process (Maybe b))]
470 -> Process b
471 handleExit ex [] = liftIO $ throwIO ex
472 handleExit ex@(ProcessExitException from msg) (h:hs) = do
473 r <- h from (abstract msg)
474 case r of
475 Nothing -> handleExit ex hs
476 Just p -> return p
478-- | Our own process ID
479getSelfPid :: Process ProcessId
480getSelfPid = processId <$> ask
482-- | Get the node ID of our local node
483getSelfNode :: Process NodeId
484getSelfNode = localNodeId . processNode <$> ask
486-- | Get information about the specified process
487getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
488getProcessInfo pid =
489 let them = processNodeId pid in do
490 us <- getSelfNode
491 dest <- mkNode them us
492 sendCtrlMsg dest $ GetInfo pid
493 receiveWait [
494 match (\(p :: ProcessInfo) -> return $ Just p)
495 , match (\(_ :: ProcessInfoNone) -> return Nothing)
496 ]
497 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
498 mkNode them us = case them == us of
499 True -> return Nothing
500 _ -> return $ Just them
502--------------------------------------------------------------------------------
503-- Monitoring and linking --
504--------------------------------------------------------------------------------
506-- | Link to a remote process (asynchronous)
507--
508-- When process A links to process B (that is, process A calls
509-- @link pidB@) then an asynchronous exception will be thrown to process A
510-- when process B terminates (normally or abnormally), or when process A gets
511-- disconnected from process B. Although it is /technically/ possible to catch
512-- these exceptions, chances are if you find yourself trying to do so you should
513-- probably be using 'monitor' rather than 'link'. In particular, code such as
514--
515-- > link pidB -- Link to process B
516-- > expect -- Wait for a message from process B
517-- > unlink pidB -- Unlink again
518--
519-- doesn't quite do what one might expect: if process B sends a message to
520-- process A, and /subsequently terminates/, then process A might or might not
521-- be terminated too, depending on whether the exception is thrown before or
522-- after the 'unlink' (i.e., this code has a race condition).
523--
524-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
525-- call to 'link' has no effect.
526--
527-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
528-- Linking makes no distinction between normal and abnormal termination of
529-- the remote process.
530link :: ProcessId -> Process ()
531link = sendCtrlMsg Nothing . Link . ProcessIdentifier
533-- | Monitor another process (asynchronous)
534--
535-- When process A monitors process B (that is, process A calls
536-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
537-- when process B terminates (normally or abnormally), or when process A gets
538-- disconnected from process B. You receive this message like any other (using
539-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
540-- 'DiedDisconnect', etc.).
541--
542-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
543-- multiple monitors are set up, multiple notifications will be delivered
544-- and monitors can be disabled individually using 'unmonitor'.
545monitor :: ProcessId -> Process MonitorRef
546monitor = monitor' . ProcessIdentifier
548-- | Establishes temporary monitoring of another process.
549--
550-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
551-- of @code@. Note: although monitoring is no longer active when
552-- @withMonitor@ returns, there might still be unreceived monitor
553-- messages in the queue.
554--
555withMonitor :: ProcessId -> Process a -> Process a
556withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
557 -- unmonitor blocks waiting for the response, so there's a possibility
558 -- that an exception might interrupt withMonitor before the unmonitor
559 -- has completed. I think that's better than making the unmonitor
560 -- uninterruptible.
562-- | Remove a link
563--
564-- This is synchronous in the sense that once it returns you are guaranteed
565-- that no exception will be raised if the remote process dies. However, it is
566-- asynchronous in the sense that we do not wait for a response from the remote
567-- node.
568unlink :: ProcessId -> Process ()
569unlink pid = do
570 unlinkAsync pid
571 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
572 (\_ -> return ())
573 ]
575-- | Remove a node link
576--
577-- This has the same synchronous/asynchronous nature as 'unlink'.
578unlinkNode :: NodeId -> Process ()
579unlinkNode nid = do
580 unlinkNodeAsync nid
581 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
582 (\_ -> return ())
583 ]
585-- | Remove a channel (send port) link
586--
587-- This has the same synchronous/asynchronous nature as 'unlink'.
588unlinkPort :: SendPort a -> Process ()
589unlinkPort sport = do
590 unlinkPortAsync sport
591 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
592 (\_ -> return ())
593 ]
595-- | Remove a monitor
596--
597-- This has the same synchronous/asynchronous nature as 'unlink'.
598unmonitor :: MonitorRef -> Process ()
599unmonitor ref = do
600 unmonitorAsync ref
601 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
602 (\_ -> return ())
603 ]
605--------------------------------------------------------------------------------
606-- Exception handling --
607--------------------------------------------------------------------------------
609-- | Lift 'Control.Exception.catch'
610catch :: Exception e => Process a -> (e -> Process a) -> Process a
611catch p h = do
612 lproc <- ask
613 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
615-- | Lift 'Control.Exception.try'
616try :: Exception e => Process a -> Process (Either e a)
617try p = do
618 lproc <- ask
619 liftIO $ Ex.try (runLocalProcess lproc p)
621-- | Lift 'Control.Exception.mask'
622mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
623mask p = do
624 lproc <- ask
625 liftIO $ Ex.mask $ \restore ->
626 runLocalProcess lproc (p (liftRestore lproc restore))
627 where
628 liftRestore :: LocalProcess -> (forall a. IO a -> IO a) -> (forall a. Process a -> Process a)
629 liftRestore lproc restoreIO = liftIO . restoreIO . runLocalProcess lproc
631-- | Lift 'Control.Exception.onException'
632onException :: Process a -> Process b -> Process a
633onException p what = p `catch` \e -> do _ <- what
634 liftIO $ throwIO (e :: SomeException)
636-- | Lift 'Control.Exception.bracket'
637bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
638bracket before after thing =
639 mask $ \restore -> do
640 a <- before
641 r <- restore (thing a) `onException` after a
642 _ <- after a
643 return r
645-- | Lift 'Control.Exception.bracket_'
646bracket_ :: Process a -> Process b -> Process c -> Process c
647bracket_ before after thing = bracket before (const after) (const thing)
649-- | Lift 'Control.Exception.finally'
650finally :: Process a -> Process b -> Process a
651finally a sequel = bracket_ (return ()) sequel a
653-- | You need this when using 'catches'
654data Handler a = forall e . Exception e => Handler (e -> Process a)
656instance Functor Handler where
657 fmap f (Handler h) = Handler (fmap f . h)
659-- | Lift 'Control.Exception.catches'
660catches :: Process a -> [Handler a] -> Process a
661catches proc handlers = proc `catch` catchesHandler handlers
663catchesHandler :: [Handler a] -> SomeException -> Process a
664catchesHandler handlers e = foldr tryHandler (throw e) handlers
665 where tryHandler (Handler handler) res
666 = case fromException e of
667 Just e' -> handler e'
668 Nothing -> res
670--------------------------------------------------------------------------------
671-- Auxiliary API --
672--------------------------------------------------------------------------------
674-- | Like 'expect' but with a timeout
675expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
676expectTimeout n = receiveTimeout n [match return]
678-- | Asynchronous version of 'spawn'
679--
680-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
681spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
682spawnAsync nid proc = do
683 spawnRef <- getSpawnRef
684 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
685 return spawnRef
687-- | Monitor a node (asynchronous)
688monitorNode :: NodeId -> Process MonitorRef
689monitorNode =
690 monitor' . NodeIdentifier
692-- | Monitor a typed channel (asynchronous)
693monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
694monitorPort (SendPort cid) =
695 monitor' (SendPortIdentifier cid)
697-- | Remove a monitor (asynchronous)
698unmonitorAsync :: MonitorRef -> Process ()
699unmonitorAsync =
700 sendCtrlMsg Nothing . Unmonitor
702-- | Link to a node (asynchronous)
703linkNode :: NodeId -> Process ()
704linkNode = link' . NodeIdentifier
706-- | Link to a channel (asynchronous)
707linkPort :: SendPort a -> Process ()
708linkPort (SendPort cid) =
709 link' (SendPortIdentifier cid)
711-- | Remove a process link (asynchronous)
712unlinkAsync :: ProcessId -> Process ()
713unlinkAsync =
714 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
716-- | Remove a node link (asynchronous)
717unlinkNodeAsync :: NodeId -> Process ()
718unlinkNodeAsync =
719 sendCtrlMsg Nothing . Unlink . NodeIdentifier
721-- | Remove a channel (send port) link (asynchronous)
722unlinkPortAsync :: SendPort a -> Process ()
723unlinkPortAsync (SendPort cid) =
724 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
726--------------------------------------------------------------------------------
727-- Logging --
728--------------------------------------------------------------------------------
730-- | Log a string
731--
732-- @say message@ sends a message (time, pid of the current process, message)
733-- to the process registered as 'logger'. By default, this process simply
734-- sends the string to 'stderr'. Individual Cloud Haskell backends might
735-- replace this with a different logger process, however.
736say :: String -> Process ()
737say string = do
738 now <- liftIO getCurrentTime
739 us <- getSelfPid
740 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
742--------------------------------------------------------------------------------
743-- Registry --
744--------------------------------------------------------------------------------
746-- | Register a process with the local registry (asynchronous).
747-- This version will wait until a response is gotten from the
748-- management process. The name must not already be registered.
749-- The process need not be on this node.
750-- A bad registration will result in a 'ProcessRegistrationException'
751--
752-- The process to be registered does not have to be local itself.
753register :: String -> ProcessId -> Process ()
754register = registerImpl False
756-- | Like 'register', but will replace an existing registration.
757-- The name must already be registered.
758reregister :: String -> ProcessId -> Process ()
759reregister = registerImpl True
761registerImpl :: Bool -> String -> ProcessId -> Process ()
762registerImpl force label pid = do
763 mynid <- getSelfNode
764 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
765 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
766 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
767 ]
769-- | Register a process with a remote registry (asynchronous).
770--
771-- The process to be registered does not have to live on the same remote node.
772-- Reply wil come in the form of a 'RegisterReply' message
773--
774-- See comments in 'whereisRemoteAsync'
775registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
776registerRemoteAsync nid label pid =
777 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
779reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
780reregisterRemoteAsync nid label pid =
781 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
783-- | Remove a process from the local registry (asynchronous).
784-- This version will wait until a response is gotten from the
785-- management process. The name must already be registered.
786unregister :: String -> Process ()
787unregister label = do
788 mynid <- getSelfNode
789 sendCtrlMsg Nothing (Register label mynid Nothing False)
790 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
791 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
792 ]
794-- | Deal with the result from an attempted registration or unregistration
795-- by throwing an exception if necessary
796handleRegistrationReply :: String -> Bool -> Process ()
797handleRegistrationReply label ok =
798 when (not ok) $
799 liftIO $ throwIO $ ProcessRegistrationException label
801-- | Remove a process from a remote registry (asynchronous).
802--
803-- Reply wil come in the form of a 'RegisterReply' message
804--
805-- See comments in 'whereisRemoteAsync'
806unregisterRemoteAsync :: NodeId -> String -> Process ()
807unregisterRemoteAsync nid label =
808 sendCtrlMsg (Just nid) (Register label nid Nothing False)
810-- | Query the local process registry
811whereis :: String -> Process (Maybe ProcessId)
812whereis label = do
813 sendCtrlMsg Nothing (WhereIs label)
814 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
815 (\(WhereIsReply _ mPid) -> return mPid)
816 ]
818-- | Query a remote process registry (asynchronous)
819--
820-- Reply will come in the form of a 'WhereIsReply' message.
821--
822-- There is currently no synchronous version of 'whereisRemoteAsync': if
823-- you implement one yourself, be sure to take into account that the remote
824-- node might die or get disconnect before it can respond (i.e. you should
825-- use 'monitorNode' and take appropriate action when you receive a
826-- 'NodeMonitorNotification').
827whereisRemoteAsync :: NodeId -> String -> Process ()
828whereisRemoteAsync nid label =
829 sendCtrlMsg (Just nid) (WhereIs label)
831-- | Named send to a process in the local registry (asynchronous)
832nsend :: Serializable a => String -> a -> Process ()
833nsend label msg =
834 sendCtrlMsg Nothing (NamedSend label (createMessage msg))
836-- | Named send to a process in a remote registry (asynchronous)
837nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
838nsendRemote nid label msg =
839 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
841--------------------------------------------------------------------------------
842-- Closures --
843--------------------------------------------------------------------------------
845-- | Resolve a static value
846unStatic :: Typeable a => Static a -> Process a
847unStatic static = do
848 rtable <- remoteTable . processNode <$> ask
849 case Static.unstatic rtable static of
850 Left err -> fail $ "Could not resolve static value: " ++ err
851 Right x -> return x
853-- | Resolve a closure
854unClosure :: Typeable a => Closure a -> Process a
855unClosure closure = do
856 rtable <- remoteTable . processNode <$> ask
857 case Static.unclosure rtable closure of
858 Left err -> fail $ "Could not resolve closure: " ++ err
859 Right x -> return x
861--------------------------------------------------------------------------------
862-- Reconnecting --
863--------------------------------------------------------------------------------
865-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
866-- message passing. However, when network connections get disrupted this
867-- illusion cannot always be maintained. Once a network connection breaks (even
868-- temporarily) no further communication on that connection will be possible.
869-- For example, if process A sends a message to process B, and A is then
870-- notified (by monitor notification) that it got disconnected from B, A will
871-- not be able to send any further messages to B, /unless/ A explicitly
872-- indicates that it is acceptable to attempt to reconnect to B using the
873-- Cloud Haskell 'reconnect' primitive.
874--
875-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
876-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
877-- then receives a monitor notification that its connection to B has been lost,
878-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
879-- and m3 but not m2.
880--
881-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
882-- to attempt to reconnect on the next send/. In particular, if no further
883-- communication attempts are made to B then A can use reconnect to clean up
884-- its connection to B.
885reconnect :: ProcessId -> Process ()
886reconnect them = do
887 us <- getSelfPid
888 node <- processNode <$> ask
889 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
891-- | Reconnect to a sendport. See 'reconnect' for more information.
892reconnectPort :: SendPort a -> Process ()
893reconnectPort them = do
894 us <- getSelfPid
895 node <- processNode <$> ask
896 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
898--------------------------------------------------------------------------------
899-- Debugging/Tracing --
900--------------------------------------------------------------------------------
902-- | Send a message to the internal (system) trace facility. If tracing is
903-- enabled, this will create a custom trace event. Note that several Cloud Haskell
904-- sub-systems also generate trace events for informational/debugging purposes,
905-- thus traces generated this way will not be the only output seen.
906--
907-- Just as with the "Debug.Trace" module, this is a debugging/tracing facility
908-- for use in development, and should not be used in a production setting -
909-- which is why the default behaviour is to trace to the GHC eventlog. For a
910-- general purpose logging facility, you should consider 'say'.
911--
912-- Trace events can be written to the GHC event log, a text file, or to the
913-- standard system logger process (see 'say'). The default behaviour for writing
914-- to the eventlog requires specific intervention to work, without which traces
915-- are silently dropped/ignored and no output will be generated.
916-- The GHC eventlog documentation provides information about enabling, viewing
917-- and working with event traces: <http://hackage.haskell.org/trac/ghc/wiki/EventLog>.
918--
919-- When a new local node is started, the contents of the environment variable
920-- @DISTRIBUTED_PROCESS_TRACE_FILE@ are checked for a valid file path. If this
921-- exists and the file can be opened for writing, all trace output will be directed
922-- thence. If the environment variable is empty, the path invalid, or the file
923-- unavailable for writing - e.g., because another node has already started
924-- tracing to it - then the @DISTRIBUTED_PROCESS_TRACE_CONSOLE@ environment
925-- variable is checked for /any/ non-empty value. If this is set, then all trace
926-- output will be directed to the system logger process. If neither evironment
927-- variable provides a valid trace configuration, all internal traces are written
928-- to "Debug.Trace.traceEventIO", which writes to the GHC eventlog.
929--
930-- Users of the /simplelocalnet/ Cloud Haskell backend should also note that
931-- because the trace file option only supports trace output from a single node
932-- (so as to avoid interleaving), a file trace configured for the master node will
933-- prevent slaves from tracing to the file and they will fall back to using the
934-- console/'say' or eventlog instead.
935--
936trace :: String -> Process ()
937trace s = do
938 node <- processNode <$> ask
939 liftIO $ Trace.trace (localTracer node) s
941--------------------------------------------------------------------------------
942-- Auxiliary functions --
943--------------------------------------------------------------------------------
945getMonitorRefFor :: Identifier -> Process MonitorRef
946getMonitorRefFor ident = do
947 proc <- ask
948 liftIO $ modifyMVar (processState proc) $ \st -> do
949 let counter = st ^. monitorCounter
950 return ( monitorCounter ^: (+ 1) $ st
951 , MonitorRef ident counter
952 )
954getSpawnRef :: Process SpawnRef
955getSpawnRef = do
956 proc <- ask
957 liftIO $ modifyMVar (processState proc) $ \st -> do
958 let counter = st ^. spawnCounter
959 return ( spawnCounter ^: (+ 1) $ st
960 , SpawnRef counter
961 )
963-- | Monitor a process/node/channel
964monitor' :: Identifier -> Process MonitorRef
965monitor' ident = do
966 monitorRef <- getMonitorRefFor ident
967 sendCtrlMsg Nothing $ Monitor monitorRef
968 return monitorRef
970-- | Link to a process/node/channel
971link' :: Identifier -> Process ()
972link' = sendCtrlMsg Nothing . Link
974-- Send a control message
975sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node
976 -> ProcessSignal -- ^ Message to send
977 -> Process ()
978sendCtrlMsg mNid signal = do
979 proc <- ask
980 let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc)
981 , ctrlMsgSignal = signal
982 }
983 case mNid of
984 Nothing -> do
985 liftIO $ writeChan (localCtrlChan (processNode proc)) msg
986 Just nid ->
987 liftIO $ sendBinary (processNode proc)
988 (ProcessIdentifier (processId proc))
989 (NodeIdentifier nid)
990 WithImplicitReconnect
991 msg