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