git.iko.soy/distributed-process
on 263cb13 help sign in
distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs
1095 lines 42 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 , createUnencodedMessage
167 , runLocalProcess
168 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
169 , LocalProcessState
170 , LocalSendPortId
171 , messageToPayload
172 )
173import Control.Distributed.Process.Internal.Messaging
174 ( sendMessage
175 , sendBinary
176 , sendPayload
177 , disconnect
178 )
179import qualified Control.Distributed.Process.Internal.Trace as Trace
180import Control.Distributed.Process.Internal.WeakTQueue
181 ( newTQueueIO
182 , readTQueue
183 , mkWeakTQueue
184 )
185import Unsafe.Coerce
187--------------------------------------------------------------------------------
188-- Basic messaging --
189--------------------------------------------------------------------------------
191-- | Send a message
192send :: Serializable a => ProcessId -> a -> Process ()
193-- This requires a lookup on every send. If we want to avoid that we need to
194-- modify serializable to allow for stateful (IO) deserialization
195send them msg = do
196 proc <- ask
197 let node = localNodeId (processNode proc)
198 destNode = (processNodeId them) in do
199 case destNode == node of
200 True -> sendLocal them msg
201 False -> liftIO $ sendMessage (processNode proc)
202 (ProcessIdentifier (processId proc))
203 (ProcessIdentifier them)
204 NoImplicitReconnect
205 msg
207-- | Wait for a message of a specific type
208expect :: forall a. Serializable a => Process a
209expect = receiveWait [match return]
211--------------------------------------------------------------------------------
212-- Channels --
213--------------------------------------------------------------------------------
215-- | Create a new typed channel
216newChan :: Serializable a => Process (SendPort a, ReceivePort a)
217newChan = do
218 proc <- ask
219 liftIO . modifyMVar (processState proc) $ \st -> do
220 let lcid = st ^. channelCounter
221 let cid = SendPortId { sendPortProcessId = processId proc
222 , sendPortLocalId = lcid
223 }
224 let sport = SendPort cid
225 chan <- liftIO newTQueueIO
226 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
227 let rport = ReceivePort $ readTQueue chan
228 let tch = TypedChannel chan'
229 return ( (channelCounter ^: (+ 1))
230 . (typedChannelWithId lcid ^= Just tch)
231 $ st
232 , (sport, rport)
233 )
234 where
235 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
236 finalizer st lcid = modifyMVar_ st $
237 return . (typedChannelWithId lcid ^= Nothing)
239-- | Send a message on a typed channel
240sendChan :: Serializable a => SendPort a -> a -> Process ()
241sendChan (SendPort cid) msg = do
242 proc <- ask
243 let node = localNodeId (processNode proc)
244 destNode = processNodeId (sendPortProcessId cid) in do
245 case destNode == node of
246 True -> sendChanLocal cid msg
247 False -> do
248 liftIO $ sendBinary (processNode proc)
249 (ProcessIdentifier (processId proc))
250 (SendPortIdentifier cid)
251 NoImplicitReconnect
252 msg
254-- | Wait for a message on a typed channel
255receiveChan :: Serializable a => ReceivePort a -> Process a
256receiveChan = liftIO . atomically . receiveSTM
258-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
259-- non-blocking check for a message.
260receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
261receiveChanTimeout 0 ch = liftIO . atomically $
262 (Just <$> receiveSTM ch) `orElse` return Nothing
263receiveChanTimeout n ch = liftIO . timeout n . atomically $
264 receiveSTM ch
266-- | Merge a list of typed channels.
267--
268-- The result port is left-biased: if there are messages available on more
269-- than one port, the first available message is returned.
270mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
271mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
273-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
274-- left-biased)
275mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
276mergePortsRR = \ps -> do
277 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
278 return $ ReceivePort (rr psVar)
279 where
280 rotate :: [a] -> [a]
281 rotate [] = []
282 rotate (x:xs) = xs ++ [x]
284 rr :: TVar [STM a] -> STM a
285 rr psVar = do
286 ps <- readTVar psVar
287 a <- foldr1 orElse ps
288 writeTVar psVar (rotate ps)
289 return a
291--------------------------------------------------------------------------------
292-- Advanced messaging --
293--------------------------------------------------------------------------------
295-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
296newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
298-- | Test the matches in order against each message in the queue
299receiveWait :: [Match b] -> Process b
300receiveWait ms = do
301 queue <- processQueue <$> ask
302 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
303 proc
305-- | Like 'receiveWait' but with a timeout.
306--
307-- If the timeout is zero do a non-blocking check for matching messages. A
308-- non-zero timeout is applied only when waiting for incoming messages (that is,
309-- /after/ we have checked the messages that are already in the mailbox).
310receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
311receiveTimeout t ms = do
312 queue <- processQueue <$> ask
313 let blockSpec = if t == 0 then NonBlocking else Timeout t
314 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
315 case mProc of
316 Nothing -> return Nothing
317 Just proc -> Just <$> proc
319matchChan :: ReceivePort a -> (a -> Process b) -> Match b
320matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
322-- | Match against any message of the right type
323match :: forall a b. Serializable a => (a -> Process b) -> Match b
324match = matchIf (const True)
326-- | Match against any message of the right type that satisfies a predicate
327matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
328matchIf c p = Match $ MatchMsg $ \msg ->
329 case messageFingerprint msg == fingerprint (undefined :: a) of
330 False -> Nothing
331 True -> case msg of
332 (UnencodedMessage _ m) ->
333 let m' = unsafeCoerce m :: a in
334 case (c m') of
335 True -> Just (p m')
336 False -> Nothing
337 (EncodedMessage _ _) ->
338 if (c decoded) then Just (p decoded) else Nothing
339 where
340 decoded :: a
341 -- Make sure the value is fully decoded so that we don't hang to
342 -- bytestrings when the process calling 'matchIf' doesn't process
343 -- the values immediately
344 !decoded = decode (messageEncoding msg)
346-- | Match against any message, regardless of the underlying (contained) type
347matchMessage :: (Message -> Process Message) -> Match Message
348matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
350-- | Match against any message (regardless of underlying type) that satisfies a predicate
351matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
352matchMessageIf c p = Match $ MatchMsg $ \msg ->
353 case (c msg) of
354 True -> Just (p msg)
355 False -> Nothing
357-- | Forward a raw 'Message' to the given 'ProcessId'.
358forward :: Message -> ProcessId -> Process ()
359forward msg them = do
360 proc <- ask
361 let node = localNodeId (processNode proc)
362 destNode = (processNodeId them) in do
363 case destNode == node of
364 True -> sendCtrlMsg Nothing (LocalSend them msg)
365 False -> liftIO $ sendPayload (processNode proc)
366 (ProcessIdentifier (processId proc))
367 (ProcessIdentifier them)
368 NoImplicitReconnect
369 (messageToPayload msg)
372-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
373-- 'Serializable' - like the datum they contain - but remember that deserializing
374-- a 'Message' will yield a 'Message', not the type within it! To obtain the
375-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
376--
377-- @
378-- do
379-- self <- getSelfPid
380-- send self (wrapMessage "blah")
381-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
382-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
383-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
384-- @
385--
386wrapMessage :: Serializable a => a -> Message
387wrapMessage = createMessage
389-- | Attempt to unwrap a raw 'Message'.
390-- If the type of the decoded message payload matches the expected type, the
391-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
392-- do not match.
393--
394-- This expression, for example, will evaluate to @Nothing@
395-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
396--
397-- Whereas this expression, will yield @Just "foo"@
398-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
399--
400unwrapMessage :: forall a. Serializable a => Message -> Process (Maybe a)
401unwrapMessage msg =
402 case messageFingerprint msg == fingerprint (undefined :: a) of
403 False -> return Nothing
404 True -> case msg of
405 (UnencodedMessage _ m) ->
406 let m' = unsafeCoerce m :: a
407 in return (Just m')
408 (EncodedMessage _ _) ->
409 return (Just (decoded))
410 where
411 decoded :: a -- note [decoding]
412 !decoded = decode (messageEncoding msg)
414-- | Attempt to handle a raw 'Message'.
415-- If the type of the message matches the type of the first argument to
416-- the supplied expression, then the message will be decoded and the expression
417-- evaluated against its value. If this runtime type checking fails however,
418-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
419-- evaluation proceeds, the resulting value with be wrapped with @Just@.
420--
421-- Intended for use in `catchesExit` and `matchAny` primitives.
422--
423handleMessage :: forall a b. (Serializable a)
424 => Message -> (a -> Process b) -> Process (Maybe b)
425handleMessage msg proc = do
426 case messageFingerprint msg == fingerprint (undefined :: a) of
427 False -> return Nothing
428 True -> case msg of
429 (UnencodedMessage _ m) ->
430 let m' = unsafeCoerce m :: a -- note [decoding]
431 in do { r <- proc m'; return (Just r) }
432 (EncodedMessage _ _) ->
433 do { r <- proc (decoded :: a); return (Just r) }
434 where
435 decoded :: a -- note [decoding]
436 !decoded = decode (messageEncoding msg)
438-- | Match against an arbitrary message. 'matchAny' removes the first available
439-- message from the process mailbox. To handle arbitrary /raw/ messages once
440-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
441--
442matchAny :: forall b. (Message -> Process b) -> Match b
443matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
445-- | Match against an arbitrary message. Intended for use with 'handleMessage'
446-- and 'unwrapMessage', this function /only/ removes a message from the process
447-- mailbox, /if/ the supplied condition matches. The success (or failure) of
448-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
449-- here, i.e., if the condition evaluates to @True@ then the message will
450-- be removed from the process mailbox and decoded, but that does /not/
451-- guarantee that an expression passed to @handleMessage@ will pass the
452-- runtime type checks and therefore be evaluated.
453--
454matchAnyIf :: forall a b. (Serializable a)
455 => (a -> Bool)
456 -> (Message -> Process b)
457 -> Match b
458matchAnyIf c p = Match $ MatchMsg $ \msg ->
459 case messageFingerprint msg == fingerprint (undefined :: a) of
460 True | check -> Just (p msg)
461 where
462 check :: Bool
463 !check =
464 case msg of
465 (EncodedMessage _ _) -> c decoded
466 (UnencodedMessage _ m') -> c (unsafeCoerce m')
468 decoded :: a -- note [decoding]
469 !decoded = decode (messageEncoding msg)
470 _ -> Nothing
472{- note [decoding]
473For an EncodedMessage, we need to ensure the value is fully decoded so that
474we don't hang to bytestrings if the calling process doesn't evaluate
475immediately. For UnencodedMessage we know (because the fingerprint comparison
476succeeds) that unsafeCoerce will not fail.
477-}
479-- | Remove any message from the queue
480matchUnknown :: Process b -> Match b
481matchUnknown p = Match $ MatchMsg (const (Just p))
483--------------------------------------------------------------------------------
484-- Process management --
485--------------------------------------------------------------------------------
487-- | Thrown by 'terminate'
488data ProcessTerminationException = ProcessTerminationException
489 deriving (Show, Typeable)
491instance Exception ProcessTerminationException
493-- | Terminate immediately (throws a ProcessTerminationException)
494terminate :: Process a
495terminate = liftIO $ throwIO ProcessTerminationException
497-- [Issue #110]
498-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
499die :: Serializable a => a -> Process b
500die reason = do
501 pid <- getSelfPid
502 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
504-- | Forceful request to kill a process. Where 'exit' provides an exception
505-- that can be caught and handled, 'kill' throws an unexposed exception type
506-- which cannot be handled explicitly (by type).
507kill :: ProcessId -> String -> Process ()
508-- NOTE: We send the message to our local node controller, which will then
509-- forward it to a remote node controller (if applicable). Sending it directly
510-- to a remote node controller means that that the message may overtake a
511-- 'monitor' or 'link' request.
512kill them reason = sendCtrlMsg Nothing (Kill them reason)
514-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
515-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
516-- manner can be handled using the 'catchExit' family of functions.
517exit :: Serializable a => ProcessId -> a -> Process ()
518-- NOTE: We send the message to our local node controller, which will then
519-- forward it to a remote node controller (if applicable). Sending it directly
520-- to a remote node controller means that that the message may overtake a
521-- 'monitor' or 'link' request.
522exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
524-- | Catches 'ProcessExitException'. The handler will not be applied unless its
525-- type matches the encoded data stored in the exception (see the /reason/
526-- argument given to the 'exit' primitive). If the handler cannot be applied,
527-- the exception will be re-thrown.
528--
529-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
530-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
531catchExit :: forall a b . (Show a, Serializable a)
532 => Process b
533 -> (ProcessId -> a -> Process b)
534 -> Process b
535catchExit act exitHandler = catch act handleExit
536 where
537 handleExit ex@(ProcessExitException from msg) =
538 if messageFingerprint msg == fingerprint (undefined :: a)
539 then exitHandler from decoded
540 else liftIO $ throwIO ex
541 where
542 decoded :: a
543 -- Make sure the value is fully decoded so that we don't hang to
544 -- bytestrings if the caller doesn't use the value immediately
545 !decoded = decode (messageEncoding msg)
547-- | Lift 'Control.Exception.catches' (almost).
548--
549-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
550-- message, a handler must accept an input of the expected type. In order to
551-- handle a list of potentially different handlers (and therefore input types),
552-- a handler passed to 'catchesExit' must accept 'AbstractMessage' and return
553-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
554--
555-- See 'maybeHandleMessage' and 'AsbtractMessage' for more details.
556catchesExit :: Process b
557 -> [(ProcessId -> Message -> (Process (Maybe b)))]
558 -> Process b
559catchesExit act handlers = catch act ((flip handleExit) handlers)
560 where
561 handleExit :: ProcessExitException
562 -> [(ProcessId -> Message -> Process (Maybe b))]
563 -> Process b
564 handleExit ex [] = liftIO $ throwIO ex
565 handleExit ex@(ProcessExitException from msg) (h:hs) = do
566 r <- h from msg
567 case r of
568 Nothing -> handleExit ex hs
569 Just p -> return p
571-- | Our own process ID
572getSelfPid :: Process ProcessId
573getSelfPid = processId <$> ask
575-- | Get the node ID of our local node
576getSelfNode :: Process NodeId
577getSelfNode = localNodeId . processNode <$> ask
579-- | Get information about the specified process
580getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
581getProcessInfo pid =
582 let them = processNodeId pid in do
583 us <- getSelfNode
584 dest <- mkNode them us
585 sendCtrlMsg dest $ GetInfo pid
586 receiveWait [
587 match (\(p :: ProcessInfo) -> return $ Just p)
588 , match (\(_ :: ProcessInfoNone) -> return Nothing)
589 ]
590 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
591 mkNode them us = case them == us of
592 True -> return Nothing
593 _ -> return $ Just them
595--------------------------------------------------------------------------------
596-- Monitoring and linking --
597--------------------------------------------------------------------------------
599-- | Link to a remote process (asynchronous)
600--
601-- When process A links to process B (that is, process A calls
602-- @link pidB@) then an asynchronous exception will be thrown to process A
603-- when process B terminates (normally or abnormally), or when process A gets
604-- disconnected from process B. Although it is /technically/ possible to catch
605-- these exceptions, chances are if you find yourself trying to do so you should
606-- probably be using 'monitor' rather than 'link'. In particular, code such as
607--
608-- > link pidB -- Link to process B
609-- > expect -- Wait for a message from process B
610-- > unlink pidB -- Unlink again
611--
612-- doesn't quite do what one might expect: if process B sends a message to
613-- process A, and /subsequently terminates/, then process A might or might not
614-- be terminated too, depending on whether the exception is thrown before or
615-- after the 'unlink' (i.e., this code has a race condition).
616--
617-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
618-- call to 'link' has no effect.
619--
620-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
621-- Linking makes no distinction between normal and abnormal termination of
622-- the remote process.
623link :: ProcessId -> Process ()
624link = sendCtrlMsg Nothing . Link . ProcessIdentifier
626-- | Monitor another process (asynchronous)
627--
628-- When process A monitors process B (that is, process A calls
629-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
630-- when process B terminates (normally or abnormally), or when process A gets
631-- disconnected from process B. You receive this message like any other (using
632-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
633-- 'DiedDisconnect', etc.).
634--
635-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
636-- multiple monitors are set up, multiple notifications will be delivered
637-- and monitors can be disabled individually using 'unmonitor'.
638monitor :: ProcessId -> Process MonitorRef
639monitor = monitor' . ProcessIdentifier
641-- | Establishes temporary monitoring of another process.
642--
643-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
644-- of @code@. Note: although monitoring is no longer active when
645-- @withMonitor@ returns, there might still be unreceived monitor
646-- messages in the queue.
647--
648withMonitor :: ProcessId -> Process a -> Process a
649withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
650 -- unmonitor blocks waiting for the response, so there's a possibility
651 -- that an exception might interrupt withMonitor before the unmonitor
652 -- has completed. I think that's better than making the unmonitor
653 -- uninterruptible.
655-- | Remove a link
656--
657-- This is synchronous in the sense that once it returns you are guaranteed
658-- that no exception will be raised if the remote process dies. However, it is
659-- asynchronous in the sense that we do not wait for a response from the remote
660-- node.
661unlink :: ProcessId -> Process ()
662unlink pid = do
663 unlinkAsync pid
664 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
665 (\_ -> return ())
666 ]
668-- | Remove a node link
669--
670-- This has the same synchronous/asynchronous nature as 'unlink'.
671unlinkNode :: NodeId -> Process ()
672unlinkNode nid = do
673 unlinkNodeAsync nid
674 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
675 (\_ -> return ())
676 ]
678-- | Remove a channel (send port) link
679--
680-- This has the same synchronous/asynchronous nature as 'unlink'.
681unlinkPort :: SendPort a -> Process ()
682unlinkPort sport = do
683 unlinkPortAsync sport
684 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
685 (\_ -> return ())
686 ]
688-- | Remove a monitor
689--
690-- This has the same synchronous/asynchronous nature as 'unlink'.
691unmonitor :: MonitorRef -> Process ()
692unmonitor ref = do
693 unmonitorAsync ref
694 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
695 (\_ -> return ())
696 ]
698--------------------------------------------------------------------------------
699-- Exception handling --
700--------------------------------------------------------------------------------
702-- | Lift 'Control.Exception.catch'
703catch :: Exception e => Process a -> (e -> Process a) -> Process a
704catch p h = do
705 lproc <- ask
706 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
708-- | Lift 'Control.Exception.try'
709try :: Exception e => Process a -> Process (Either e a)
710try p = do
711 lproc <- ask
712 liftIO $ Ex.try (runLocalProcess lproc p)
714-- | Lift 'Control.Exception.mask'
715mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
716mask p = do
717 lproc <- ask
718 liftIO $ Ex.mask $ \restore ->
719 runLocalProcess lproc (p (liftRestore lproc restore))
720 where
721 liftRestore :: LocalProcess -> (forall a. IO a -> IO a) -> (forall a. Process a -> Process a)
722 liftRestore lproc restoreIO = liftIO . restoreIO . runLocalProcess lproc
724-- | Lift 'Control.Exception.onException'
725onException :: Process a -> Process b -> Process a
726onException p what = p `catch` \e -> do _ <- what
727 liftIO $ throwIO (e :: SomeException)
729-- | Lift 'Control.Exception.bracket'
730bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
731bracket before after thing =
732 mask $ \restore -> do
733 a <- before
734 r <- restore (thing a) `onException` after a
735 _ <- after a
736 return r
738-- | Lift 'Control.Exception.bracket_'
739bracket_ :: Process a -> Process b -> Process c -> Process c
740bracket_ before after thing = bracket before (const after) (const thing)
742-- | Lift 'Control.Exception.finally'
743finally :: Process a -> Process b -> Process a
744finally a sequel = bracket_ (return ()) sequel a
746-- | You need this when using 'catches'
747data Handler a = forall e . Exception e => Handler (e -> Process a)
749instance Functor Handler where
750 fmap f (Handler h) = Handler (fmap f . h)
752-- | Lift 'Control.Exception.catches'
753catches :: Process a -> [Handler a] -> Process a
754catches proc handlers = proc `catch` catchesHandler handlers
756catchesHandler :: [Handler a] -> SomeException -> Process a
757catchesHandler handlers e = foldr tryHandler (throw e) handlers
758 where tryHandler (Handler handler) res
759 = case fromException e of
760 Just e' -> handler e'
761 Nothing -> res
763--------------------------------------------------------------------------------
764-- Auxiliary API --
765--------------------------------------------------------------------------------
767-- | Like 'expect' but with a timeout
768expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
769expectTimeout n = receiveTimeout n [match return]
771-- | Asynchronous version of 'spawn'
772--
773-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
774spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
775spawnAsync nid proc = do
776 spawnRef <- getSpawnRef
777 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
778 return spawnRef
780-- | Monitor a node (asynchronous)
781monitorNode :: NodeId -> Process MonitorRef
782monitorNode =
783 monitor' . NodeIdentifier
785-- | Monitor a typed channel (asynchronous)
786monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
787monitorPort (SendPort cid) =
788 monitor' (SendPortIdentifier cid)
790-- | Remove a monitor (asynchronous)
791unmonitorAsync :: MonitorRef -> Process ()
792unmonitorAsync =
793 sendCtrlMsg Nothing . Unmonitor
795-- | Link to a node (asynchronous)
796linkNode :: NodeId -> Process ()
797linkNode = link' . NodeIdentifier
799-- | Link to a channel (asynchronous)
800linkPort :: SendPort a -> Process ()
801linkPort (SendPort cid) =
802 link' (SendPortIdentifier cid)
804-- | Remove a process link (asynchronous)
805unlinkAsync :: ProcessId -> Process ()
806unlinkAsync =
807 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
809-- | Remove a node link (asynchronous)
810unlinkNodeAsync :: NodeId -> Process ()
811unlinkNodeAsync =
812 sendCtrlMsg Nothing . Unlink . NodeIdentifier
814-- | Remove a channel (send port) link (asynchronous)
815unlinkPortAsync :: SendPort a -> Process ()
816unlinkPortAsync (SendPort cid) =
817 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
819--------------------------------------------------------------------------------
820-- Logging --
821--------------------------------------------------------------------------------
823-- | Log a string
824--
825-- @say message@ sends a message (time, pid of the current process, message)
826-- to the process registered as 'logger'. By default, this process simply
827-- sends the string to 'stderr'. Individual Cloud Haskell backends might
828-- replace this with a different logger process, however.
829say :: String -> Process ()
830say string = do
831 now <- liftIO getCurrentTime
832 us <- getSelfPid
833 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
835--------------------------------------------------------------------------------
836-- Registry --
837--------------------------------------------------------------------------------
839-- | Register a process with the local registry (asynchronous).
840-- This version will wait until a response is gotten from the
841-- management process. The name must not already be registered.
842-- The process need not be on this node.
843-- A bad registration will result in a 'ProcessRegistrationException'
844--
845-- The process to be registered does not have to be local itself.
846register :: String -> ProcessId -> Process ()
847register = registerImpl False
849-- | Like 'register', but will replace an existing registration.
850-- The name must already be registered.
851reregister :: String -> ProcessId -> Process ()
852reregister = registerImpl True
854registerImpl :: Bool -> String -> ProcessId -> Process ()
855registerImpl force label pid = do
856 mynid <- getSelfNode
857 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
858 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
859 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
860 ]
862-- | Register a process with a remote registry (asynchronous).
863--
864-- The process to be registered does not have to live on the same remote node.
865-- Reply wil come in the form of a 'RegisterReply' message
866--
867-- See comments in 'whereisRemoteAsync'
868registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
869registerRemoteAsync nid label pid =
870 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
872reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
873reregisterRemoteAsync nid label pid =
874 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
876-- | Remove a process from the local registry (asynchronous).
877-- This version will wait until a response is gotten from the
878-- management process. The name must already be registered.
879unregister :: String -> Process ()
880unregister label = do
881 mynid <- getSelfNode
882 sendCtrlMsg Nothing (Register label mynid Nothing False)
883 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
884 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
885 ]
887-- | Deal with the result from an attempted registration or unregistration
888-- by throwing an exception if necessary
889handleRegistrationReply :: String -> Bool -> Process ()
890handleRegistrationReply label ok =
891 when (not ok) $
892 liftIO $ throwIO $ ProcessRegistrationException label
894-- | Remove a process from a remote registry (asynchronous).
895--
896-- Reply wil come in the form of a 'RegisterReply' message
897--
898-- See comments in 'whereisRemoteAsync'
899unregisterRemoteAsync :: NodeId -> String -> Process ()
900unregisterRemoteAsync nid label =
901 sendCtrlMsg (Just nid) (Register label nid Nothing False)
903-- | Query the local process registry
904whereis :: String -> Process (Maybe ProcessId)
905whereis label = do
906 sendCtrlMsg Nothing (WhereIs label)
907 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
908 (\(WhereIsReply _ mPid) -> return mPid)
909 ]
911-- | Query a remote process registry (asynchronous)
912--
913-- Reply will come in the form of a 'WhereIsReply' message.
914--
915-- There is currently no synchronous version of 'whereisRemoteAsync': if
916-- you implement one yourself, be sure to take into account that the remote
917-- node might die or get disconnect before it can respond (i.e. you should
918-- use 'monitorNode' and take appropriate action when you receive a
919-- 'NodeMonitorNotification').
920whereisRemoteAsync :: NodeId -> String -> Process ()
921whereisRemoteAsync nid label =
922 sendCtrlMsg (Just nid) (WhereIs label)
924-- | Named send to a process in the local registry (asynchronous)
925nsend :: Serializable a => String -> a -> Process ()
926nsend label msg =
927 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
929-- | Named send to a process in a remote registry (asynchronous)
930nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
931nsendRemote nid label msg =
932 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
934--------------------------------------------------------------------------------
935-- Closures --
936--------------------------------------------------------------------------------
938-- | Resolve a static value
939unStatic :: Typeable a => Static a -> Process a
940unStatic static = do
941 rtable <- remoteTable . processNode <$> ask
942 case Static.unstatic rtable static of
943 Left err -> fail $ "Could not resolve static value: " ++ err
944 Right x -> return x
946-- | Resolve a closure
947unClosure :: Typeable a => Closure a -> Process a
948unClosure closure = do
949 rtable <- remoteTable . processNode <$> ask
950 case Static.unclosure rtable closure of
951 Left err -> fail $ "Could not resolve closure: " ++ err
952 Right x -> return x
954--------------------------------------------------------------------------------
955-- Reconnecting --
956--------------------------------------------------------------------------------
958-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
959-- message passing. However, when network connections get disrupted this
960-- illusion cannot always be maintained. Once a network connection breaks (even
961-- temporarily) no further communication on that connection will be possible.
962-- For example, if process A sends a message to process B, and A is then
963-- notified (by monitor notification) that it got disconnected from B, A will
964-- not be able to send any further messages to B, /unless/ A explicitly
965-- indicates that it is acceptable to attempt to reconnect to B using the
966-- Cloud Haskell 'reconnect' primitive.
967--
968-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
969-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
970-- then receives a monitor notification that its connection to B has been lost,
971-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
972-- and m3 but not m2.
973--
974-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
975-- to attempt to reconnect on the next send/. In particular, if no further
976-- communication attempts are made to B then A can use reconnect to clean up
977-- its connection to B.
978reconnect :: ProcessId -> Process ()
979reconnect them = do
980 us <- getSelfPid
981 node <- processNode <$> ask
982 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
984-- | Reconnect to a sendport. See 'reconnect' for more information.
985reconnectPort :: SendPort a -> Process ()
986reconnectPort them = do
987 us <- getSelfPid
988 node <- processNode <$> ask
989 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
991--------------------------------------------------------------------------------
992-- Debugging/Tracing --
993--------------------------------------------------------------------------------
995-- | Send a message to the internal (system) trace facility. If tracing is
996-- enabled, this will create a custom trace event. Note that several Cloud Haskell
997-- sub-systems also generate trace events for informational/debugging purposes,
998-- thus traces generated this way will not be the only output seen.
999--
1000-- Just as with the "Debug.Trace" module, this is a debugging/tracing facility
1001-- for use in development, and should not be used in a production setting -
1002-- which is why the default behaviour is to trace to the GHC eventlog. For a
1003-- general purpose logging facility, you should consider 'say'.
1005-- Trace events can be written to the GHC event log, a text file, or to the
1006-- standard system logger process (see 'say'). The default behaviour for writing
1007-- to the eventlog requires specific intervention to work, without which traces
1008-- are silently dropped/ignored and no output will be generated.
1009-- The GHC eventlog documentation provides information about enabling, viewing
1010-- and working with event traces: <http://hackage.haskell.org/trac/ghc/wiki/EventLog>.
1012-- When a new local node is started, the contents of the environment variable
1013-- @DISTRIBUTED_PROCESS_TRACE_FILE@ are checked for a valid file path. If this
1014-- exists and the file can be opened for writing, all trace output will be directed
1015-- thence. If the environment variable is empty, the path invalid, or the file
1016-- unavailable for writing - e.g., because another node has already started
1017-- tracing to it - then the @DISTRIBUTED_PROCESS_TRACE_CONSOLE@ environment
1018-- variable is checked for /any/ non-empty value. If this is set, then all trace
1019-- output will be directed to the system logger process. If neither evironment
1020-- variable provides a valid trace configuration, all internal traces are written
1021-- to "Debug.Trace.traceEventIO", which writes to the GHC eventlog.
1023-- Users of the /simplelocalnet/ Cloud Haskell backend should also note that
1024-- because the trace file option only supports trace output from a single node
1025-- (so as to avoid interleaving), a file trace configured for the master node will
1026-- prevent slaves from tracing to the file and they will fall back to using the
1027-- console/'say' or eventlog instead.
1029trace :: String -> Process ()
1030trace s = do
1031 node <- processNode <$> ask
1032 liftIO $ Trace.trace (localTracer node) s
1034--------------------------------------------------------------------------------
1035-- Auxiliary functions --
1036--------------------------------------------------------------------------------
1038sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1039sendLocal pid msg =
1040 sendCtrlMsg Nothing $ LocalSend pid (createUnencodedMessage msg)
1042sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1043sendChanLocal spId msg =
1044 -- we *must* fully serialize/encode the message here, because
1045 -- attempting to use `unsafeCoerce' in the node controller
1046 -- won't work since we know nothing about the required type
1047 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1049getMonitorRefFor :: Identifier -> Process MonitorRef
1050getMonitorRefFor ident = do
1051 proc <- ask
1052 liftIO $ modifyMVar (processState proc) $ \st -> do
1053 let counter = st ^. monitorCounter
1054 return ( monitorCounter ^: (+ 1) $ st
1055 , MonitorRef ident counter
1058getSpawnRef :: Process SpawnRef
1059getSpawnRef = do
1060 proc <- ask
1061 liftIO $ modifyMVar (processState proc) $ \st -> do
1062 let counter = st ^. spawnCounter
1063 return ( spawnCounter ^: (+ 1) $ st
1064 , SpawnRef counter
1067-- | Monitor a process/node/channel
1068monitor' :: Identifier -> Process MonitorRef
1069monitor' ident = do
1070 monitorRef <- getMonitorRefFor ident
1071 sendCtrlMsg Nothing $ Monitor monitorRef
1072 return monitorRef
1074-- | Link to a process/node/channel
1075link' :: Identifier -> Process ()
1076link' = sendCtrlMsg Nothing . Link
1078-- Send a control message
1079sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node
1080 -> ProcessSignal -- ^ Message to send
1081 -> Process ()
1082sendCtrlMsg mNid signal = do
1083 proc <- ask
1084 let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc)
1085 , ctrlMsgSignal = signal
1087 case mNid of
1088 Nothing -> do
1089 liftIO $ writeChan (localCtrlChan (processNode proc)) msg
1090 Just nid ->
1091 liftIO $ sendBinary (processNode proc)
1092 (ProcessIdentifier (processId proc))
1093 (NodeIdentifier nid)
1094 WithImplicitReconnect
1095 msg