1-- | Cloud Haskell primitives
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
15 -- * Advanced messaging
35 -- * Process management
37 , ProcessTerminationException(..)
43 -- keep the exception constructor hidden, so that handling exit
44 -- reasons /must/ take place via the 'catchExit' family of primitives
45 , ProcessExitException()
50 -- * Monitoring and linking
65 , reregisterRemoteAsync
66 , unregisterRemoteAsync
72 -- * Exception handling
97#if ! MIN_VERSION_base(4,6,0)
98import Prelude hiding (catch)
101import Data.Binary (decode)
102import Data.Time.Clock (getCurrentTime)
103import Data.Time.Format (formatTime)
104import System.Locale (defaultTimeLocale)
105import System.Timeout (timeout)
106import Control.Monad (when)
107import Control.Monad.Reader (ask)
108import Control.Monad.IO.Class (MonadIO, liftIO)
109import Control.Applicative ((<$>))
110import Control.Exception (Exception(..), throw, throwIO, SomeException)
111import qualified Control.Exception as Ex (catch, mask, try)
112import Control.Distributed.Process.Internal.StrictMVar
117import Control.Concurrent.Chan (writeChan)
118import Control.Concurrent.STM
127import Control.Distributed.Process.Internal.CQueue
132import Control.Distributed.Process.Serializable (Serializable, fingerprint)
133import Data.Accessor ((^.), (^:), (^=))
134import Control.Distributed.Static (Closure, Static)
135import Data.Rank1Typeable (Typeable)
136import qualified Control.Distributed.Static as Static (unstatic, unclosure)
137import Control.Distributed.Process.Internal.Types
157 , ProcessExitException(..)
159 , DidUnlinkProcess(..)
164 , ProcessRegistrationException(..)
166 , ProcessInfoNone(..)
168 , createUnencodedMessage
170 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
175import Control.Distributed.Process.Internal.Messaging
181import Control.Distributed.Process.Internal.Trace.Primitives
185import Control.Distributed.Process.Internal.WeakTQueue
193--------------------------------------------------------------------------------
195--------------------------------------------------------------------------------
198send :: Serializable a => ProcessId -> a -> Process ()
199-- This requires a lookup on every send. If we want to avoid that we need to
200-- modify serializable to allow for stateful (IO) deserialization.
203 let us = processId proc
204 node = processNode proc
205 nodeId = localNodeId node
206 destNode = (processNodeId them) in do
207 case destNode == nodeId of
208 True -> sendLocal them msg
209 False -> liftIO $ sendMessage (processNode proc)
210 (ProcessIdentifier (processId proc))
211 (ProcessIdentifier them)
214 -- We do not fire the trace event until after the sending is complete;
215 -- In the remote case, 'sendMessage' can block in the networking stack.
216 liftIO $ traceEvent (localTracer node)
217 (TraceEvSent them us (createUnencodedMessage msg))
219-- | Wait for a message of a specific type
220expect :: forall a. Serializable a => Process a
221expect = receiveWait [match return]
223--------------------------------------------------------------------------------
225--------------------------------------------------------------------------------
227-- | Create a new typed channel
228newChan :: Serializable a => Process (SendPort a, ReceivePort a)
231 liftIO . modifyMVar (processState proc) $ \st -> do
232 let lcid = st ^. channelCounter
233 let cid = SendPortId { sendPortProcessId = processId proc
234 , sendPortLocalId = lcid
236 let sport = SendPort cid
237 chan <- liftIO newTQueueIO
238 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
239 let rport = ReceivePort $ readTQueue chan
240 let tch = TypedChannel chan'
241 return ( (channelCounter ^: (+ 1))
242 . (typedChannelWithId lcid ^= Just tch)
247 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
248 finalizer st lcid = modifyMVar_ st $
249 return . (typedChannelWithId lcid ^= Nothing)
251-- | Send a message on a typed channel
252sendChan :: Serializable a => SendPort a -> a -> Process ()
253sendChan (SendPort cid) msg = do
255 let node = localNodeId (processNode proc)
256 destNode = processNodeId (sendPortProcessId cid) in do
257 case destNode == node of
258 True -> sendChanLocal cid msg
260 liftIO $ sendBinary (processNode proc)
261 (ProcessIdentifier (processId proc))
262 (SendPortIdentifier cid)
266-- | Wait for a message on a typed channel
267receiveChan :: Serializable a => ReceivePort a -> Process a
268receiveChan = liftIO . atomically . receiveSTM
270-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
271-- non-blocking check for a message.
272receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
273receiveChanTimeout 0 ch = liftIO . atomically $
274 (Just <$> receiveSTM ch) `orElse` return Nothing
275receiveChanTimeout n ch = liftIO . timeout n . atomically $
278-- | Merge a list of typed channels.
280-- The result port is left-biased: if there are messages available on more
281-- than one port, the first available message is returned.
282mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
283mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
285-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
287mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
288mergePortsRR = \ps -> do
289 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
290 return $ ReceivePort (rr psVar)
294 rotate (x:xs) = xs ++ [x]
296 rr :: TVar [STM a] -> STM a
299 a <- foldr1 orElse ps
300 writeTVar psVar (rotate ps)
303--------------------------------------------------------------------------------
304-- Advanced messaging --
305--------------------------------------------------------------------------------
307-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
308newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
310-- | Test the matches in order against each message in the queue
311receiveWait :: [Match b] -> Process b
313 queue <- processQueue <$> ask
314 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
317-- | Like 'receiveWait' but with a timeout.
319-- If the timeout is zero do a non-blocking check for matching messages. A
320-- non-zero timeout is applied only when waiting for incoming messages (that is,
321-- /after/ we have checked the messages that are already in the mailbox).
322receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
323receiveTimeout t ms = do
324 queue <- processQueue <$> ask
325 let blockSpec = if t == 0 then NonBlocking else Timeout t
326 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
328 Nothing -> return Nothing
329 Just proc -> Just <$> proc
331matchChan :: ReceivePort a -> (a -> Process b) -> Match b
332matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
334-- | Match against any message of the right type
335match :: forall a b. Serializable a => (a -> Process b) -> Match b
336match = matchIf (const True)
338-- | Match against any message of the right type that satisfies a predicate
339matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
340matchIf c p = Match $ MatchMsg $ \msg ->
341 case messageFingerprint msg == fingerprint (undefined :: a) of
344 (UnencodedMessage _ m) ->
345 let m' = unsafeCoerce m :: a in
349 (EncodedMessage _ _) ->
350 if (c decoded) then Just (p decoded) else Nothing
353 -- Make sure the value is fully decoded so that we don't hang to
354 -- bytestrings when the process calling 'matchIf' doesn't process
355 -- the values immediately
356 !decoded = decode (messageEncoding msg)
358-- | Match against any message, regardless of the underlying (contained) type
359matchMessage :: (Message -> Process Message) -> Match Message
360matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
362-- | Match against any message (regardless of underlying type) that satisfies a predicate
363matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
364matchMessageIf c p = Match $ MatchMsg $ \msg ->
369-- | Forward a raw 'Message' to the given 'ProcessId'.
370forward :: Message -> ProcessId -> Process ()
373 let node = processNode proc
375 nid = localNodeId node
376 destNode = (processNodeId them) in do
377 case destNode == nid of
378 True -> sendCtrlMsg Nothing (LocalSend them msg)
379 False -> liftIO $ sendPayload (processNode proc)
380 (ProcessIdentifier (processId proc))
381 (ProcessIdentifier them)
383 (messageToPayload msg)
384 -- We do not fire the trace event until after the sending is complete;
385 -- In the remote case, 'sendMessage' can block in the networking stack.
386 liftIO $ traceEvent (localTracer node)
387 (TraceEvSent them us (createUnencodedMessage msg))
390-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
391-- 'Serializable' - like the datum they contain - but remember that deserializing
392-- a 'Message' will yield a 'Message', not the type within it! To obtain the
393-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
398-- send self (wrapMessage "blah")
399-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
400-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
401-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
404wrapMessage :: Serializable a => a -> Message
405wrapMessage = createMessage
407-- | Like 'wrapMessage', but produces a 'Message' that is only usable in the
408-- local node. This is primarily useful when consuming the tracing and debugging
409-- APIs in "Control.Distributed.Process.Debug".
410wrapMessageLocal :: Serializable a => a -> Message
411wrapMessageLocal = createUnencodedMessage
413-- | Attempt to unwrap a raw 'Message'.
414-- If the type of the decoded message payload matches the expected type, the
415-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
418-- This expression, for example, will evaluate to @Nothing@
419-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
421-- Whereas this expression, will yield @Just "foo"@
422-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
424unwrapMessage :: forall a. Serializable a => Message -> Process (Maybe a)
426 case messageFingerprint msg == fingerprint (undefined :: a) of
427 False -> return Nothing
429 (UnencodedMessage _ m) ->
430 let m' = unsafeCoerce m :: a
432 (EncodedMessage _ _) ->
433 return (Just (decoded))
435 decoded :: a -- note [decoding]
436 !decoded = decode (messageEncoding msg)
438-- | Attempt to handle a raw 'Message'.
439-- If the type of the message matches the type of the first argument to
440-- the supplied expression, then the message will be decoded and the expression
441-- evaluated against its value. If this runtime type checking fails however,
442-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
443-- evaluation proceeds, the resulting value with be wrapped with @Just@.
445-- Intended for use in `catchesExit` and `matchAny` primitives.
447handleMessage :: forall a b. (Serializable a)
448 => Message -> (a -> Process b) -> Process (Maybe b)
449handleMessage msg proc = do
450 case messageFingerprint msg == fingerprint (undefined :: a) of
451 False -> return Nothing
453 (UnencodedMessage _ m) ->
454 let m' = unsafeCoerce m :: a -- note [decoding]
455 in do { r <- proc m'; return (Just r) }
456 (EncodedMessage _ _) ->
457 do { r <- proc (decoded :: a); return (Just r) }
459 decoded :: a -- note [decoding]
460 !decoded = decode (messageEncoding msg)
462-- | Match against an arbitrary message. 'matchAny' removes the first available
463-- message from the process mailbox. To handle arbitrary /raw/ messages once
464-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
466matchAny :: forall b. (Message -> Process b) -> Match b
467matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
469-- | Match against an arbitrary message. Intended for use with 'handleMessage'
470-- and 'unwrapMessage', this function /only/ removes a message from the process
471-- mailbox, /if/ the supplied condition matches. The success (or failure) of
472-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
473-- here, i.e., if the condition evaluates to @True@ then the message will
474-- be removed from the process mailbox and decoded, but that does /not/
475-- guarantee that an expression passed to @handleMessage@ will pass the
476-- runtime type checks and therefore be evaluated.
478matchAnyIf :: forall a b. (Serializable a)
480 -> (Message -> Process b)
482matchAnyIf c p = Match $ MatchMsg $ \msg ->
483 case messageFingerprint msg == fingerprint (undefined :: a) of
484 True | check -> Just (p msg)
489 (EncodedMessage _ _) -> c decoded
490 (UnencodedMessage _ m') -> c (unsafeCoerce m')
492 decoded :: a -- note [decoding]
493 !decoded = decode (messageEncoding msg)
497For an EncodedMessage, we need to ensure the value is fully decoded so that
498we don't hang to bytestrings if the calling process doesn't evaluate
499immediately. For UnencodedMessage we know (because the fingerprint comparison
500succeeds) that unsafeCoerce will not fail.
503-- | Remove any message from the queue
504matchUnknown :: Process b -> Match b
505matchUnknown p = Match $ MatchMsg (const (Just p))
507-- | Receives messages and forwards them to @pid@ if @p msg == True@.
508delegate :: ProcessId -> (Message -> Bool) -> Process ()
511 matchAny (\m -> case (p m) of
512 True -> forward m pid
517-- | A straight relay that simply forwards all messages to the supplied pid.
518relay :: ProcessId -> Process ()
519relay pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
521-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
522-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
523-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
524-- fails, no action is taken and the proxy process will continue running.
525proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
529 say "handling proxied message..."
530 next <- handleMessage m proc
532 Just True -> say ("forwarding msg to " ++ (show pid)) >> forward m pid
533 Just False -> say "ignoring explicitly ignored msg"
534 Nothing -> say "ignoring unroutable msg")
538--------------------------------------------------------------------------------
539-- Process management --
540--------------------------------------------------------------------------------
542-- | Thrown by 'terminate'
543data ProcessTerminationException = ProcessTerminationException
544 deriving (Show, Typeable)
546instance Exception ProcessTerminationException
548-- | Terminate immediately (throws a ProcessTerminationException)
549terminate :: Process a
550terminate = liftIO $ throwIO ProcessTerminationException
553-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
554die :: Serializable a => a -> Process b
557 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
559-- | Forceful request to kill a process. Where 'exit' provides an exception
560-- that can be caught and handled, 'kill' throws an unexposed exception type
561-- which cannot be handled explicitly (by type).
562kill :: ProcessId -> String -> Process ()
563-- NOTE: We send the message to our local node controller, which will then
564-- forward it to a remote node controller (if applicable). Sending it directly
565-- to a remote node controller means that that the message may overtake a
566-- 'monitor' or 'link' request.
567kill them reason = sendCtrlMsg Nothing (Kill them reason)
569-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
570-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
571-- manner can be handled using the 'catchExit' family of functions.
572exit :: Serializable a => ProcessId -> a -> Process ()
573-- NOTE: We send the message to our local node controller, which will then
574-- forward it to a remote node controller (if applicable). Sending it directly
575-- to a remote node controller means that that the message may overtake a
576-- 'monitor' or 'link' request.
577exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
579-- | Catches 'ProcessExitException'. The handler will not be applied unless its
580-- type matches the encoded data stored in the exception (see the /reason/
581-- argument given to the 'exit' primitive). If the handler cannot be applied,
582-- the exception will be re-thrown.
584-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
585-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
586catchExit :: forall a b . (Show a, Serializable a)
588 -> (ProcessId -> a -> Process b)
590catchExit act exitHandler = catch act handleExit
592 handleExit ex@(ProcessExitException from msg) =
593 if messageFingerprint msg == fingerprint (undefined :: a)
594 then exitHandler from decoded
595 else liftIO $ throwIO ex
598 -- Make sure the value is fully decoded so that we don't hang to
599 -- bytestrings if the caller doesn't use the value immediately
600 !decoded = decode (messageEncoding msg)
602-- | Lift 'Control.Exception.catches' (almost).
604-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
605-- message, a handler must accept an input of the expected type. In order to
606-- handle a list of potentially different handlers (and therefore input types),
607-- a handler passed to 'catchesExit' must accept 'AbstractMessage' and return
608-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
610-- See 'maybeHandleMessage' and 'AsbtractMessage' for more details.
611catchesExit :: Process b
612 -> [(ProcessId -> Message -> (Process (Maybe b)))]
614catchesExit act handlers = catch act ((flip handleExit) handlers)
616 handleExit :: ProcessExitException
617 -> [(ProcessId -> Message -> Process (Maybe b))]
619 handleExit ex [] = liftIO $ throwIO ex
620 handleExit ex@(ProcessExitException from msg) (h:hs) = do
623 Nothing -> handleExit ex hs
626-- | Our own process ID
627getSelfPid :: Process ProcessId
628getSelfPid = processId <$> ask
630-- | Get the node ID of our local node
631getSelfNode :: Process NodeId
632getSelfNode = localNodeId . processNode <$> ask
634-- | Get information about the specified process
635getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
637 let them = processNodeId pid in do
639 dest <- mkNode them us
640 sendCtrlMsg dest $ GetInfo pid
642 match (\(p :: ProcessInfo) -> return $ Just p)
643 , match (\(_ :: ProcessInfoNone) -> return Nothing)
645 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
646 mkNode them us = case them == us of
647 True -> return Nothing
648 _ -> return $ Just them
650--------------------------------------------------------------------------------
651-- Monitoring and linking --
652--------------------------------------------------------------------------------
654-- | Link to a remote process (asynchronous)
656-- When process A links to process B (that is, process A calls
657-- @link pidB@) then an asynchronous exception will be thrown to process A
658-- when process B terminates (normally or abnormally), or when process A gets
659-- disconnected from process B. Although it is /technically/ possible to catch
660-- these exceptions, chances are if you find yourself trying to do so you should
661-- probably be using 'monitor' rather than 'link'. In particular, code such as
663-- > link pidB -- Link to process B
664-- > expect -- Wait for a message from process B
665-- > unlink pidB -- Unlink again
667-- doesn't quite do what one might expect: if process B sends a message to
668-- process A, and /subsequently terminates/, then process A might or might not
669-- be terminated too, depending on whether the exception is thrown before or
670-- after the 'unlink' (i.e., this code has a race condition).
672-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
673-- call to 'link' has no effect.
675-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
676-- Linking makes no distinction between normal and abnormal termination of
677-- the remote process.
678link :: ProcessId -> Process ()
679link = sendCtrlMsg Nothing . Link . ProcessIdentifier
681-- | Monitor another process (asynchronous)
683-- When process A monitors process B (that is, process A calls
684-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
685-- when process B terminates (normally or abnormally), or when process A gets
686-- disconnected from process B. You receive this message like any other (using
687-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
688-- 'DiedDisconnect', etc.).
690-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
691-- multiple monitors are set up, multiple notifications will be delivered
692-- and monitors can be disabled individually using 'unmonitor'.
693monitor :: ProcessId -> Process MonitorRef
694monitor = monitor' . ProcessIdentifier
696-- | Establishes temporary monitoring of another process.
698-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
699-- of @code@. Note: although monitoring is no longer active when
700-- @withMonitor@ returns, there might still be unreceived monitor
701-- messages in the queue.
703withMonitor :: ProcessId -> Process a -> Process a
704withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
705 -- unmonitor blocks waiting for the response, so there's a possibility
706 -- that an exception might interrupt withMonitor before the unmonitor
707 -- has completed. I think that's better than making the unmonitor
712-- This is synchronous in the sense that once it returns you are guaranteed
713-- that no exception will be raised if the remote process dies. However, it is
714-- asynchronous in the sense that we do not wait for a response from the remote
716unlink :: ProcessId -> Process ()
719 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
723-- | Remove a node link
725-- This has the same synchronous/asynchronous nature as 'unlink'.
726unlinkNode :: NodeId -> Process ()
729 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
733-- | Remove a channel (send port) link
735-- This has the same synchronous/asynchronous nature as 'unlink'.
736unlinkPort :: SendPort a -> Process ()
738 unlinkPortAsync sport
739 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
745-- This has the same synchronous/asynchronous nature as 'unlink'.
746unmonitor :: MonitorRef -> Process ()
749 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
753--------------------------------------------------------------------------------
754-- Exception handling --
755--------------------------------------------------------------------------------
757-- | Lift 'Control.Exception.catch'
758catch :: Exception e => Process a -> (e -> Process a) -> Process a
761 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
763-- | Lift 'Control.Exception.try'
764try :: Exception e => Process a -> Process (Either e a)
767 liftIO $ Ex.try (runLocalProcess lproc p)
769-- | Lift 'Control.Exception.mask'
770mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
773 liftIO $ Ex.mask $ \restore ->
774 runLocalProcess lproc (p (liftRestore restore))
776 liftRestore :: (forall a. IO a -> IO a)
777 -> (forall a. Process a -> Process a)
778 liftRestore restoreIO = \p2 -> do
780 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
782-- | Lift 'Control.Exception.onException'
783onException :: Process a -> Process b -> Process a
784onException p what = p `catch` \e -> do _ <- what
785 liftIO $ throwIO (e :: SomeException)
787-- | Lift 'Control.Exception.bracket'
788bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
789bracket before after thing =
790 mask $ \restore -> do
792 r <- restore (thing a) `onException` after a
796-- | Lift 'Control.Exception.bracket_'
797bracket_ :: Process a -> Process b -> Process c -> Process c
798bracket_ before after thing = bracket before (const after) (const thing)
800-- | Lift 'Control.Exception.finally'
801finally :: Process a -> Process b -> Process a
802finally a sequel = bracket_ (return ()) sequel a
804-- | You need this when using 'catches'
805data Handler a = forall e . Exception e => Handler (e -> Process a)
807instance Functor Handler where
808 fmap f (Handler h) = Handler (fmap f . h)
810-- | Lift 'Control.Exception.catches'
811catches :: Process a -> [Handler a] -> Process a
812catches proc handlers = proc `catch` catchesHandler handlers
814catchesHandler :: [Handler a] -> SomeException -> Process a
815catchesHandler handlers e = foldr tryHandler (throw e) handlers
816 where tryHandler (Handler handler) res
817 = case fromException e of
818 Just e' -> handler e'
821--------------------------------------------------------------------------------
823--------------------------------------------------------------------------------
825-- | Like 'expect' but with a timeout
826expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
827expectTimeout n = receiveTimeout n [match return]
829-- | Asynchronous version of 'spawn'
831-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
832spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
833spawnAsync nid proc = do
834 spawnRef <- getSpawnRef
835 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
838-- | Monitor a node (asynchronous)
839monitorNode :: NodeId -> Process MonitorRef
841 monitor' . NodeIdentifier
843-- | Monitor a typed channel (asynchronous)
844monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
845monitorPort (SendPort cid) =
846 monitor' (SendPortIdentifier cid)
848-- | Remove a monitor (asynchronous)
849unmonitorAsync :: MonitorRef -> Process ()
851 sendCtrlMsg Nothing . Unmonitor
853-- | Link to a node (asynchronous)
854linkNode :: NodeId -> Process ()
855linkNode = link' . NodeIdentifier
857-- | Link to a channel (asynchronous)
858linkPort :: SendPort a -> Process ()
859linkPort (SendPort cid) =
860 link' (SendPortIdentifier cid)
862-- | Remove a process link (asynchronous)
863unlinkAsync :: ProcessId -> Process ()
865 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
867-- | Remove a node link (asynchronous)
868unlinkNodeAsync :: NodeId -> Process ()
870 sendCtrlMsg Nothing . Unlink . NodeIdentifier
872-- | Remove a channel (send port) link (asynchronous)
873unlinkPortAsync :: SendPort a -> Process ()
874unlinkPortAsync (SendPort cid) =
875 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
877--------------------------------------------------------------------------------
879--------------------------------------------------------------------------------
883-- @say message@ sends a message (time, pid of the current process, message)
884-- to the process registered as 'logger'. By default, this process simply
885-- sends the string to 'stderr'. Individual Cloud Haskell backends might
886-- replace this with a different logger process, however.
887say :: String -> Process ()
889 now <- liftIO getCurrentTime
891 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
893--------------------------------------------------------------------------------
895--------------------------------------------------------------------------------
897-- | Register a process with the local registry (asynchronous).
898-- This version will wait until a response is gotten from the
899-- management process. The name must not already be registered.
900-- The process need not be on this node.
901-- A bad registration will result in a 'ProcessRegistrationException'
903-- The process to be registered does not have to be local itself.
904register :: String -> ProcessId -> Process ()
905register = registerImpl False
907-- | Like 'register', but will replace an existing registration.
908-- The name must already be registered.
909reregister :: String -> ProcessId -> Process ()
910reregister = registerImpl True
912registerImpl :: Bool -> String -> ProcessId -> Process ()
913registerImpl force label pid = do
915 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
916 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
917 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
920-- | Register a process with a remote registry (asynchronous).
922-- The process to be registered does not have to live on the same remote node.
923-- Reply wil come in the form of a 'RegisterReply' message
925-- See comments in 'whereisRemoteAsync'
926registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
927registerRemoteAsync nid label pid =
928 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
930reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
931reregisterRemoteAsync nid label pid =
932 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
934-- | Remove a process from the local registry (asynchronous).
935-- This version will wait until a response is gotten from the
936-- management process. The name must already be registered.
937unregister :: String -> Process ()
940 sendCtrlMsg Nothing (Register label mynid Nothing False)
941 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
942 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
945-- | Deal with the result from an attempted registration or unregistration
946-- by throwing an exception if necessary
947handleRegistrationReply :: String -> Bool -> Process ()
948handleRegistrationReply label ok =
950 liftIO $ throwIO $ ProcessRegistrationException label
952-- | Remove a process from a remote registry (asynchronous).
954-- Reply wil come in the form of a 'RegisterReply' message
956-- See comments in 'whereisRemoteAsync'
957unregisterRemoteAsync :: NodeId -> String -> Process ()
958unregisterRemoteAsync nid label =
959 sendCtrlMsg (Just nid) (Register label nid Nothing False)
961-- | Query the local process registry
962whereis :: String -> Process (Maybe ProcessId)
964 sendCtrlMsg Nothing (WhereIs label)
965 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
966 (\(WhereIsReply _ mPid) -> return mPid)
969-- | Query a remote process registry (asynchronous)
971-- Reply will come in the form of a 'WhereIsReply' message.
973-- There is currently no synchronous version of 'whereisRemoteAsync': if
974-- you implement one yourself, be sure to take into account that the remote
975-- node might die or get disconnect before it can respond (i.e. you should
976-- use 'monitorNode' and take appropriate action when you receive a
977-- 'NodeMonitorNotification').
978whereisRemoteAsync :: NodeId -> String -> Process ()
979whereisRemoteAsync nid label =
980 sendCtrlMsg (Just nid) (WhereIs label)
982-- | Named send to a process in the local registry (asynchronous)
983nsend :: Serializable a => String -> a -> Process ()
985 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
987-- | Named send to a process in a remote registry (asynchronous)
988nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
989nsendRemote nid label msg =
990 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
992--------------------------------------------------------------------------------
994--------------------------------------------------------------------------------
996-- | Resolve a static value
997unStatic :: Typeable a => Static a -> Process a
999 rtable <- remoteTable . processNode <$> ask
1000 case Static.unstatic rtable static of
1001 Left err -> fail $ "Could not resolve static value: " ++ err
1004-- | Resolve a closure
1005unClosure :: Typeable a => Closure a -> Process a
1006unClosure closure = do
1007 rtable <- remoteTable . processNode <$> ask
1008 case Static.unclosure rtable closure of
1009 Left err -> fail $ "Could not resolve closure: " ++ err
1012--------------------------------------------------------------------------------
1014--------------------------------------------------------------------------------
1016-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1017-- message passing. However, when network connections get disrupted this
1018-- illusion cannot always be maintained. Once a network connection breaks (even
1019-- temporarily) no further communication on that connection will be possible.
1020-- For example, if process A sends a message to process B, and A is then
1021-- notified (by monitor notification) that it got disconnected from B, A will
1022-- not be able to send any further messages to B, /unless/ A explicitly
1023-- indicates that it is acceptable to attempt to reconnect to B using the
1024-- Cloud Haskell 'reconnect' primitive.
1026-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1027-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1028-- then receives a monitor notification that its connection to B has been lost,
1029-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1030-- and m3 but not m2.
1032-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1033-- to attempt to reconnect on the next send/. In particular, if no further
1034-- communication attempts are made to B then A can use reconnect to clean up
1035-- its connection to B.
1036reconnect :: ProcessId -> Process ()
1039 node <- processNode <$> ask
1040 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1042-- | Reconnect to a sendport. See 'reconnect' for more information.
1043reconnectPort :: SendPort a -> Process ()
1044reconnectPort them = do
1046 node <- processNode <$> ask
1047 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1049--------------------------------------------------------------------------------
1050-- Auxiliary functions --
1051--------------------------------------------------------------------------------
1053sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1055 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1057sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1058sendChanLocal spId msg =
1059 -- we *must* fully serialize/encode the message here, because
1060 -- attempting to use `unsafeCoerce' in the node controller
1061 -- won't work since we know nothing about the required type
1062 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1064getMonitorRefFor :: Identifier -> Process MonitorRef
1065getMonitorRefFor ident = do
1067 liftIO $ modifyMVar (processState proc) $ \st -> do
1068 let counter = st ^. monitorCounter
1069 return ( monitorCounter ^: (+ 1) $ st
1070 , MonitorRef ident counter
1073getSpawnRef :: Process SpawnRef
1076 liftIO $ modifyMVar (processState proc) $ \st -> do
1077 let counter = st ^. spawnCounter
1078 return ( spawnCounter ^: (+ 1) $ st
1082-- | Monitor a process/node/channel
1083monitor' :: Identifier -> Process MonitorRef
1085 monitorRef <- getMonitorRefFor ident
1086 sendCtrlMsg Nothing $ Monitor monitorRef
1089-- | Link to a process/node/channel
1090link' :: Identifier -> Process ()
1091link' = sendCtrlMsg Nothing . Link
1093-- Send a control message
1094sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node
1095 -> ProcessSignal -- ^ Message to send
1097sendCtrlMsg mNid signal = do
1099 let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc)
1100 , ctrlMsgSignal = signal
1104 liftIO $ writeChan (localCtrlChan (processNode proc)) msg
1106 liftIO $ sendBinary (processNode proc)
1107 (ProcessIdentifier (processId proc))
1108 (NodeIdentifier nid)
1109 WithImplicitReconnect
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list