1{-# LANGUAGE BangPatterns #-}
3-- | Cloud Haskell primitives
5-- We define these in a separate module so that we don't have to rely on
6-- the closure combinators
7module Control.Distributed.Process.Internal.Primitives
17 -- * Advanced messaging
37 -- * Process management
39 , ProcessTerminationException(..)
45 -- keep the exception constructor hidden, so that handling exit
46 -- reasons /must/ take place via the 'catchExit' family of primitives
47 , ProcessExitException()
52 -- * Monitoring and linking
67 , reregisterRemoteAsync
68 , unregisterRemoteAsync
74 -- * Exception handling
99#if ! MIN_VERSION_base(4,6,0)
100import Prelude hiding (catch)
103import Data.Binary (decode)
104import Data.Time.Clock (getCurrentTime)
105import Data.Time.Format (formatTime)
106import System.Locale (defaultTimeLocale)
107import System.Timeout (timeout)
108import Control.Monad (when)
109import Control.Monad.Reader (ask)
110import Control.Monad.IO.Class (MonadIO, liftIO)
111import Control.Applicative ((<$>))
112import Control.Exception (Exception(..), throw, throwIO, SomeException)
113import qualified Control.Exception as Ex (catch, mask, try)
114import Control.Distributed.Process.Internal.StrictMVar
119import Control.Concurrent.Chan (writeChan)
120import Control.Concurrent.STM
129import Control.Distributed.Process.Internal.CQueue
134import Control.Distributed.Process.Serializable (Serializable, fingerprint)
135import Data.Accessor ((^.), (^:), (^=))
136import Control.Distributed.Static
149import Data.Rank1Typeable (Typeable)
150import qualified Control.Distributed.Static as Static (unstatic, unclosure)
151import Control.Distributed.Process.Internal.Types
171 , ProcessExitException(..)
173 , DidUnlinkProcess(..)
178 , ProcessRegistrationException(..)
180 , ProcessInfoNone(..)
182 , createUnencodedMessage
184 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
189import Control.Distributed.Process.Internal.Messaging
195import Control.Distributed.Process.Internal.Trace.Types
199import Control.Distributed.Process.Internal.WeakTQueue
207--------------------------------------------------------------------------------
209--------------------------------------------------------------------------------
212send :: Serializable a => ProcessId -> a -> Process ()
213-- This requires a lookup on every send. If we want to avoid that we need to
214-- modify serializable to allow for stateful (IO) deserialization.
217 let us = processId proc
218 node = processNode proc
219 nodeId = localNodeId node
220 destNode = (processNodeId them) in do
221 case destNode == nodeId of
222 True -> sendLocal them msg
223 False -> liftIO $ sendMessage (processNode proc)
224 (ProcessIdentifier (processId proc))
225 (ProcessIdentifier them)
228 -- We do not fire the trace event until after the sending is complete;
229 -- In the remote case, 'sendMessage' can block in the networking stack.
230 liftIO $ traceEvent (localTracer node)
231 (TraceEvSent them us (createUnencodedMessage msg))
233-- | Wait for a message of a specific type
234expect :: forall a. Serializable a => Process a
235expect = receiveWait [match return]
237--------------------------------------------------------------------------------
239--------------------------------------------------------------------------------
241-- | Create a new typed channel
242newChan :: Serializable a => Process (SendPort a, ReceivePort a)
245 liftIO . modifyMVar (processState proc) $ \st -> do
246 let lcid = st ^. channelCounter
247 let cid = SendPortId { sendPortProcessId = processId proc
248 , sendPortLocalId = lcid
250 let sport = SendPort cid
251 chan <- liftIO newTQueueIO
252 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
253 let rport = ReceivePort $ readTQueue chan
254 let tch = TypedChannel chan'
255 return ( (channelCounter ^: (+ 1))
256 . (typedChannelWithId lcid ^= Just tch)
261 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
262 finalizer st lcid = modifyMVar_ st $
263 return . (typedChannelWithId lcid ^= Nothing)
265-- | Send a message on a typed channel
266sendChan :: Serializable a => SendPort a -> a -> Process ()
267sendChan (SendPort cid) msg = do
269 let node = localNodeId (processNode proc)
270 destNode = processNodeId (sendPortProcessId cid) in do
271 case destNode == node of
272 True -> sendChanLocal cid msg
274 liftIO $ sendBinary (processNode proc)
275 (ProcessIdentifier (processId proc))
276 (SendPortIdentifier cid)
280-- | Wait for a message on a typed channel
281receiveChan :: Serializable a => ReceivePort a -> Process a
282receiveChan = liftIO . atomically . receiveSTM
284-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
285-- non-blocking check for a message.
286receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
287receiveChanTimeout 0 ch = liftIO . atomically $
288 (Just <$> receiveSTM ch) `orElse` return Nothing
289receiveChanTimeout n ch = liftIO . timeout n . atomically $
292-- | Merge a list of typed channels.
294-- The result port is left-biased: if there are messages available on more
295-- than one port, the first available message is returned.
296mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
297mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
299-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
301mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
302mergePortsRR = \ps -> do
303 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
304 return $ ReceivePort (rr psVar)
308 rotate (x:xs) = xs ++ [x]
310 rr :: TVar [STM a] -> STM a
313 a <- foldr1 orElse ps
314 writeTVar psVar (rotate ps)
317--------------------------------------------------------------------------------
318-- Advanced messaging --
319--------------------------------------------------------------------------------
321-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
322newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
324-- | Test the matches in order against each message in the queue
325receiveWait :: [Match b] -> Process b
327 queue <- processQueue <$> ask
328 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
331-- | Like 'receiveWait' but with a timeout.
333-- If the timeout is zero do a non-blocking check for matching messages. A
334-- non-zero timeout is applied only when waiting for incoming messages (that is,
335-- /after/ we have checked the messages that are already in the mailbox).
336receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
337receiveTimeout t ms = do
338 queue <- processQueue <$> ask
339 let blockSpec = if t == 0 then NonBlocking else Timeout t
340 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
342 Nothing -> return Nothing
343 Just proc -> Just <$> proc
345matchChan :: ReceivePort a -> (a -> Process b) -> Match b
346matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
348-- | Match against any message of the right type
349match :: forall a b. Serializable a => (a -> Process b) -> Match b
350match = matchIf (const True)
352-- | Match against any message of the right type that satisfies a predicate
353matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
354matchIf c p = Match $ MatchMsg $ \msg ->
355 case messageFingerprint msg == fingerprint (undefined :: a) of
358 (UnencodedMessage _ m) ->
359 let m' = unsafeCoerce m :: a in
363 (EncodedMessage _ _) ->
364 if (c decoded) then Just (p decoded) else Nothing
367 -- Make sure the value is fully decoded so that we don't hang to
368 -- bytestrings when the process calling 'matchIf' doesn't process
369 -- the values immediately
370 !decoded = decode (messageEncoding msg)
372-- | Match against any message, regardless of the underlying (contained) type
373matchMessage :: (Message -> Process Message) -> Match Message
374matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
376-- | Match against any message (regardless of underlying type) that satisfies a predicate
377matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
378matchMessageIf c p = Match $ MatchMsg $ \msg ->
383-- | Forward a raw 'Message' to the given 'ProcessId'.
384forward :: Message -> ProcessId -> Process ()
387 let node = processNode proc
389 nid = localNodeId node
390 destNode = (processNodeId them) in do
391 case destNode == nid of
392 True -> sendCtrlMsg Nothing (LocalSend them msg)
393 False -> liftIO $ sendPayload (processNode proc)
394 (ProcessIdentifier (processId proc))
395 (ProcessIdentifier them)
397 (messageToPayload msg)
398 -- We do not fire the trace event until after the sending is complete;
399 -- In the remote case, 'sendMessage' can block in the networking stack.
400 liftIO $ traceEvent (localTracer node)
401 (TraceEvSent them us (createUnencodedMessage msg))
404-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
405-- 'Serializable' - like the datum they contain - but remember that deserializing
406-- a 'Message' will yield a 'Message', not the type within it! To obtain the
407-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
412-- send self (wrapMessage "blah")
413-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
414-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
415-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
418wrapMessage :: Serializable a => a -> Message
419wrapMessage = createMessage
421-- | Like 'wrapMessage', but produces a 'Message' that is only usable in the
422-- local node. This is primarily useful when consuming the tracing and debugging
423-- APIs in "Control.Distributed.Process.Debug".
424wrapMessageLocal :: Serializable a => a -> Message
425wrapMessageLocal = createUnencodedMessage
427-- | Attempt to unwrap a raw 'Message'.
428-- If the type of the decoded message payload matches the expected type, the
429-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
432-- This expression, for example, will evaluate to @Nothing@
433-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
435-- Whereas this expression, will yield @Just "foo"@
436-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
438unwrapMessage :: forall a. Serializable a => Message -> Process (Maybe a)
440 case messageFingerprint msg == fingerprint (undefined :: a) of
441 False -> return Nothing
443 (UnencodedMessage _ m) ->
444 let m' = unsafeCoerce m :: a
446 (EncodedMessage _ _) ->
447 return (Just (decoded))
449 decoded :: a -- note [decoding]
450 !decoded = decode (messageEncoding msg)
452-- | Attempt to handle a raw 'Message'.
453-- If the type of the message matches the type of the first argument to
454-- the supplied expression, then the message will be decoded and the expression
455-- evaluated against its value. If this runtime type checking fails however,
456-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
457-- evaluation proceeds, the resulting value with be wrapped with @Just@.
459-- Intended for use in `catchesExit` and `matchAny` primitives.
461handleMessage :: forall a b. (Serializable a)
462 => Message -> (a -> Process b) -> Process (Maybe b)
463handleMessage msg proc = do
464 case messageFingerprint msg == fingerprint (undefined :: a) of
465 False -> return Nothing
467 (UnencodedMessage _ m) ->
468 let m' = unsafeCoerce m :: a -- note [decoding]
469 in do { r <- proc m'; return (Just r) }
470 (EncodedMessage _ _) ->
471 do { r <- proc (decoded :: a); return (Just r) }
473 decoded :: a -- note [decoding]
474 !decoded = decode (messageEncoding msg)
476-- | Match against an arbitrary message. 'matchAny' removes the first available
477-- message from the process mailbox. To handle arbitrary /raw/ messages once
478-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
480matchAny :: forall b. (Message -> Process b) -> Match b
481matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
483-- | Match against an arbitrary message. Intended for use with 'handleMessage'
484-- and 'unwrapMessage', this function /only/ removes a message from the process
485-- mailbox, /if/ the supplied condition matches. The success (or failure) of
486-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
487-- here, i.e., if the condition evaluates to @True@ then the message will
488-- be removed from the process mailbox and decoded, but that does /not/
489-- guarantee that an expression passed to @handleMessage@ will pass the
490-- runtime type checks and therefore be evaluated.
492matchAnyIf :: forall a b. (Serializable a)
494 -> (Message -> Process b)
496matchAnyIf c p = Match $ MatchMsg $ \msg ->
497 case messageFingerprint msg == fingerprint (undefined :: a) of
498 True | check -> Just (p msg)
503 (EncodedMessage _ _) -> c decoded
504 (UnencodedMessage _ m') -> c (unsafeCoerce m')
506 decoded :: a -- note [decoding]
507 !decoded = decode (messageEncoding msg)
511For an EncodedMessage, we need to ensure the value is fully decoded so that
512we don't hang to bytestrings if the calling process doesn't evaluate
513immediately. For UnencodedMessage we know (because the fingerprint comparison
514succeeds) that unsafeCoerce will not fail.
517-- | Remove any message from the queue
518matchUnknown :: Process b -> Match b
519matchUnknown p = Match $ MatchMsg (const (Just p))
521-- | Receives messages and forwards them to @pid@ if @p msg == True@.
522delegate :: ProcessId -> (Message -> Bool) -> Process ()
525 matchAny (\m -> case (p m) of
526 True -> forward m pid
531-- | A straight relay that simply forwards all messages to the supplied pid.
532relay :: ProcessId -> Process ()
533relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
535-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
536-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
537-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
538-- fails, no action is taken and the proxy process will continue running.
539proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
543 next <- handleMessage m proc
545 Just True -> forward m pid
546 Just False -> return () -- explicitly ignored
547 Nothing -> return ()) -- un-routable
551--------------------------------------------------------------------------------
552-- Process management --
553--------------------------------------------------------------------------------
555-- | Thrown by 'terminate'
556data ProcessTerminationException = ProcessTerminationException
557 deriving (Show, Typeable)
559instance Exception ProcessTerminationException
561-- | Terminate immediately (throws a ProcessTerminationException)
562terminate :: Process a
563terminate = liftIO $ throwIO ProcessTerminationException
566-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
567die :: Serializable a => a -> Process b
570 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
572-- | Forceful request to kill a process. Where 'exit' provides an exception
573-- that can be caught and handled, 'kill' throws an unexposed exception type
574-- which cannot be handled explicitly (by type).
575kill :: ProcessId -> String -> Process ()
576-- NOTE: We send the message to our local node controller, which will then
577-- forward it to a remote node controller (if applicable). Sending it directly
578-- to a remote node controller means that that the message may overtake a
579-- 'monitor' or 'link' request.
580kill them reason = sendCtrlMsg Nothing (Kill them reason)
582-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
583-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
584-- manner can be handled using the 'catchExit' family of functions.
585exit :: Serializable a => ProcessId -> a -> Process ()
586-- NOTE: We send the message to our local node controller, which will then
587-- forward it to a remote node controller (if applicable). Sending it directly
588-- to a remote node controller means that that the message may overtake a
589-- 'monitor' or 'link' request.
590exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
592-- | Catches 'ProcessExitException'. The handler will not be applied unless its
593-- type matches the encoded data stored in the exception (see the /reason/
594-- argument given to the 'exit' primitive). If the handler cannot be applied,
595-- the exception will be re-thrown.
597-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
598-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
599catchExit :: forall a b . (Show a, Serializable a)
601 -> (ProcessId -> a -> Process b)
603catchExit act exitHandler = catch act handleExit
605 handleExit ex@(ProcessExitException from msg) =
606 if messageFingerprint msg == fingerprint (undefined :: a)
607 then exitHandler from decoded
608 else liftIO $ throwIO ex
611 -- Make sure the value is fully decoded so that we don't hang to
612 -- bytestrings if the caller doesn't use the value immediately
613 !decoded = decode (messageEncoding msg)
615-- | Lift 'Control.Exception.catches' (almost).
617-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
618-- message, a handler must accept an input of the expected type. In order to
619-- handle a list of potentially different handlers (and therefore input types),
620-- a handler passed to 'catchesExit' must accept 'AbstractMessage' and return
621-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
623-- See 'maybeHandleMessage' and 'AsbtractMessage' for more details.
624catchesExit :: Process b
625 -> [(ProcessId -> Message -> (Process (Maybe b)))]
627catchesExit act handlers = catch act ((flip handleExit) handlers)
629 handleExit :: ProcessExitException
630 -> [(ProcessId -> Message -> Process (Maybe b))]
632 handleExit ex [] = liftIO $ throwIO ex
633 handleExit ex@(ProcessExitException from msg) (h:hs) = do
636 Nothing -> handleExit ex hs
639-- | Our own process ID
640getSelfPid :: Process ProcessId
641getSelfPid = processId <$> ask
643-- | Get the node ID of our local node
644getSelfNode :: Process NodeId
645getSelfNode = localNodeId . processNode <$> ask
647-- | Get information about the specified process
648getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
650 let them = processNodeId pid in do
652 dest <- mkNode them us
653 sendCtrlMsg dest $ GetInfo pid
655 match (\(p :: ProcessInfo) -> return $ Just p)
656 , match (\(_ :: ProcessInfoNone) -> return Nothing)
658 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
659 mkNode them us = case them == us of
660 True -> return Nothing
661 _ -> return $ Just them
663--------------------------------------------------------------------------------
664-- Monitoring and linking --
665--------------------------------------------------------------------------------
667-- | Link to a remote process (asynchronous)
669-- When process A links to process B (that is, process A calls
670-- @link pidB@) then an asynchronous exception will be thrown to process A
671-- when process B terminates (normally or abnormally), or when process A gets
672-- disconnected from process B. Although it is /technically/ possible to catch
673-- these exceptions, chances are if you find yourself trying to do so you should
674-- probably be using 'monitor' rather than 'link'. In particular, code such as
676-- > link pidB -- Link to process B
677-- > expect -- Wait for a message from process B
678-- > unlink pidB -- Unlink again
680-- doesn't quite do what one might expect: if process B sends a message to
681-- process A, and /subsequently terminates/, then process A might or might not
682-- be terminated too, depending on whether the exception is thrown before or
683-- after the 'unlink' (i.e., this code has a race condition).
685-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
686-- call to 'link' has no effect.
688-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
689-- Linking makes no distinction between normal and abnormal termination of
690-- the remote process.
691link :: ProcessId -> Process ()
692link = sendCtrlMsg Nothing . Link . ProcessIdentifier
694-- | Monitor another process (asynchronous)
696-- When process A monitors process B (that is, process A calls
697-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
698-- when process B terminates (normally or abnormally), or when process A gets
699-- disconnected from process B. You receive this message like any other (using
700-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
701-- 'DiedDisconnect', etc.).
703-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
704-- multiple monitors are set up, multiple notifications will be delivered
705-- and monitors can be disabled individually using 'unmonitor'.
706monitor :: ProcessId -> Process MonitorRef
707monitor = monitor' . ProcessIdentifier
709-- | Establishes temporary monitoring of another process.
711-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
712-- of @code@. Note: although monitoring is no longer active when
713-- @withMonitor@ returns, there might still be unreceived monitor
714-- messages in the queue.
716withMonitor :: ProcessId -> Process a -> Process a
717withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
718 -- unmonitor blocks waiting for the response, so there's a possibility
719 -- that an exception might interrupt withMonitor before the unmonitor
720 -- has completed. I think that's better than making the unmonitor
725-- This is synchronous in the sense that once it returns you are guaranteed
726-- that no exception will be raised if the remote process dies. However, it is
727-- asynchronous in the sense that we do not wait for a response from the remote
729unlink :: ProcessId -> Process ()
732 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
736-- | Remove a node link
738-- This has the same synchronous/asynchronous nature as 'unlink'.
739unlinkNode :: NodeId -> Process ()
742 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
746-- | Remove a channel (send port) link
748-- This has the same synchronous/asynchronous nature as 'unlink'.
749unlinkPort :: SendPort a -> Process ()
751 unlinkPortAsync sport
752 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
758-- This has the same synchronous/asynchronous nature as 'unlink'.
759unmonitor :: MonitorRef -> Process ()
762 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
766--------------------------------------------------------------------------------
767-- Exception handling --
768--------------------------------------------------------------------------------
770-- | Lift 'Control.Exception.catch'
771catch :: Exception e => Process a -> (e -> Process a) -> Process a
774 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
776-- | Lift 'Control.Exception.try'
777try :: Exception e => Process a -> Process (Either e a)
780 liftIO $ Ex.try (runLocalProcess lproc p)
782-- | Lift 'Control.Exception.mask'
783mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
786 liftIO $ Ex.mask $ \restore ->
787 runLocalProcess lproc (p (liftRestore restore))
789 liftRestore :: (forall a. IO a -> IO a)
790 -> (forall a. Process a -> Process a)
791 liftRestore restoreIO = \p2 -> do
793 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
795-- | Lift 'Control.Exception.onException'
796onException :: Process a -> Process b -> Process a
797onException p what = p `catch` \e -> do _ <- what
798 liftIO $ throwIO (e :: SomeException)
800-- | Lift 'Control.Exception.bracket'
801bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
802bracket before after thing =
803 mask $ \restore -> do
805 r <- restore (thing a) `onException` after a
809-- | Lift 'Control.Exception.bracket_'
810bracket_ :: Process a -> Process b -> Process c -> Process c
811bracket_ before after thing = bracket before (const after) (const thing)
813-- | Lift 'Control.Exception.finally'
814finally :: Process a -> Process b -> Process a
815finally a sequel = bracket_ (return ()) sequel a
817-- | You need this when using 'catches'
818data Handler a = forall e . Exception e => Handler (e -> Process a)
820instance Functor Handler where
821 fmap f (Handler h) = Handler (fmap f . h)
823-- | Lift 'Control.Exception.catches'
824catches :: Process a -> [Handler a] -> Process a
825catches proc handlers = proc `catch` catchesHandler handlers
827catchesHandler :: [Handler a] -> SomeException -> Process a
828catchesHandler handlers e = foldr tryHandler (throw e) handlers
829 where tryHandler (Handler handler) res
830 = case fromException e of
831 Just e' -> handler e'
834--------------------------------------------------------------------------------
836--------------------------------------------------------------------------------
838-- | Like 'expect' but with a timeout
839expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
840expectTimeout n = receiveTimeout n [match return]
842-- | Asynchronous version of 'spawn'
844-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
845spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
846spawnAsync nid proc = do
847 spawnRef <- getSpawnRef
848 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
851-- | Monitor a node (asynchronous)
852monitorNode :: NodeId -> Process MonitorRef
854 monitor' . NodeIdentifier
856-- | Monitor a typed channel (asynchronous)
857monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
858monitorPort (SendPort cid) =
859 monitor' (SendPortIdentifier cid)
861-- | Remove a monitor (asynchronous)
862unmonitorAsync :: MonitorRef -> Process ()
864 sendCtrlMsg Nothing . Unmonitor
866-- | Link to a node (asynchronous)
867linkNode :: NodeId -> Process ()
868linkNode = link' . NodeIdentifier
870-- | Link to a channel (asynchronous)
871linkPort :: SendPort a -> Process ()
872linkPort (SendPort cid) =
873 link' (SendPortIdentifier cid)
875-- | Remove a process link (asynchronous)
876unlinkAsync :: ProcessId -> Process ()
878 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
880-- | Remove a node link (asynchronous)
881unlinkNodeAsync :: NodeId -> Process ()
883 sendCtrlMsg Nothing . Unlink . NodeIdentifier
885-- | Remove a channel (send port) link (asynchronous)
886unlinkPortAsync :: SendPort a -> Process ()
887unlinkPortAsync (SendPort cid) =
888 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
890--------------------------------------------------------------------------------
892--------------------------------------------------------------------------------
896-- @say message@ sends a message (time, pid of the current process, message)
897-- to the process registered as 'logger'. By default, this process simply
898-- sends the string to 'stderr'. Individual Cloud Haskell backends might
899-- replace this with a different logger process, however.
900say :: String -> Process ()
902 now <- liftIO getCurrentTime
904 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
906--------------------------------------------------------------------------------
908--------------------------------------------------------------------------------
910-- | Register a process with the local registry (asynchronous).
911-- This version will wait until a response is gotten from the
912-- management process. The name must not already be registered.
913-- The process need not be on this node.
914-- A bad registration will result in a 'ProcessRegistrationException'
916-- The process to be registered does not have to be local itself.
917register :: String -> ProcessId -> Process ()
918register = registerImpl False
920-- | Like 'register', but will replace an existing registration.
921-- The name must already be registered.
922reregister :: String -> ProcessId -> Process ()
923reregister = registerImpl True
925registerImpl :: Bool -> String -> ProcessId -> Process ()
926registerImpl force label pid = do
928 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
929 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
930 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
933-- | Register a process with a remote registry (asynchronous).
935-- The process to be registered does not have to live on the same remote node.
936-- Reply wil come in the form of a 'RegisterReply' message
938-- See comments in 'whereisRemoteAsync'
939registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
940registerRemoteAsync nid label pid =
941 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
943reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
944reregisterRemoteAsync nid label pid =
945 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
947-- | Remove a process from the local registry (asynchronous).
948-- This version will wait until a response is gotten from the
949-- management process. The name must already be registered.
950unregister :: String -> Process ()
953 sendCtrlMsg Nothing (Register label mynid Nothing False)
954 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
955 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
958-- | Deal with the result from an attempted registration or unregistration
959-- by throwing an exception if necessary
960handleRegistrationReply :: String -> Bool -> Process ()
961handleRegistrationReply label ok =
963 liftIO $ throwIO $ ProcessRegistrationException label
965-- | Remove a process from a remote registry (asynchronous).
967-- Reply wil come in the form of a 'RegisterReply' message
969-- See comments in 'whereisRemoteAsync'
970unregisterRemoteAsync :: NodeId -> String -> Process ()
971unregisterRemoteAsync nid label =
972 sendCtrlMsg (Just nid) (Register label nid Nothing False)
974-- | Query the local process registry
975whereis :: String -> Process (Maybe ProcessId)
977 sendCtrlMsg Nothing (WhereIs label)
978 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
979 (\(WhereIsReply _ mPid) -> return mPid)
982-- | Query a remote process registry (asynchronous)
984-- Reply will come in the form of a 'WhereIsReply' message.
986-- There is currently no synchronous version of 'whereisRemoteAsync': if
987-- you implement one yourself, be sure to take into account that the remote
988-- node might die or get disconnect before it can respond (i.e. you should
989-- use 'monitorNode' and take appropriate action when you receive a
990-- 'NodeMonitorNotification').
991whereisRemoteAsync :: NodeId -> String -> Process ()
992whereisRemoteAsync nid label =
993 sendCtrlMsg (Just nid) (WhereIs label)
995-- | Named send to a process in the local registry (asynchronous)
996nsend :: Serializable a => String -> a -> Process ()
998 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1000-- | Named send to a process in a remote registry (asynchronous)
1001nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1002nsendRemote nid label msg =
1003 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1005--------------------------------------------------------------------------------
1007--------------------------------------------------------------------------------
1009-- | Resolve a static value
1010unStatic :: Typeable a => Static a -> Process a
1012 rtable <- remoteTable . processNode <$> ask
1013 case Static.unstatic rtable static of
1014 Left err -> fail $ "Could not resolve static value: " ++ err
1017-- | Resolve a closure
1018unClosure :: Typeable a => Closure a -> Process a
1019unClosure closure = do
1020 rtable <- remoteTable . processNode <$> ask
1021 case Static.unclosure rtable closure of
1022 Left err -> fail $ "Could not resolve closure: " ++ err
1025--------------------------------------------------------------------------------
1027--------------------------------------------------------------------------------
1029-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1030-- message passing. However, when network connections get disrupted this
1031-- illusion cannot always be maintained. Once a network connection breaks (even
1032-- temporarily) no further communication on that connection will be possible.
1033-- For example, if process A sends a message to process B, and A is then
1034-- notified (by monitor notification) that it got disconnected from B, A will
1035-- not be able to send any further messages to B, /unless/ A explicitly
1036-- indicates that it is acceptable to attempt to reconnect to B using the
1037-- Cloud Haskell 'reconnect' primitive.
1039-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1040-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1041-- then receives a monitor notification that its connection to B has been lost,
1042-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1043-- and m3 but not m2.
1045-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1046-- to attempt to reconnect on the next send/. In particular, if no further
1047-- communication attempts are made to B then A can use reconnect to clean up
1048-- its connection to B.
1049reconnect :: ProcessId -> Process ()
1052 node <- processNode <$> ask
1053 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1055-- | Reconnect to a sendport. See 'reconnect' for more information.
1056reconnectPort :: SendPort a -> Process ()
1057reconnectPort them = do
1059 node <- processNode <$> ask
1060 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1062--------------------------------------------------------------------------------
1063-- Auxiliary functions --
1064--------------------------------------------------------------------------------
1066sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1068 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1070sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1071sendChanLocal spId msg =
1072 -- we *must* fully serialize/encode the message here, because
1073 -- attempting to use `unsafeCoerce' in the node controller
1074 -- won't work since we know nothing about the required type
1075 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1077getMonitorRefFor :: Identifier -> Process MonitorRef
1078getMonitorRefFor ident = do
1080 liftIO $ modifyMVar (processState proc) $ \st -> do
1081 let counter = st ^. monitorCounter
1082 return ( monitorCounter ^: (+ 1) $ st
1083 , MonitorRef ident counter
1086getSpawnRef :: Process SpawnRef
1089 liftIO $ modifyMVar (processState proc) $ \st -> do
1090 let counter = st ^. spawnCounter
1091 return ( spawnCounter ^: (+ 1) $ st
1095-- | Monitor a process/node/channel
1096monitor' :: Identifier -> Process MonitorRef
1098 monitorRef <- getMonitorRefFor ident
1099 sendCtrlMsg Nothing $ Monitor monitorRef
1102-- | Link to a process/node/channel
1103link' :: Identifier -> Process ()
1104link' = sendCtrlMsg Nothing . Link
1106-- Send a control message
1107sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node
1108 -> ProcessSignal -- ^ Message to send
1110sendCtrlMsg mNid signal = do
1112 let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc)
1113 , ctrlMsgSignal = signal
1117 liftIO $ writeChan (localCtrlChan (processNode proc)) msg
1119 liftIO $ sendBinary (processNode proc)
1120 (ProcessIdentifier (processId proc))
1121 (NodeIdentifier nid)
1122 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