2{-# LANGUAGE RankNTypes #-}
3{-# LANGUAGE DeriveDataTypeable #-}
4{-# LANGUAGE ScopedTypeVariables #-}
5{-# LANGUAGE ExistentialQuantification #-}
6{-# LANGUAGE BangPatterns #-}
8-- | Cloud Haskell primitives
10-- We define these in a separate module so that we don't have to rely on
11-- the closure combinators
12module Control.Distributed.Process.Internal.Primitives
13 ( -- * Basic messaging
22 -- * Unsafe messaging variants
26 -- * Advanced messaging
51 -- * Process management
53 , ProcessTerminationException(..)
59 -- keep the exception constructor hidden, so that handling exit
60 -- reasons /must/ take place via the 'catchExit' family of primitives
61 , ProcessExitException()
69 -- * Monitoring and linking
84 , reregisterRemoteAsync
85 , unregisterRemoteAsync
91 -- * Exception handling
115 -- * Internal Exports
119#if ! MIN_VERSION_base(4,6,0)
120import Prelude hiding (catch)
123import Data.Binary (decode)
124import Data.Time.Clock (getCurrentTime)
125import Data.Time.Format (formatTime)
126import System.Locale (defaultTimeLocale)
127import System.Timeout (timeout)
128import Control.Monad (when)
129import Control.Monad.Reader (ask)
130import Control.Monad.IO.Class (liftIO)
131import Control.Applicative ((<$>))
132import Control.Exception (Exception(..), throw, throwIO, SomeException)
133import qualified Control.Exception as Ex (catch, mask, mask_, try)
134import Control.Distributed.Process.Internal.StrictMVar
139import Control.Concurrent.STM
148import Control.Distributed.Process.Internal.CQueue
153import Control.Distributed.Process.Serializable (Serializable, fingerprint)
154import Data.Accessor ((^.), (^:), (^=))
155import Control.Distributed.Static
159import Data.Rank1Typeable (Typeable)
160import qualified Control.Distributed.Static as Static (unstatic, unclosure)
161import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
162import Control.Distributed.Process.Internal.Types
172 , NodeMonitorNotification(..)
182 , ProcessExitException(..)
185 , DidUnlinkProcess(..)
190 , ProcessRegistrationException(..)
192 , ProcessInfoNone(..)
196 , createUnencodedMessage
198 , ImplicitReconnect( NoImplicitReconnect)
203import Control.Distributed.Process.Internal.Messaging
210import Control.Distributed.Process.Management.Internal.Types
213import Control.Distributed.Process.Management.Internal.Trace.Types
216import Control.Distributed.Process.Internal.WeakTQueue
224--------------------------------------------------------------------------------
226--------------------------------------------------------------------------------
229send :: Serializable a => ProcessId -> a -> Process ()
230-- This requires a lookup on every send. If we want to avoid that we need to
231-- modify serializable to allow for stateful (IO) deserialization.
234 let us = processId proc
235 node = processNode proc
236 nodeId = localNodeId node
237 destNode = (processNodeId them) in do
238 case destNode == nodeId of
239 True -> sendLocal them msg
240 False -> liftIO $ sendMessage (processNode proc)
241 (ProcessIdentifier (processId proc))
242 (ProcessIdentifier them)
245 -- We do not fire the trace event until after the sending is complete;
246 -- In the remote case, 'sendMessage' can block in the networking stack.
247 liftIO $ traceEvent (localEventBus node)
248 (MxSent them us (createUnencodedMessage msg))
250-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
251-- and (in the case when the destination process resides on the same local
252-- node) therefore ensure that the payload is fully evaluated before it is
254unsafeSend :: Serializable a => ProcessId -> a -> Process ()
255unsafeSend = Unsafe.send
257-- | Wait for a message of a specific type
258expect :: forall a. Serializable a => Process a
259expect = receiveWait [match return]
261--------------------------------------------------------------------------------
263--------------------------------------------------------------------------------
265-- | Create a new typed channel
266newChan :: Serializable a => Process (SendPort a, ReceivePort a)
269 liftIO . modifyMVar (processState proc) $ \st -> do
270 let lcid = st ^. channelCounter
271 let cid = SendPortId { sendPortProcessId = processId proc
272 , sendPortLocalId = lcid
274 let sport = SendPort cid
275 chan <- liftIO newTQueueIO
276 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
277 let rport = ReceivePort $ readTQueue chan
278 let tch = TypedChannel chan'
279 return ( (channelCounter ^: (+ 1))
280 . (typedChannelWithId lcid ^= Just tch)
285 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
286 finalizer st lcid = modifyMVar_ st $
287 return . (typedChannelWithId lcid ^= Nothing)
289-- | Send a message on a typed channel
290sendChan :: Serializable a => SendPort a -> a -> Process ()
291sendChan (SendPort cid) msg = do
293 let node = localNodeId (processNode proc)
294 destNode = processNodeId (sendPortProcessId cid) in do
295 case destNode == node of
296 True -> sendChanLocal cid msg
298 liftIO $ sendBinary (processNode proc)
299 (ProcessIdentifier (processId proc))
300 (SendPortIdentifier cid)
304-- | Send a message on a typed channel. This function makes /no/ attempt to
305-- serialize and (in the case when the @ReceivePort@ resides on the same local
306-- node) therefore ensure that the payload is fully evaluated before it is
308unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
309unsafeSendChan = Unsafe.sendChan
311-- | Wait for a message on a typed channel
312receiveChan :: Serializable a => ReceivePort a -> Process a
313receiveChan = liftIO . atomically . receiveSTM
315-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
316-- non-blocking check for a message.
317receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
318receiveChanTimeout 0 ch = liftIO . atomically $
319 (Just <$> receiveSTM ch) `orElse` return Nothing
320receiveChanTimeout n ch = liftIO . timeout n . atomically $
323-- | Merge a list of typed channels.
325-- The result port is left-biased: if there are messages available on more
326-- than one port, the first available message is returned.
327mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
328mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
330-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
332mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
333mergePortsRR = \ps -> do
334 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
335 return $ ReceivePort (rr psVar)
339 rotate (x:xs) = xs ++ [x]
341 rr :: TVar [STM a] -> STM a
344 a <- foldr1 orElse ps
345 writeTVar psVar (rotate ps)
348--------------------------------------------------------------------------------
349-- Advanced messaging --
350--------------------------------------------------------------------------------
352-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
353newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
355-- | Test the matches in order against each message in the queue
356receiveWait :: [Match b] -> Process b
358 queue <- processQueue <$> ask
359 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
362-- | Like 'receiveWait' but with a timeout.
364-- If the timeout is zero do a non-blocking check for matching messages. A
365-- non-zero timeout is applied only when waiting for incoming messages (that is,
366-- /after/ we have checked the messages that are already in the mailbox).
367receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
368receiveTimeout t ms = do
369 queue <- processQueue <$> ask
370 let blockSpec = if t == 0 then NonBlocking else Timeout t
371 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
373 Nothing -> return Nothing
374 Just proc -> Just <$> proc
376-- | Match on a typed channel
377matchChan :: ReceivePort a -> (a -> Process b) -> Match b
378matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
380-- | Match on an arbitrary STM action.
382-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
383-- actions with checks against our process' mailbox and/or any typed channel
384-- @ReceivePort@s we may hold.
386-- This allows us to process multiple input streams /along with our mailbox/,
387-- in just the same way that 'matchChan' supports checking both the mailbox
388-- /and/ an arbitrary set of typed channels in one atomic transaction.
390-- Note there are no ordering guarnatees with respect to these disparate input
393matchSTM :: STM a -> (a -> Process b) -> Match b
394matchSTM stm fn = Match $ MatchChan (fmap fn stm)
396-- | Match against any message of the right type
397match :: forall a b. Serializable a => (a -> Process b) -> Match b
398match = matchIf (const True)
400-- | Match against any message of the right type that satisfies a predicate
401matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
402matchIf c p = Match $ MatchMsg $ \msg ->
403 case messageFingerprint msg == fingerprint (undefined :: a) of
406 (UnencodedMessage _ m) ->
407 let m' = unsafeCoerce m :: a in
411 (EncodedMessage _ _) ->
412 if (c decoded) then Just (p decoded) else Nothing
415 -- Make sure the value is fully decoded so that we don't hang to
416 -- bytestrings when the process calling 'matchIf' doesn't process
417 -- the values immediately
418 !decoded = decode (messageEncoding msg)
420-- | Match against any message, regardless of the underlying (contained) type
421matchMessage :: (Message -> Process Message) -> Match Message
422matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
424-- | Match against any message (regardless of underlying type) that satisfies a predicate
425matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
426matchMessageIf c p = Match $ MatchMsg $ \msg ->
431-- | Forward a raw 'Message' to the given 'ProcessId'.
432forward :: Message -> ProcessId -> Process ()
435 let node = processNode proc
437 nid = localNodeId node
438 destNode = (processNodeId them) in do
439 case destNode == nid of
440 True -> sendCtrlMsg Nothing (LocalSend them msg)
441 False -> liftIO $ sendPayload (processNode proc)
442 (ProcessIdentifier (processId proc))
443 (ProcessIdentifier them)
445 (messageToPayload msg)
446 -- We do not fire the trace event until after the sending is complete;
447 -- In the remote case, 'sendMessage' can block in the networking stack.
448 liftIO $ traceEvent (localEventBus node)
452-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
453-- 'Serializable' - like the datum they contain - but also note, deserialising
454-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
455-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
460-- send self (wrapMessage "blah")
461-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
462-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
463-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
466wrapMessage :: Serializable a => a -> Message
467wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
469-- | This is the /unsafe/ variant of 'wrapMessage'. See
470-- "Control.Distributed.Process.UnsafePrimitives" for details.
471unsafeWrapMessage :: Serializable a => a -> Message
472unsafeWrapMessage = Unsafe.wrapMessage
474-- [note Serializable UnencodedMessage]
475-- if we attempt to serialise an UnencodedMessage, it will be converted to an
476-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
477-- Primitives.hs for the gory details
479-- | Attempt to unwrap a raw 'Message'.
480-- If the type of the decoded message payload matches the expected type, the
481-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
484-- This expression, for example, will evaluate to @Nothing@
485-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
487-- Whereas this expression, will yield @Just "foo"@
488-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
490unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
492 case messageFingerprint msg == fingerprint (undefined :: a) of
493 False -> return Nothing :: m (Maybe a)
495 (UnencodedMessage _ ms) ->
496 let ms' = unsafeCoerce ms :: a
498 (EncodedMessage _ _) ->
499 return (Just (decoded))
501 decoded :: a -- note [decoding]
502 !decoded = decode (messageEncoding msg)
504-- | Attempt to handle a raw 'Message'.
505-- If the type of the message matches the type of the first argument to
506-- the supplied expression, then the message will be decoded and the expression
507-- evaluated against its value. If this runtime type checking fails however,
508-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
509-- evaluation proceeds, the resulting value with be wrapped with @Just@.
511-- Intended for use in `catchesExit` and `matchAny` primitives.
513handleMessage :: forall m a b. (Monad m, Serializable a)
514 => Message -> (a -> m b) -> m (Maybe b)
515handleMessage msg proc = handleMessageIf msg (const True) proc
517-- | Conditionally handle a raw 'Message'.
518-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
519-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
520-- for further information about runtime type checking.
522handleMessageIf :: forall m a b . (Monad m, Serializable a)
527handleMessageIf msg c proc = do
528 case messageFingerprint msg == fingerprint (undefined :: a) of
529 False -> return Nothing :: m (Maybe b)
531 (UnencodedMessage _ ms) ->
532 let ms' = unsafeCoerce ms :: a in
534 True -> do { r <- proc ms'; return (Just r) }
535 False -> return Nothing :: m (Maybe b)
536 (EncodedMessage _ _) ->
538 True -> do { r <- proc (decoded :: a); return (Just r) }
539 False -> return Nothing :: m (Maybe b)
541 decoded :: a -- note [decoding]
542 !decoded = decode (messageEncoding msg)
544-- | As 'handleMessage' but ignores result, which is useful if you don't
545-- care whether or not the handler succeeded.
547handleMessage_ :: forall m a . (Monad m, Serializable a)
548 => Message -> (a -> m ()) -> m ()
549handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
551-- | Conditional version of 'handleMessage_'.
553handleMessageIf_ :: forall m a . (Monad m, Serializable a)
558handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
560-- | Match against an arbitrary message. 'matchAny' removes the first available
561-- message from the process mailbox. To handle arbitrary /raw/ messages once
562-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
564matchAny :: forall b. (Message -> Process b) -> Match b
565matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
567-- | Match against an arbitrary message. Intended for use with 'handleMessage'
568-- and 'unwrapMessage', this function /only/ removes a message from the process
569-- mailbox, /if/ the supplied condition matches. The success (or failure) of
570-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
571-- here, i.e., if the condition evaluates to @True@ then the message will
572-- be removed from the process mailbox and decoded, but that does /not/
573-- guarantee that an expression passed to @handleMessage@ will pass the
574-- runtime type checks and therefore be evaluated.
576matchAnyIf :: forall a b. (Serializable a)
578 -> (Message -> Process b)
580matchAnyIf c p = Match $ MatchMsg $ \msg ->
581 case messageFingerprint msg == fingerprint (undefined :: a) of
582 True | check -> Just (p msg)
587 (EncodedMessage _ _) -> c decoded
588 (UnencodedMessage _ m') -> c (unsafeCoerce m')
590 decoded :: a -- note [decoding]
591 !decoded = decode (messageEncoding msg)
595For an EncodedMessage, we need to ensure the value is fully decoded so that
596we don't hang to bytestrings if the calling process doesn't evaluate
597immediately. For UnencodedMessage we know (because the fingerprint comparison
598succeeds) that unsafeCoerce will not fail.
601-- | Remove any message from the queue
602matchUnknown :: Process b -> Match b
603matchUnknown p = Match $ MatchMsg (const (Just p))
605-- | Receives messages and forwards them to @pid@ if @p msg == True@.
606delegate :: ProcessId -> (Message -> Bool) -> Process ()
609 matchAny (\m -> case (p m) of
610 True -> forward m pid
615-- | A straight relay that forwards all messages to the supplied pid.
616relay :: ProcessId -> Process ()
617relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
619-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
620-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
621-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
622-- fails, no action is taken and the proxy process will continue running.
623proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
627 next <- handleMessage m proc
629 Just True -> forward m pid
630 Just False -> return () -- explicitly ignored
631 Nothing -> return ()) -- un-routable
635--------------------------------------------------------------------------------
636-- Process management --
637--------------------------------------------------------------------------------
639-- | Thrown by 'terminate'
640data ProcessTerminationException = ProcessTerminationException
641 deriving (Show, Typeable)
643instance Exception ProcessTerminationException
645-- | Terminate immediately (throws a ProcessTerminationException)
646terminate :: Process a
647terminate = liftIO $ throwIO ProcessTerminationException
650-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
651die :: Serializable a => a -> Process b
653-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
654-- could be decoded by a handler passed to 'catchExit', re-thrown or even
655-- passed to another process via the tracing mechanism.
657 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
659-- | Forceful request to kill a process. Where 'exit' provides an exception
660-- that can be caught and handled, 'kill' throws an unexposed exception type
661-- which cannot be handled explicitly (by type).
662kill :: ProcessId -> String -> Process ()
663-- NOTE: We send the message to our local node controller, which will then
664-- forward it to a remote node controller (if applicable). Sending it directly
665-- to a remote node controller means that that the message may overtake a
666-- 'monitor' or 'link' request.
667kill them reason = sendCtrlMsg Nothing (Kill them reason)
669-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
670-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
671-- manner can be handled using the 'catchExit' family of functions.
672exit :: Serializable a => ProcessId -> a -> Process ()
673-- NOTE: We send the message to our local node controller, which will then
674-- forward it to a remote node controller (if applicable). Sending it directly
675-- to a remote node controller means that that the message may overtake a
676-- 'monitor' or 'link' request.
677exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
679-- | Catches 'ProcessExitException'. The handler will not be applied unless its
680-- type matches the encoded data stored in the exception (see the /reason/
681-- argument given to the 'exit' primitive). If the handler cannot be applied,
682-- the exception will be re-thrown.
684-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
685-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
686catchExit :: forall a b . (Show a, Serializable a)
688 -> (ProcessId -> a -> Process b)
690catchExit act exitHandler = catch act handleExit
692 handleExit ex@(ProcessExitException from msg) =
693 if messageFingerprint msg == fingerprint (undefined :: a)
694 then exitHandler from decoded
695 else liftIO $ throwIO ex
698 -- Make sure the value is fully decoded so that we don't hang to
699 -- bytestrings if the caller doesn't use the value immediately
700 !decoded = decode (messageEncoding msg)
702-- | Lift 'Control.Exception.catches' (almost).
704-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
705-- message, a handler must accept inputs of the expected type. In order to
706-- handle a list of potentially different handlers (and therefore input types),
707-- a handler passed to 'catchesExit' must accept 'Message' and return
708-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
710-- See 'maybeHandleMessage' and 'Message' for more details.
711catchesExit :: Process b
712 -> [(ProcessId -> Message -> (Process (Maybe b)))]
714catchesExit act handlers = catch act ((flip handleExit) handlers)
716 handleExit :: ProcessExitException
717 -> [(ProcessId -> Message -> Process (Maybe b))]
719 handleExit ex [] = liftIO $ throwIO ex
720 handleExit ex@(ProcessExitException from msg) (h:hs) = do
723 Nothing -> handleExit ex hs
726-- | Our own process ID
727getSelfPid :: Process ProcessId
728getSelfPid = processId <$> ask
730-- | Get the node ID of our local node
731getSelfNode :: Process NodeId
732getSelfNode = localNodeId . processNode <$> ask
735-- | Get statistics about the specified node
736getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
738 selfNode <- getSelfNode
740 then Right `fmap` getLocalNodeStats -- optimisation
741 else getNodeStatsRemote
743 getNodeStatsRemote :: Process (Either DiedReason NodeStats)
744 getNodeStatsRemote = do
745 sendCtrlMsg (Just nid) $ GetNodeStats nid
746 bracket (monitorNode nid) unmonitor $ \mRef ->
747 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
748 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
749 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
752-- | Get statistics about our local node
753getLocalNodeStats :: Process NodeStats
754getLocalNodeStats = do
756 sendCtrlMsg Nothing $ GetNodeStats self
757 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
759-- | Get information about the specified process
760getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
762 let them = processNodeId pid in do
764 dest <- mkNode them us
765 sendCtrlMsg dest $ GetInfo pid
767 match (\(p :: ProcessInfo) -> return $ Just p)
768 , match (\(_ :: ProcessInfoNone) -> return Nothing)
770 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
771 mkNode them us = case them == us of
772 True -> return Nothing
773 _ -> return $ Just them
775--------------------------------------------------------------------------------
776-- Monitoring and linking --
777--------------------------------------------------------------------------------
779-- | Link to a remote process (asynchronous)
781-- When process A links to process B (that is, process A calls
782-- @link pidB@) then an asynchronous exception will be thrown to process A
783-- when process B terminates (normally or abnormally), or when process A gets
784-- disconnected from process B. Although it is /technically/ possible to catch
785-- these exceptions, chances are if you find yourself trying to do so you should
786-- probably be using 'monitor' rather than 'link'. In particular, code such as
788-- > link pidB -- Link to process B
789-- > expect -- Wait for a message from process B
790-- > unlink pidB -- Unlink again
792-- doesn't quite do what one might expect: if process B sends a message to
793-- process A, and /subsequently terminates/, then process A might or might not
794-- be terminated too, depending on whether the exception is thrown before or
795-- after the 'unlink' (i.e., this code has a race condition).
797-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
798-- call to 'link' has no effect.
800-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
801-- Linking makes no distinction between normal and abnormal termination of
802-- the remote process.
803link :: ProcessId -> Process ()
804link = sendCtrlMsg Nothing . Link . ProcessIdentifier
806-- | Monitor another process (asynchronous)
808-- When process A monitors process B (that is, process A calls
809-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
810-- when process B terminates (normally or abnormally), or when process A gets
811-- disconnected from process B. You receive this message like any other (using
812-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
813-- 'DiedDisconnect', etc.).
815-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
816-- multiple monitors are set up, multiple notifications will be delivered
817-- and monitors can be disabled individually using 'unmonitor'.
818monitor :: ProcessId -> Process MonitorRef
819monitor = monitor' . ProcessIdentifier
821-- | Establishes temporary monitoring of another process.
823-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
824-- of @code@. Note: although monitoring is no longer active when
825-- @withMonitor@ returns, there might still be unreceived monitor
826-- messages in the queue.
828withMonitor :: ProcessId -> Process a -> Process a
829withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
830 -- unmonitor blocks waiting for the response, so there's a possibility
831 -- that an exception might interrupt withMonitor before the unmonitor
832 -- has completed. I think that's better than making the unmonitor
837-- This is synchronous in the sense that once it returns you are guaranteed
838-- that no exception will be raised if the remote process dies. However, it is
839-- asynchronous in the sense that we do not wait for a response from the remote
841unlink :: ProcessId -> Process ()
844 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
848-- | Remove a node link
850-- This has the same synchronous/asynchronous nature as 'unlink'.
851unlinkNode :: NodeId -> Process ()
854 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
858-- | Remove a channel (send port) link
860-- This has the same synchronous/asynchronous nature as 'unlink'.
861unlinkPort :: SendPort a -> Process ()
863 unlinkPortAsync sport
864 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
870-- This has the same synchronous/asynchronous nature as 'unlink'.
871unmonitor :: MonitorRef -> Process ()
874 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
878--------------------------------------------------------------------------------
879-- Exception handling --
880--------------------------------------------------------------------------------
882-- | Lift 'Control.Exception.catch'
883catch :: Exception e => Process a -> (e -> Process a) -> Process a
886 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
888-- | Lift 'Control.Exception.try'
889try :: Exception e => Process a -> Process (Either e a)
892 liftIO $ Ex.try (runLocalProcess lproc p)
894-- | Lift 'Control.Exception.mask'
895mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
898 liftIO $ Ex.mask $ \restore ->
899 runLocalProcess lproc (p (liftRestore restore))
901 liftRestore :: (forall a. IO a -> IO a)
902 -> (forall a. Process a -> Process a)
903 liftRestore restoreIO = \p2 -> do
905 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
907-- | Lift 'Control.Exception.mask_'
908mask_ :: Process a -> Process a
911 liftIO $ Ex.mask_ $ runLocalProcess lproc p
913-- | Lift 'Control.Exception.onException'
914onException :: Process a -> Process b -> Process a
915onException p what = p `catch` \e -> do _ <- what
916 liftIO $ throwIO (e :: SomeException)
918-- | Lift 'Control.Exception.bracket'
919bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
920bracket before after thing =
921 mask $ \restore -> do
923 r <- restore (thing a) `onException` after a
927-- | Lift 'Control.Exception.bracket_'
928bracket_ :: Process a -> Process b -> Process c -> Process c
929bracket_ before after thing = bracket before (const after) (const thing)
931-- | Lift 'Control.Exception.finally'
932finally :: Process a -> Process b -> Process a
933finally a sequel = bracket_ (return ()) sequel a
935-- | You need this when using 'catches'
936data Handler a = forall e . Exception e => Handler (e -> Process a)
938instance Functor Handler where
939 fmap f (Handler h) = Handler (fmap f . h)
941-- | Lift 'Control.Exception.catches'
942catches :: Process a -> [Handler a] -> Process a
943catches proc handlers = proc `catch` catchesHandler handlers
945catchesHandler :: [Handler a] -> SomeException -> Process a
946catchesHandler handlers e = foldr tryHandler (throw e) handlers
947 where tryHandler (Handler handler) res
948 = case fromException e of
949 Just e' -> handler e'
952--------------------------------------------------------------------------------
954--------------------------------------------------------------------------------
956-- | Like 'expect' but with a timeout
957expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
958expectTimeout n = receiveTimeout n [match return]
960-- | Asynchronous version of 'spawn'
962-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
963spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
964spawnAsync nid proc = do
965 spawnRef <- getSpawnRef
966 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
969-- | Monitor a node (asynchronous)
970monitorNode :: NodeId -> Process MonitorRef
972 monitor' . NodeIdentifier
974-- | Monitor a typed channel (asynchronous)
975monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
976monitorPort (SendPort cid) =
977 monitor' (SendPortIdentifier cid)
979-- | Remove a monitor (asynchronous)
980unmonitorAsync :: MonitorRef -> Process ()
982 sendCtrlMsg Nothing . Unmonitor
984-- | Link to a node (asynchronous)
985linkNode :: NodeId -> Process ()
986linkNode = link' . NodeIdentifier
988-- | Link to a channel (asynchronous)
989linkPort :: SendPort a -> Process ()
990linkPort (SendPort cid) =
991 link' (SendPortIdentifier cid)
993-- | Remove a process link (asynchronous)
994unlinkAsync :: ProcessId -> Process ()
996 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
998-- | Remove a node link (asynchronous)
999unlinkNodeAsync :: NodeId -> Process ()
1001 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1003-- | Remove a channel (send port) link (asynchronous)
1004unlinkPortAsync :: SendPort a -> Process ()
1005unlinkPortAsync (SendPort cid) =
1006 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1008--------------------------------------------------------------------------------
1010--------------------------------------------------------------------------------
1014-- @say message@ sends a message (time, pid of the current process, message)
1015-- to the process registered as 'logger'. By default, this process simply
1016-- sends the string to 'stderr'. Individual Cloud Haskell backends might
1017-- replace this with a different logger process, however.
1018say :: String -> Process ()
1020 now <- liftIO getCurrentTime
1022 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
1024--------------------------------------------------------------------------------
1026--------------------------------------------------------------------------------
1028-- | Register a process with the local registry (asynchronous).
1029-- This version will wait until a response is gotten from the
1030-- management process. The name must not already be registered.
1031-- The process need not be on this node.
1032-- A bad registration will result in a 'ProcessRegistrationException'
1034-- The process to be registered does not have to be local itself.
1035register :: String -> ProcessId -> Process ()
1036register = registerImpl False
1038-- | Like 'register', but will replace an existing registration.
1039-- The name must already be registered.
1040reregister :: String -> ProcessId -> Process ()
1041reregister = registerImpl True
1043registerImpl :: Bool -> String -> ProcessId -> Process ()
1044registerImpl force label pid = do
1045 mynid <- getSelfNode
1046 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1047 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
1048 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
1051-- | Register a process with a remote registry (asynchronous).
1053-- The process to be registered does not have to live on the same remote node.
1054-- Reply wil come in the form of a 'RegisterReply' message
1056-- See comments in 'whereisRemoteAsync'
1057registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1058registerRemoteAsync nid label pid =
1059 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
1061reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1062reregisterRemoteAsync nid label pid =
1063 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1065-- | Remove a process from the local registry (asynchronous).
1066-- This version will wait until a response is gotten from the
1067-- management process. The name must already be registered.
1068unregister :: String -> Process ()
1069unregister label = do
1070 mynid <- getSelfNode
1071 sendCtrlMsg Nothing (Register label mynid Nothing False)
1072 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
1073 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
1076-- | Deal with the result from an attempted registration or unregistration
1077-- by throwing an exception if necessary
1078handleRegistrationReply :: String -> Bool -> Process ()
1079handleRegistrationReply label ok =
1081 liftIO $ throwIO $ ProcessRegistrationException label
1083-- | Remove a process from a remote registry (asynchronous).
1085-- Reply wil come in the form of a 'RegisterReply' message
1087-- See comments in 'whereisRemoteAsync'
1088unregisterRemoteAsync :: NodeId -> String -> Process ()
1089unregisterRemoteAsync nid label =
1090 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1092-- | Query the local process registry
1093whereis :: String -> Process (Maybe ProcessId)
1095 sendCtrlMsg Nothing (WhereIs label)
1096 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1097 (\(WhereIsReply _ mPid) -> return mPid)
1100-- | Query a remote process registry (asynchronous)
1102-- Reply will come in the form of a 'WhereIsReply' message.
1104-- There is currently no synchronous version of 'whereisRemoteAsync': if
1105-- you implement one yourself, be sure to take into account that the remote
1106-- node might die or get disconnect before it can respond (i.e. you should
1107-- use 'monitorNode' and take appropriate action when you receive a
1108-- 'NodeMonitorNotification').
1109whereisRemoteAsync :: NodeId -> String -> Process ()
1110whereisRemoteAsync nid label =
1111 sendCtrlMsg (Just nid) (WhereIs label)
1113-- | Named send to a process in the local registry (asynchronous)
1114nsend :: Serializable a => String -> a -> Process ()
1116 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1118-- | Named send to a process in the local registry (asynchronous).
1119-- This function makes /no/ attempt to serialize and (in the case when the
1120-- destination process resides on the same local node) therefore ensure that
1121-- the payload is fully evaluated before it is delivered.
1122unsafeNSend :: Serializable a => String -> a -> Process ()
1123unsafeNSend = Unsafe.nsend
1125-- | Named send to a process in a remote registry (asynchronous)
1126nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1127nsendRemote nid label msg =
1128 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1130--------------------------------------------------------------------------------
1132--------------------------------------------------------------------------------
1134-- | Resolve a static value
1135unStatic :: Typeable a => Static a -> Process a
1137 rtable <- remoteTable . processNode <$> ask
1138 case Static.unstatic rtable static of
1139 Left err -> fail $ "Could not resolve static value: " ++ err
1142-- | Resolve a closure
1143unClosure :: Typeable a => Closure a -> Process a
1144unClosure closure = do
1145 rtable <- remoteTable . processNode <$> ask
1146 case Static.unclosure rtable closure of
1147 Left err -> fail $ "Could not resolve closure: " ++ err
1150--------------------------------------------------------------------------------
1152--------------------------------------------------------------------------------
1154-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1155-- message passing. However, when network connections get disrupted this
1156-- illusion cannot always be maintained. Once a network connection breaks (even
1157-- temporarily) no further communication on that connection will be possible.
1158-- For example, if process A sends a message to process B, and A is then
1159-- notified (by monitor notification) that it got disconnected from B, A will
1160-- not be able to send any further messages to B, /unless/ A explicitly
1161-- indicates that it is acceptable to attempt to reconnect to B using the
1162-- Cloud Haskell 'reconnect' primitive.
1164-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1165-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1166-- then receives a monitor notification that its connection to B has been lost,
1167-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1168-- and m3 but not m2.
1170-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1171-- to attempt to reconnect on the next send/. In particular, if no further
1172-- communication attempts are made to B then A can use reconnect to clean up
1173-- its connection to B.
1174reconnect :: ProcessId -> Process ()
1177 node <- processNode <$> ask
1178 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1180-- | Reconnect to a sendport. See 'reconnect' for more information.
1181reconnectPort :: SendPort a -> Process ()
1182reconnectPort them = do
1184 node <- processNode <$> ask
1185 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1187--------------------------------------------------------------------------------
1188-- Auxiliary functions --
1189--------------------------------------------------------------------------------
1191sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1193 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1195sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1196sendChanLocal spId msg =
1197 -- we *must* fully serialize/encode the message here, because
1198 -- attempting to use `unsafeCoerce' in the node controller
1199 -- won't work since we know nothing about the required type
1200 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1202getMonitorRefFor :: Identifier -> Process MonitorRef
1203getMonitorRefFor ident = do
1205 liftIO $ modifyMVar (processState proc) $ \st -> do
1206 let counter = st ^. monitorCounter
1207 return ( monitorCounter ^: (+ 1) $ st
1208 , MonitorRef ident counter
1211getSpawnRef :: Process SpawnRef
1214 liftIO $ modifyMVar (processState proc) $ \st -> do
1215 let counter = st ^. spawnCounter
1216 return ( spawnCounter ^: (+ 1) $ st
1220-- | Monitor a process/node/channel
1221monitor' :: Identifier -> Process MonitorRef
1223 monitorRef <- getMonitorRefFor ident
1224 sendCtrlMsg Nothing $ Monitor monitorRef
1227-- | Link to a process/node/channel
1228link' :: Identifier -> Process ()
1229link' = sendCtrlMsg Nothing . Link
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