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
23 -- * Unsafe messaging variants
27 -- * Advanced messaging
52 -- * Process management
54 , ProcessTerminationException(..)
60 -- keep the exception constructor hidden, so that handling exit
61 -- reasons /must/ take place via the 'catchExit' family of primitives
62 , ProcessExitException()
70 -- * Monitoring and linking
85 , reregisterRemoteAsync
86 , unregisterRemoteAsync
92 -- * Exception handling
116 -- * Internal Exports
120#if ! MIN_VERSION_base(4,6,0)
121import Prelude hiding (catch)
124import Data.Binary (decode)
125import Data.Time.Clock (getCurrentTime)
126import Data.Time.Format (formatTime)
127#if MIN_VERSION_time(1,5,0)
128import Data.Time.Format (defaultTimeLocale)
130import System.Locale (defaultTimeLocale)
132import System.Timeout (timeout)
133import Control.Monad (when)
134import Control.Monad.Reader (ask)
135import Control.Monad.IO.Class (liftIO)
136import Control.Applicative ((<$>))
137import Control.Exception (Exception(..), throw, throwIO, SomeException)
138import qualified Control.Exception as Ex (catch, mask, mask_, try)
139import Control.Distributed.Process.Internal.StrictMVar
144import Control.Concurrent.STM
153import Control.Distributed.Process.Internal.CQueue
158import Control.Distributed.Process.Serializable (Serializable, fingerprint)
159import Data.Accessor ((^.), (^:), (^=))
160import Control.Distributed.Static
164import Data.Rank1Typeable (Typeable)
165import qualified Control.Distributed.Static as Static (unstatic, unclosure)
166import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
167import Control.Distributed.Process.Internal.Types
177 , NodeMonitorNotification(..)
187 , ProcessExitException(..)
190 , DidUnlinkProcess(..)
195 , ProcessRegistrationException(..)
197 , ProcessInfoNone(..)
201 , createUnencodedMessage
203 , ImplicitReconnect( NoImplicitReconnect)
208import Control.Distributed.Process.Internal.Messaging
215import Control.Distributed.Process.Management.Internal.Types
218import Control.Distributed.Process.Management.Internal.Trace.Types
221import Control.Distributed.Process.Internal.WeakTQueue
229--------------------------------------------------------------------------------
231--------------------------------------------------------------------------------
234send :: Serializable a => ProcessId -> a -> Process ()
235-- This requires a lookup on every send. If we want to avoid that we need to
236-- modify serializable to allow for stateful (IO) deserialization.
239 let us = processId proc
240 node = processNode proc
241 nodeId = localNodeId node
242 destNode = (processNodeId them) in do
243 case destNode == nodeId of
244 True -> sendLocal them msg
245 False -> liftIO $ sendMessage (processNode proc)
246 (ProcessIdentifier (processId proc))
247 (ProcessIdentifier them)
250 -- We do not fire the trace event until after the sending is complete;
251 -- In the remote case, 'sendMessage' can block in the networking stack.
252 liftIO $ traceEvent (localEventBus node)
253 (MxSent them us (createUnencodedMessage msg))
255-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
256-- and (in the case when the destination process resides on the same local
257-- node) therefore ensure that the payload is fully evaluated before it is
259unsafeSend :: Serializable a => ProcessId -> a -> Process ()
260unsafeSend = Unsafe.send
262-- | Send a message unreliably.
264-- Unlike 'send', this function is insensitive to 'reconnect'. It will
265-- try to send the message regardless of the history of connection failures
268-- Message passing with 'usend' is ordered for a given sender and receiver
269-- if the messages arrive at all.
271usend :: Serializable a => ProcessId -> a -> Process ()
274 let there = processNodeId them
276 then sendLocal them msg
277 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
280-- | Wait for a message of a specific type
281expect :: forall a. Serializable a => Process a
282expect = receiveWait [match return]
284--------------------------------------------------------------------------------
286--------------------------------------------------------------------------------
288-- | Create a new typed channel
289newChan :: Serializable a => Process (SendPort a, ReceivePort a)
292 liftIO . modifyMVar (processState proc) $ \st -> do
293 let lcid = st ^. channelCounter
294 let cid = SendPortId { sendPortProcessId = processId proc
295 , sendPortLocalId = lcid
297 let sport = SendPort cid
298 chan <- liftIO newTQueueIO
299 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
300 let rport = ReceivePort $ readTQueue chan
301 let tch = TypedChannel chan'
302 return ( (channelCounter ^: (+ 1))
303 . (typedChannelWithId lcid ^= Just tch)
308 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
309 finalizer st lcid = modifyMVar_ st $
310 return . (typedChannelWithId lcid ^= Nothing)
312-- | Send a message on a typed channel
313sendChan :: Serializable a => SendPort a -> a -> Process ()
314sendChan (SendPort cid) msg = do
316 let node = localNodeId (processNode proc)
317 destNode = processNodeId (sendPortProcessId cid) in do
318 case destNode == node of
319 True -> sendChanLocal cid msg
321 liftIO $ sendBinary (processNode proc)
322 (ProcessIdentifier (processId proc))
323 (SendPortIdentifier cid)
327-- | Send a message on a typed channel. This function makes /no/ attempt to
328-- serialize and (in the case when the @ReceivePort@ resides on the same local
329-- node) therefore ensure that the payload is fully evaluated before it is
331unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
332unsafeSendChan = Unsafe.sendChan
334-- | Wait for a message on a typed channel
335receiveChan :: Serializable a => ReceivePort a -> Process a
336receiveChan = liftIO . atomically . receiveSTM
338-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
339-- non-blocking check for a message.
340receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
341receiveChanTimeout 0 ch = liftIO . atomically $
342 (Just <$> receiveSTM ch) `orElse` return Nothing
343receiveChanTimeout n ch = liftIO . timeout n . atomically $
346-- | Merge a list of typed channels.
348-- The result port is left-biased: if there are messages available on more
349-- than one port, the first available message is returned.
350mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
351mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
353-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
355mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
356mergePortsRR = \ps -> do
357 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
358 return $ ReceivePort (rr psVar)
362 rotate (x:xs) = xs ++ [x]
364 rr :: TVar [STM a] -> STM a
367 a <- foldr1 orElse ps
368 writeTVar psVar (rotate ps)
371--------------------------------------------------------------------------------
372-- Advanced messaging --
373--------------------------------------------------------------------------------
375-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
376newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
378-- | Test the matches in order against each message in the queue
379receiveWait :: [Match b] -> Process b
381 queue <- processQueue <$> ask
382 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
385-- | Like 'receiveWait' but with a timeout.
387-- If the timeout is zero do a non-blocking check for matching messages. A
388-- non-zero timeout is applied only when waiting for incoming messages (that is,
389-- /after/ we have checked the messages that are already in the mailbox).
390receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
391receiveTimeout t ms = do
392 queue <- processQueue <$> ask
393 let blockSpec = if t == 0 then NonBlocking else Timeout t
394 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
396 Nothing -> return Nothing
397 Just proc -> Just <$> proc
399-- | Match on a typed channel
400matchChan :: ReceivePort a -> (a -> Process b) -> Match b
401matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
403-- | Match on an arbitrary STM action.
405-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
406-- actions with checks against our process' mailbox and/or any typed channel
407-- @ReceivePort@s we may hold.
409-- This allows us to process multiple input streams /along with our mailbox/,
410-- in just the same way that 'matchChan' supports checking both the mailbox
411-- /and/ an arbitrary set of typed channels in one atomic transaction.
413-- Note there are no ordering guarnatees with respect to these disparate input
416matchSTM :: STM a -> (a -> Process b) -> Match b
417matchSTM stm fn = Match $ MatchChan (fmap fn stm)
419-- | Match against any message of the right type
420match :: forall a b. Serializable a => (a -> Process b) -> Match b
421match = matchIf (const True)
423-- | Match against any message of the right type that satisfies a predicate
424matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
425matchIf c p = Match $ MatchMsg $ \msg ->
426 case messageFingerprint msg == fingerprint (undefined :: a) of
429 (UnencodedMessage _ m) ->
430 let m' = unsafeCoerce m :: a in
434 (EncodedMessage _ _) ->
435 if (c decoded) then Just (p decoded) else Nothing
438 -- Make sure the value is fully decoded so that we don't hang to
439 -- bytestrings when the process calling 'matchIf' doesn't process
440 -- the values immediately
441 !decoded = decode (messageEncoding msg)
443-- | Match against any message, regardless of the underlying (contained) type
444matchMessage :: (Message -> Process Message) -> Match Message
445matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
447-- | Match against any message (regardless of underlying type) that satisfies a predicate
448matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
449matchMessageIf c p = Match $ MatchMsg $ \msg ->
454-- | Forward a raw 'Message' to the given 'ProcessId'.
455forward :: Message -> ProcessId -> Process ()
458 let node = processNode proc
460 nid = localNodeId node
461 destNode = (processNodeId them) in do
462 case destNode == nid of
463 True -> sendCtrlMsg Nothing (LocalSend them msg)
464 False -> liftIO $ sendPayload (processNode proc)
465 (ProcessIdentifier (processId proc))
466 (ProcessIdentifier them)
468 (messageToPayload msg)
469 -- We do not fire the trace event until after the sending is complete;
470 -- In the remote case, 'sendMessage' can block in the networking stack.
471 liftIO $ traceEvent (localEventBus node)
475-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
476-- 'Serializable' - like the datum they contain - but also note, deserialising
477-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
478-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
483-- send self (wrapMessage "blah")
484-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
485-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
486-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
489wrapMessage :: Serializable a => a -> Message
490wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
492-- | This is the /unsafe/ variant of 'wrapMessage'. See
493-- "Control.Distributed.Process.UnsafePrimitives" for details.
494unsafeWrapMessage :: Serializable a => a -> Message
495unsafeWrapMessage = Unsafe.wrapMessage
497-- [note Serializable UnencodedMessage]
498-- if we attempt to serialise an UnencodedMessage, it will be converted to an
499-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
500-- Primitives.hs for the gory details
502-- | Attempt to unwrap a raw 'Message'.
503-- If the type of the decoded message payload matches the expected type, the
504-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
507-- This expression, for example, will evaluate to @Nothing@
508-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
510-- Whereas this expression, will yield @Just "foo"@
511-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
513unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
515 case messageFingerprint msg == fingerprint (undefined :: a) of
516 False -> return Nothing :: m (Maybe a)
518 (UnencodedMessage _ ms) ->
519 let ms' = unsafeCoerce ms :: a
521 (EncodedMessage _ _) ->
522 return (Just (decoded))
524 decoded :: a -- note [decoding]
525 !decoded = decode (messageEncoding msg)
527-- | Attempt to handle a raw 'Message'.
528-- If the type of the message matches the type of the first argument to
529-- the supplied expression, then the message will be decoded and the expression
530-- evaluated against its value. If this runtime type checking fails however,
531-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
532-- evaluation proceeds, the resulting value with be wrapped with @Just@.
534-- Intended for use in `catchesExit` and `matchAny` primitives.
536handleMessage :: forall m a b. (Monad m, Serializable a)
537 => Message -> (a -> m b) -> m (Maybe b)
538handleMessage msg proc = handleMessageIf msg (const True) proc
540-- | Conditionally handle a raw 'Message'.
541-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
542-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
543-- for further information about runtime type checking.
545handleMessageIf :: forall m a b . (Monad m, Serializable a)
550handleMessageIf msg c proc = do
551 case messageFingerprint msg == fingerprint (undefined :: a) of
552 False -> return Nothing :: m (Maybe b)
554 (UnencodedMessage _ ms) ->
555 let ms' = unsafeCoerce ms :: a in
557 True -> do { r <- proc ms'; return (Just r) }
558 False -> return Nothing :: m (Maybe b)
559 (EncodedMessage _ _) ->
561 True -> do { r <- proc (decoded :: a); return (Just r) }
562 False -> return Nothing :: m (Maybe b)
564 decoded :: a -- note [decoding]
565 !decoded = decode (messageEncoding msg)
567-- | As 'handleMessage' but ignores result, which is useful if you don't
568-- care whether or not the handler succeeded.
570handleMessage_ :: forall m a . (Monad m, Serializable a)
571 => Message -> (a -> m ()) -> m ()
572handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
574-- | Conditional version of 'handleMessage_'.
576handleMessageIf_ :: forall m a . (Monad m, Serializable a)
581handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
583-- | Match against an arbitrary message. 'matchAny' removes the first available
584-- message from the process mailbox. To handle arbitrary /raw/ messages once
585-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
587matchAny :: forall b. (Message -> Process b) -> Match b
588matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
590-- | Match against an arbitrary message. Intended for use with 'handleMessage'
591-- and 'unwrapMessage', this function /only/ removes a message from the process
592-- mailbox, /if/ the supplied condition matches. The success (or failure) of
593-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
594-- here, i.e., if the condition evaluates to @True@ then the message will
595-- be removed from the process mailbox and decoded, but that does /not/
596-- guarantee that an expression passed to @handleMessage@ will pass the
597-- runtime type checks and therefore be evaluated.
599matchAnyIf :: forall a b. (Serializable a)
601 -> (Message -> Process b)
603matchAnyIf c p = Match $ MatchMsg $ \msg ->
604 case messageFingerprint msg == fingerprint (undefined :: a) of
605 True | check -> Just (p msg)
610 (EncodedMessage _ _) -> c decoded
611 (UnencodedMessage _ m') -> c (unsafeCoerce m')
613 decoded :: a -- note [decoding]
614 !decoded = decode (messageEncoding msg)
618For an EncodedMessage, we need to ensure the value is fully decoded so that
619we don't hang to bytestrings if the calling process doesn't evaluate
620immediately. For UnencodedMessage we know (because the fingerprint comparison
621succeeds) that unsafeCoerce will not fail.
624-- | Remove any message from the queue
625matchUnknown :: Process b -> Match b
626matchUnknown p = Match $ MatchMsg (const (Just p))
628-- | Receives messages and forwards them to @pid@ if @p msg == True@.
629delegate :: ProcessId -> (Message -> Bool) -> Process ()
632 matchAny (\m -> case (p m) of
633 True -> forward m pid
638-- | A straight relay that forwards all messages to the supplied pid.
639relay :: ProcessId -> Process ()
640relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
642-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
643-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
644-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
645-- fails, no action is taken and the proxy process will continue running.
646proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
650 next <- handleMessage m proc
652 Just True -> forward m pid
653 Just False -> return () -- explicitly ignored
654 Nothing -> return ()) -- un-routable
658--------------------------------------------------------------------------------
659-- Process management --
660--------------------------------------------------------------------------------
662-- | Thrown by 'terminate'
663data ProcessTerminationException = ProcessTerminationException
664 deriving (Show, Typeable)
666instance Exception ProcessTerminationException
668-- | Terminate immediately (throws a ProcessTerminationException)
669terminate :: Process a
670terminate = liftIO $ throwIO ProcessTerminationException
673-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
674die :: Serializable a => a -> Process b
676-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
677-- could be decoded by a handler passed to 'catchExit', re-thrown or even
678-- passed to another process via the tracing mechanism.
680 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
682-- | Forceful request to kill a process. Where 'exit' provides an exception
683-- that can be caught and handled, 'kill' throws an unexposed exception type
684-- which cannot be handled explicitly (by type).
685kill :: ProcessId -> String -> Process ()
686-- NOTE: We send the message to our local node controller, which will then
687-- forward it to a remote node controller (if applicable). Sending it directly
688-- to a remote node controller means that that the message may overtake a
689-- 'monitor' or 'link' request.
690kill them reason = sendCtrlMsg Nothing (Kill them reason)
692-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
693-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
694-- manner can be handled using the 'catchExit' family of functions.
695exit :: Serializable a => ProcessId -> a -> Process ()
696-- NOTE: We send the message to our local node controller, which will then
697-- forward it to a remote node controller (if applicable). Sending it directly
698-- to a remote node controller means that that the message may overtake a
699-- 'monitor' or 'link' request.
700exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
702-- | Catches 'ProcessExitException'. The handler will not be applied unless its
703-- type matches the encoded data stored in the exception (see the /reason/
704-- argument given to the 'exit' primitive). If the handler cannot be applied,
705-- the exception will be re-thrown.
707-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
708-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
709catchExit :: forall a b . (Show a, Serializable a)
711 -> (ProcessId -> a -> Process b)
713catchExit act exitHandler = catch act handleExit
715 handleExit ex@(ProcessExitException from msg) =
716 if messageFingerprint msg == fingerprint (undefined :: a)
717 then exitHandler from decoded
718 else liftIO $ throwIO ex
721 -- Make sure the value is fully decoded so that we don't hang to
722 -- bytestrings if the caller doesn't use the value immediately
723 !decoded = decode (messageEncoding msg)
725-- | Lift 'Control.Exception.catches' (almost).
727-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
728-- message, a handler must accept inputs of the expected type. In order to
729-- handle a list of potentially different handlers (and therefore input types),
730-- a handler passed to 'catchesExit' must accept 'Message' and return
731-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
733-- See 'maybeHandleMessage' and 'Message' for more details.
734catchesExit :: Process b
735 -> [(ProcessId -> Message -> (Process (Maybe b)))]
737catchesExit act handlers = catch act ((flip handleExit) handlers)
739 handleExit :: ProcessExitException
740 -> [(ProcessId -> Message -> Process (Maybe b))]
742 handleExit ex [] = liftIO $ throwIO ex
743 handleExit ex@(ProcessExitException from msg) (h:hs) = do
746 Nothing -> handleExit ex hs
749-- | Our own process ID
750getSelfPid :: Process ProcessId
751getSelfPid = processId <$> ask
753-- | Get the node ID of our local node
754getSelfNode :: Process NodeId
755getSelfNode = localNodeId . processNode <$> ask
758-- | Get statistics about the specified node
759getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
761 selfNode <- getSelfNode
763 then Right `fmap` getLocalNodeStats -- optimisation
764 else getNodeStatsRemote selfNode
766 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
767 getNodeStatsRemote selfNode = do
768 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
769 bracket (monitorNode nid) unmonitor $ \mRef ->
770 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
771 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
772 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
775-- | Get statistics about our local node
776getLocalNodeStats :: Process NodeStats
777getLocalNodeStats = do
779 sendCtrlMsg Nothing $ GetNodeStats self
780 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
782-- | Get information about the specified process
783getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
785 let them = processNodeId pid in do
787 dest <- mkNode them us
788 sendCtrlMsg dest $ GetInfo pid
790 match (\(p :: ProcessInfo) -> return $ Just p)
791 , match (\(_ :: ProcessInfoNone) -> return Nothing)
793 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
794 mkNode them us = case them == us of
795 True -> return Nothing
796 _ -> return $ Just them
798--------------------------------------------------------------------------------
799-- Monitoring and linking --
800--------------------------------------------------------------------------------
802-- | Link to a remote process (asynchronous)
804-- When process A links to process B (that is, process A calls
805-- @link pidB@) then an asynchronous exception will be thrown to process A
806-- when process B terminates (normally or abnormally), or when process A gets
807-- disconnected from process B. Although it is /technically/ possible to catch
808-- these exceptions, chances are if you find yourself trying to do so you should
809-- probably be using 'monitor' rather than 'link'. In particular, code such as
811-- > link pidB -- Link to process B
812-- > expect -- Wait for a message from process B
813-- > unlink pidB -- Unlink again
815-- doesn't quite do what one might expect: if process B sends a message to
816-- process A, and /subsequently terminates/, then process A might or might not
817-- be terminated too, depending on whether the exception is thrown before or
818-- after the 'unlink' (i.e., this code has a race condition).
820-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
821-- call to 'link' has no effect.
823-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
824-- Linking makes no distinction between normal and abnormal termination of
825-- the remote process.
826link :: ProcessId -> Process ()
827link = sendCtrlMsg Nothing . Link . ProcessIdentifier
829-- | Monitor another process (asynchronous)
831-- When process A monitors process B (that is, process A calls
832-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
833-- when process B terminates (normally or abnormally), or when process A gets
834-- disconnected from process B. You receive this message like any other (using
835-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
836-- 'DiedDisconnect', etc.).
838-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
839-- multiple monitors are set up, multiple notifications will be delivered
840-- and monitors can be disabled individually using 'unmonitor'.
841monitor :: ProcessId -> Process MonitorRef
842monitor = monitor' . ProcessIdentifier
844-- | Establishes temporary monitoring of another process.
846-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
847-- of @code@. Note: although monitoring is no longer active when
848-- @withMonitor@ returns, there might still be unreceived monitor
849-- messages in the queue.
851withMonitor :: ProcessId -> Process a -> Process a
852withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
853 -- unmonitor blocks waiting for the response, so there's a possibility
854 -- that an exception might interrupt withMonitor before the unmonitor
855 -- has completed. I think that's better than making the unmonitor
860-- This is synchronous in the sense that once it returns you are guaranteed
861-- that no exception will be raised if the remote process dies. However, it is
862-- asynchronous in the sense that we do not wait for a response from the remote
864unlink :: ProcessId -> Process ()
867 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
871-- | Remove a node link
873-- This has the same synchronous/asynchronous nature as 'unlink'.
874unlinkNode :: NodeId -> Process ()
877 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
881-- | Remove a channel (send port) link
883-- This has the same synchronous/asynchronous nature as 'unlink'.
884unlinkPort :: SendPort a -> Process ()
886 unlinkPortAsync sport
887 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
893-- This has the same synchronous/asynchronous nature as 'unlink'.
894unmonitor :: MonitorRef -> Process ()
897 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
901--------------------------------------------------------------------------------
902-- Exception handling --
903--------------------------------------------------------------------------------
905-- | Lift 'Control.Exception.catch'
906catch :: Exception e => Process a -> (e -> Process a) -> Process a
909 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
911-- | Lift 'Control.Exception.try'
912try :: Exception e => Process a -> Process (Either e a)
915 liftIO $ Ex.try (runLocalProcess lproc p)
917-- | Lift 'Control.Exception.mask'
918mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
921 liftIO $ Ex.mask $ \restore ->
922 runLocalProcess lproc (p (liftRestore restore))
924 liftRestore :: (forall a. IO a -> IO a)
925 -> (forall a. Process a -> Process a)
926 liftRestore restoreIO = \p2 -> do
928 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
930-- | Lift 'Control.Exception.mask_'
931mask_ :: Process a -> Process a
934 liftIO $ Ex.mask_ $ runLocalProcess lproc p
936-- | Lift 'Control.Exception.onException'
937onException :: Process a -> Process b -> Process a
938onException p what = p `catch` \e -> do _ <- what
939 liftIO $ throwIO (e :: SomeException)
941-- | Lift 'Control.Exception.bracket'
942bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
943bracket before after thing =
944 mask $ \restore -> do
946 r <- restore (thing a) `onException` after a
950-- | Lift 'Control.Exception.bracket_'
951bracket_ :: Process a -> Process b -> Process c -> Process c
952bracket_ before after thing = bracket before (const after) (const thing)
954-- | Lift 'Control.Exception.finally'
955finally :: Process a -> Process b -> Process a
956finally a sequel = bracket_ (return ()) sequel a
958-- | You need this when using 'catches'
959data Handler a = forall e . Exception e => Handler (e -> Process a)
961instance Functor Handler where
962 fmap f (Handler h) = Handler (fmap f . h)
964-- | Lift 'Control.Exception.catches'
965catches :: Process a -> [Handler a] -> Process a
966catches proc handlers = proc `catch` catchesHandler handlers
968catchesHandler :: [Handler a] -> SomeException -> Process a
969catchesHandler handlers e = foldr tryHandler (throw e) handlers
970 where tryHandler (Handler handler) res
971 = case fromException e of
972 Just e' -> handler e'
975--------------------------------------------------------------------------------
977--------------------------------------------------------------------------------
979-- | Like 'expect' but with a timeout
980expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
981expectTimeout n = receiveTimeout n [match return]
983-- | Asynchronous version of 'spawn'
985-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
986spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
987spawnAsync nid proc = do
988 spawnRef <- getSpawnRef
991 then sendCtrlMsg Nothing $ Spawn proc spawnRef
992 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
995-- | Monitor a node (asynchronous)
996monitorNode :: NodeId -> Process MonitorRef
998 monitor' . NodeIdentifier
1000-- | Monitor a typed channel (asynchronous)
1001monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1002monitorPort (SendPort cid) =
1003 monitor' (SendPortIdentifier cid)
1005-- | Remove a monitor (asynchronous)
1006unmonitorAsync :: MonitorRef -> Process ()
1008 sendCtrlMsg Nothing . Unmonitor
1010-- | Link to a node (asynchronous)
1011linkNode :: NodeId -> Process ()
1012linkNode = link' . NodeIdentifier
1014-- | Link to a channel (asynchronous)
1015linkPort :: SendPort a -> Process ()
1016linkPort (SendPort cid) =
1017 link' (SendPortIdentifier cid)
1019-- | Remove a process link (asynchronous)
1020unlinkAsync :: ProcessId -> Process ()
1022 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1024-- | Remove a node link (asynchronous)
1025unlinkNodeAsync :: NodeId -> Process ()
1027 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1029-- | Remove a channel (send port) link (asynchronous)
1030unlinkPortAsync :: SendPort a -> Process ()
1031unlinkPortAsync (SendPort cid) =
1032 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1034--------------------------------------------------------------------------------
1036--------------------------------------------------------------------------------
1040-- @say message@ sends a message (time, pid of the current process, message)
1041-- to the process registered as 'logger'. By default, this process simply
1042-- sends the string to 'stderr'. Individual Cloud Haskell backends might
1043-- replace this with a different logger process, however.
1044say :: String -> Process ()
1046 now <- liftIO getCurrentTime
1048 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
1050--------------------------------------------------------------------------------
1052--------------------------------------------------------------------------------
1054-- | Register a process with the local registry (synchronous). The name must not
1055-- already be registered. The process need not be on this node. A bad
1056-- registration will result in a 'ProcessRegistrationException'
1058-- The process to be registered does not have to be local itself.
1059register :: String -> ProcessId -> Process ()
1060register = registerImpl False
1062-- | Like 'register', but will replace an existing registration.
1063-- The name must already be registered.
1064reregister :: String -> ProcessId -> Process ()
1065reregister = registerImpl True
1067registerImpl :: Bool -> String -> ProcessId -> Process ()
1068registerImpl force label pid = do
1069 mynid <- getSelfNode
1070 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1071 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
1072 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
1075-- | Register a process with a remote registry (asynchronous).
1077-- The process to be registered does not have to live on the same remote node.
1078-- Reply wil come in the form of a 'RegisterReply' message
1080-- See comments in 'whereisRemoteAsync'
1081registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1082registerRemoteAsync nid label pid =
1083 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
1085reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1086reregisterRemoteAsync nid label pid =
1087 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1089-- | Remove a process from the local registry (asynchronous).
1090-- This version will wait until a response is gotten from the
1091-- management process. The name must already be registered.
1092unregister :: String -> Process ()
1093unregister label = do
1094 mynid <- getSelfNode
1095 sendCtrlMsg Nothing (Register label mynid Nothing False)
1096 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
1097 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
1100-- | Deal with the result from an attempted registration or unregistration
1101-- by throwing an exception if necessary
1102handleRegistrationReply :: String -> Bool -> Process ()
1103handleRegistrationReply label ok =
1105 liftIO $ throwIO $ ProcessRegistrationException label
1107-- | Remove a process from a remote registry (asynchronous).
1109-- Reply wil come in the form of a 'RegisterReply' message
1111-- See comments in 'whereisRemoteAsync'
1112unregisterRemoteAsync :: NodeId -> String -> Process ()
1113unregisterRemoteAsync nid label =
1114 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1116-- | Query the local process registry
1117whereis :: String -> Process (Maybe ProcessId)
1119 sendCtrlMsg Nothing (WhereIs label)
1120 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1121 (\(WhereIsReply _ mPid) -> return mPid)
1124-- | Query a remote process registry (asynchronous)
1126-- Reply will come in the form of a 'WhereIsReply' message.
1128-- There is currently no synchronous version of 'whereisRemoteAsync': if
1129-- you implement one yourself, be sure to take into account that the remote
1130-- node might die or get disconnect before it can respond (i.e. you should
1131-- use 'monitorNode' and take appropriate action when you receive a
1132-- 'NodeMonitorNotification').
1133whereisRemoteAsync :: NodeId -> String -> Process ()
1134whereisRemoteAsync nid label =
1135 sendCtrlMsg (Just nid) (WhereIs label)
1137-- | Named send to a process in the local registry (asynchronous)
1138nsend :: Serializable a => String -> a -> Process ()
1140 sendCtrlMsg Nothing (NamedSend label (createMessage msg))
1142-- | Named send to a process in the local registry (asynchronous).
1143-- This function makes /no/ attempt to serialize and (in the case when the
1144-- destination process resides on the same local node) therefore ensure that
1145-- the payload is fully evaluated before it is delivered.
1146unsafeNSend :: Serializable a => String -> a -> Process ()
1147unsafeNSend = Unsafe.nsend
1149-- | Named send to a process in a remote registry (asynchronous)
1150nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1151nsendRemote nid label msg =
1152 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1154--------------------------------------------------------------------------------
1156--------------------------------------------------------------------------------
1158-- | Resolve a static value
1159unStatic :: Typeable a => Static a -> Process a
1161 rtable <- remoteTable . processNode <$> ask
1162 case Static.unstatic rtable static of
1163 Left err -> fail $ "Could not resolve static value: " ++ err
1166-- | Resolve a closure
1167unClosure :: Typeable a => Closure a -> Process a
1168unClosure closure = do
1169 rtable <- remoteTable . processNode <$> ask
1170 case Static.unclosure rtable closure of
1171 Left err -> fail $ "Could not resolve closure: " ++ err
1174--------------------------------------------------------------------------------
1176--------------------------------------------------------------------------------
1178-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1179-- message passing. However, when network connections get disrupted this
1180-- illusion cannot always be maintained. Once a network connection breaks (even
1181-- temporarily) no further communication on that connection will be possible.
1182-- For example, if process A sends a message to process B, and A is then
1183-- notified (by monitor notification) that it got disconnected from B, A will
1184-- not be able to send any further messages to B, /unless/ A explicitly
1185-- indicates that it is acceptable to attempt to reconnect to B using the
1186-- Cloud Haskell 'reconnect' primitive.
1188-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1189-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1190-- then receives a monitor notification that its connection to B has been lost,
1191-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1192-- and m3 but not m2.
1194-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1195-- to attempt to reconnect on the next send/. In particular, if no further
1196-- communication attempts are made to B then A can use reconnect to clean up
1197-- its connection to B.
1198reconnect :: ProcessId -> Process ()
1201 node <- processNode <$> ask
1202 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1204-- | Reconnect to a sendport. See 'reconnect' for more information.
1205reconnectPort :: SendPort a -> Process ()
1206reconnectPort them = do
1208 node <- processNode <$> ask
1209 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1211--------------------------------------------------------------------------------
1212-- Auxiliary functions --
1213--------------------------------------------------------------------------------
1215sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1217 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1219sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1220sendChanLocal spId msg =
1221 -- we *must* fully serialize/encode the message here, because
1222 -- attempting to use `unsafeCoerce' in the node controller
1223 -- won't work since we know nothing about the required type
1224 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1226getMonitorRefFor :: Identifier -> Process MonitorRef
1227getMonitorRefFor ident = do
1229 liftIO $ modifyMVar (processState proc) $ \st -> do
1230 let counter = st ^. monitorCounter
1231 return ( monitorCounter ^: (+ 1) $ st
1232 , MonitorRef ident counter
1235getSpawnRef :: Process SpawnRef
1238 liftIO $ modifyMVar (processState proc) $ \st -> do
1239 let counter = st ^. spawnCounter
1240 return ( spawnCounter ^: (+ 1) $ st
1244-- | Monitor a process/node/channel
1245monitor' :: Identifier -> Process MonitorRef
1247 monitorRef <- getMonitorRefFor ident
1248 sendCtrlMsg Nothing $ Monitor monitorRef
1251-- | Link to a process/node/channel
1252link' :: Identifier -> Process ()
1253link' = 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