2{-# LANGUAGE RankNTypes #-}
3{-# LANGUAGE DeriveDataTypeable #-}
4{-# LANGUAGE DeriveFunctor #-}
5{-# LANGUAGE ScopedTypeVariables #-}
6{-# LANGUAGE ExistentialQuantification #-}
7{-# LANGUAGE BangPatterns #-}
9-- | Cloud Haskell primitives
11-- We define these in a separate module so that we don't have to rely on
12-- the closure combinators
13module Control.Distributed.Process.Internal.Primitives
14 ( -- * Basic messaging
24 -- * Unsafe messaging variants
30 -- * Advanced messaging
56 -- * Process management
58 , ProcessTerminationException(..)
64 -- keep the exception constructor hidden, so that handling exit
65 -- reasons /must/ take place via the 'catchExit' family of primitives
66 , ProcessExitException()
74 -- * Monitoring and linking
89 , reregisterRemoteAsync
90 , unregisterRemoteAsync
96 -- * Exception handling
120 -- * Internal Exports
124#if ! MIN_VERSION_base(4,6,0)
125import Prelude hiding (catch)
128import Data.Binary (decode)
129import Data.Time.Clock (getCurrentTime)
130import Data.Time.Format (formatTime)
131#if MIN_VERSION_time(1,5,0)
132import Data.Time.Format (defaultTimeLocale)
134import System.Locale (defaultTimeLocale)
136import System.Timeout (timeout)
137import Control.Monad (when)
138import Control.Monad.Reader (ask)
139import Control.Monad.IO.Class (liftIO)
140import Control.Applicative ((<$>))
141import Control.Exception (Exception(..), throw, throwIO, SomeException)
142import qualified Control.Exception as Ex (catch, mask, mask_, try)
143import Control.Distributed.Process.Internal.StrictMVar
148import Control.Concurrent.STM
157import Control.Distributed.Process.Internal.CQueue
162import Control.Distributed.Process.Serializable (Serializable, fingerprint)
163import Data.Accessor ((^.), (^:), (^=))
164import Control.Distributed.Static
168import Data.Rank1Typeable (Typeable)
169import qualified Control.Distributed.Static as Static (unstatic, unclosure)
170import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
171import Control.Distributed.Process.Internal.Types
181 , NodeMonitorNotification(..)
191 , ProcessExitException(..)
194 , DidUnlinkProcess(..)
199 , ProcessRegistrationException(..)
201 , ProcessInfoNone(..)
205 , createUnencodedMessage
207 , ImplicitReconnect( NoImplicitReconnect)
212import Control.Distributed.Process.Internal.Messaging
219import Control.Distributed.Process.Management.Internal.Types
222import Control.Distributed.Process.Management.Internal.Trace.Types
225import Control.Distributed.Process.Internal.WeakTQueue
233--------------------------------------------------------------------------------
235--------------------------------------------------------------------------------
238send :: Serializable a => ProcessId -> a -> Process ()
239-- This requires a lookup on every send. If we want to avoid that we need to
240-- modify serializable to allow for stateful (IO) deserialization.
243 let us = processId proc
244 node = processNode proc
245 nodeId = localNodeId node
246 destNode = (processNodeId them) in do
247 case destNode == nodeId of
248 True -> sendLocal them msg
249 False -> liftIO $ sendMessage (processNode proc)
250 (ProcessIdentifier (processId proc))
251 (ProcessIdentifier them)
254 -- We do not fire the trace event until after the sending is complete;
255 -- In the remote case, 'sendMessage' can block in the networking stack.
256 liftIO $ traceEvent (localEventBus node)
257 (MxSent them us (createUnencodedMessage msg))
259-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
260-- and (in the case when the destination process resides on the same local
261-- node) therefore ensure that the payload is fully evaluated before it is
263unsafeSend :: Serializable a => ProcessId -> a -> Process ()
264unsafeSend = Unsafe.send
266-- | Send a message unreliably.
268-- Unlike 'send', this function is insensitive to 'reconnect'. It will
269-- try to send the message regardless of the history of connection failures
272-- Message passing with 'usend' is ordered for a given sender and receiver
273-- if the messages arrive at all.
275usend :: Serializable a => ProcessId -> a -> Process ()
278 let there = processNodeId them
280 then sendLocal them msg
281 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
284-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize
285-- the message when the destination process resides on the same local
286-- node. Therefore, a local receiver would need to be prepared to cope with any
287-- errors resulting from evaluation of the message.
288unsafeUSend :: Serializable a => ProcessId -> a -> Process ()
289unsafeUSend = Unsafe.usend
291-- | Wait for a message of a specific type
292expect :: forall a. Serializable a => Process a
293expect = receiveWait [match return]
295--------------------------------------------------------------------------------
297--------------------------------------------------------------------------------
299-- | Create a new typed channel
300newChan :: Serializable a => Process (SendPort a, ReceivePort a)
303 liftIO . modifyMVar (processState proc) $ \st -> do
304 let lcid = st ^. channelCounter
305 let cid = SendPortId { sendPortProcessId = processId proc
306 , sendPortLocalId = lcid
308 let sport = SendPort cid
309 chan <- liftIO newTQueueIO
310 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
311 let rport = ReceivePort $ readTQueue chan
312 let tch = TypedChannel chan'
313 return ( (channelCounter ^: (+ 1))
314 . (typedChannelWithId lcid ^= Just tch)
319 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
320 finalizer st lcid = modifyMVar_ st $
321 return . (typedChannelWithId lcid ^= Nothing)
323-- | Send a message on a typed channel
324sendChan :: Serializable a => SendPort a -> a -> Process ()
325sendChan (SendPort cid) msg = do
327 let node = localNodeId (processNode proc)
328 destNode = processNodeId (sendPortProcessId cid) in do
329 case destNode == node of
330 True -> sendChanLocal cid msg
332 liftIO $ sendBinary (processNode proc)
333 (ProcessIdentifier (processId proc))
334 (SendPortIdentifier cid)
338-- | Send a message on a typed channel. This function makes /no/ attempt to
339-- serialize and (in the case when the @ReceivePort@ resides on the same local
340-- node) therefore ensure that the payload is fully evaluated before it is
342unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
343unsafeSendChan = Unsafe.sendChan
345-- | Wait for a message on a typed channel
346receiveChan :: Serializable a => ReceivePort a -> Process a
347receiveChan = liftIO . atomically . receiveSTM
349-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
350-- non-blocking check for a message.
351receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
352receiveChanTimeout 0 ch = liftIO . atomically $
353 (Just <$> receiveSTM ch) `orElse` return Nothing
354receiveChanTimeout n ch = liftIO . timeout n . atomically $
357-- | Merge a list of typed channels.
359-- The result port is left-biased: if there are messages available on more
360-- than one port, the first available message is returned.
361mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
362mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
364-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
366mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
367mergePortsRR = \ps -> do
368 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
369 return $ ReceivePort (rr psVar)
373 rotate (x:xs) = xs ++ [x]
375 rr :: TVar [STM a] -> STM a
378 a <- foldr1 orElse ps
379 writeTVar psVar (rotate ps)
382--------------------------------------------------------------------------------
383-- Advanced messaging --
384--------------------------------------------------------------------------------
386-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
387newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
390-- | Test the matches in order against each message in the queue
391receiveWait :: [Match b] -> Process b
393 queue <- processQueue <$> ask
394 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
397-- | Like 'receiveWait' but with a timeout.
399-- If the timeout is zero do a non-blocking check for matching messages. A
400-- non-zero timeout is applied only when waiting for incoming messages (that is,
401-- /after/ we have checked the messages that are already in the mailbox).
402receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
403receiveTimeout t ms = do
404 queue <- processQueue <$> ask
405 let blockSpec = if t == 0 then NonBlocking else Timeout t
406 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
408 Nothing -> return Nothing
409 Just proc -> Just <$> proc
411-- | Match on a typed channel
412matchChan :: ReceivePort a -> (a -> Process b) -> Match b
413matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
415-- | Match on an arbitrary STM action.
417-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
418-- actions with checks against our process' mailbox and/or any typed channel
419-- @ReceivePort@s we may hold.
421-- This allows us to process multiple input streams /along with our mailbox/,
422-- in just the same way that 'matchChan' supports checking both the mailbox
423-- /and/ an arbitrary set of typed channels in one atomic transaction.
425-- Note there are no ordering guarnatees with respect to these disparate input
428matchSTM :: STM a -> (a -> Process b) -> Match b
429matchSTM stm fn = Match $ MatchChan (fmap fn stm)
431-- | Match against any message of the right type
432match :: forall a b. Serializable a => (a -> Process b) -> Match b
433match = matchIf (const True)
435-- | Match against any message of the right type that satisfies a predicate
436matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
437matchIf c p = Match $ MatchMsg $ \msg ->
438 case messageFingerprint msg == fingerprint (undefined :: a) of
441 (UnencodedMessage _ m) ->
442 let m' = unsafeCoerce m :: a in
446 (EncodedMessage _ _) ->
447 if (c decoded) then Just (p decoded) else Nothing
450 -- Make sure the value is fully decoded so that we don't hang to
451 -- bytestrings when the process calling 'matchIf' doesn't process
452 -- the values immediately
453 !decoded = decode (messageEncoding msg)
455-- | Match against any message, regardless of the underlying (contained) type
456matchMessage :: (Message -> Process Message) -> Match Message
457matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
459-- | Match against any message (regardless of underlying type) that satisfies a predicate
460matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
461matchMessageIf c p = Match $ MatchMsg $ \msg ->
466-- | Forward a raw 'Message' to the given 'ProcessId'.
467forward :: Message -> ProcessId -> Process ()
470 let node = processNode proc
472 nid = localNodeId node
473 destNode = (processNodeId them) in do
474 case destNode == nid of
475 True -> sendCtrlMsg Nothing (LocalSend them msg)
476 False -> liftIO $ sendPayload (processNode proc)
477 (ProcessIdentifier (processId proc))
478 (ProcessIdentifier them)
480 (messageToPayload msg)
481 -- We do not fire the trace event until after the sending is complete;
482 -- In the remote case, 'sendMessage' can block in the networking stack.
483 liftIO $ traceEvent (localEventBus node)
486-- | Forward a raw 'Message' to the given 'ProcessId'.
488-- Unlike 'forward', this function is insensitive to 'reconnect'. It will
489-- try to send the message regardless of the history of connection failures
491uforward :: Message -> ProcessId -> Process ()
492uforward msg them = do
494 let node = processNode proc
496 nid = localNodeId node
497 destNode = (processNodeId them) in do
498 case destNode == nid of
499 True -> sendCtrlMsg Nothing (LocalSend them msg)
500 False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)
502 -- We do not fire the trace event until after the sending is complete;
503 -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
504 liftIO $ traceEvent (localEventBus node)
507-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
508-- 'Serializable' - like the datum they contain - but also note, deserialising
509-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
510-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
515-- send self (wrapMessage "blah")
516-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
517-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
518-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
521wrapMessage :: Serializable a => a -> Message
522wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
524-- | This is the /unsafe/ variant of 'wrapMessage'. See
525-- "Control.Distributed.Process.UnsafePrimitives" for details.
526unsafeWrapMessage :: Serializable a => a -> Message
527unsafeWrapMessage = Unsafe.wrapMessage
529-- [note Serializable UnencodedMessage]
530-- if we attempt to serialise an UnencodedMessage, it will be converted to an
531-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
532-- Primitives.hs for the gory details
534-- | Attempt to unwrap a raw 'Message'.
535-- If the type of the decoded message payload matches the expected type, the
536-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
539-- This expression, for example, will evaluate to @Nothing@
540-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
542-- Whereas this expression, will yield @Just "foo"@
543-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
545unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
547 case messageFingerprint msg == fingerprint (undefined :: a) of
548 False -> return Nothing :: m (Maybe a)
550 (UnencodedMessage _ ms) ->
551 let ms' = unsafeCoerce ms :: a
553 (EncodedMessage _ _) ->
554 return (Just (decoded))
556 decoded :: a -- note [decoding]
557 !decoded = decode (messageEncoding msg)
559-- | Attempt to handle a raw 'Message'.
560-- If the type of the message matches the type of the first argument to
561-- the supplied expression, then the message will be decoded and the expression
562-- evaluated against its value. If this runtime type checking fails however,
563-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
564-- evaluation proceeds, the resulting value with be wrapped with @Just@.
566-- Intended for use in `catchesExit` and `matchAny` primitives.
568handleMessage :: forall m a b. (Monad m, Serializable a)
569 => Message -> (a -> m b) -> m (Maybe b)
570handleMessage msg proc = handleMessageIf msg (const True) proc
572-- | Conditionally handle a raw 'Message'.
573-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
574-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
575-- for further information about runtime type checking.
577handleMessageIf :: forall m a b . (Monad m, Serializable a)
582handleMessageIf msg c proc = do
583 case messageFingerprint msg == fingerprint (undefined :: a) of
584 False -> return Nothing :: m (Maybe b)
586 (UnencodedMessage _ ms) ->
587 let ms' = unsafeCoerce ms :: a in
589 True -> do { r <- proc ms'; return (Just r) }
590 False -> return Nothing :: m (Maybe b)
591 (EncodedMessage _ _) ->
593 True -> do { r <- proc (decoded :: a); return (Just r) }
594 False -> return Nothing :: m (Maybe b)
596 decoded :: a -- note [decoding]
597 !decoded = decode (messageEncoding msg)
599-- | As 'handleMessage' but ignores result, which is useful if you don't
600-- care whether or not the handler succeeded.
602handleMessage_ :: forall m a . (Monad m, Serializable a)
603 => Message -> (a -> m ()) -> m ()
604handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
606-- | Conditional version of 'handleMessage_'.
608handleMessageIf_ :: forall m a . (Monad m, Serializable a)
613handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
615-- | Match against an arbitrary message. 'matchAny' removes the first available
616-- message from the process mailbox. To handle arbitrary /raw/ messages once
617-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
619matchAny :: forall b. (Message -> Process b) -> Match b
620matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
622-- | Match against an arbitrary message. Intended for use with 'handleMessage'
623-- and 'unwrapMessage', this function /only/ removes a message from the process
624-- mailbox, /if/ the supplied condition matches. The success (or failure) of
625-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
626-- here, i.e., if the condition evaluates to @True@ then the message will
627-- be removed from the process mailbox and decoded, but that does /not/
628-- guarantee that an expression passed to @handleMessage@ will pass the
629-- runtime type checks and therefore be evaluated.
631matchAnyIf :: forall a b. (Serializable a)
633 -> (Message -> Process b)
635matchAnyIf c p = Match $ MatchMsg $ \msg ->
636 case messageFingerprint msg == fingerprint (undefined :: a) of
637 True | check -> Just (p msg)
642 (EncodedMessage _ _) -> c decoded
643 (UnencodedMessage _ m') -> c (unsafeCoerce m')
645 decoded :: a -- note [decoding]
646 !decoded = decode (messageEncoding msg)
650For an EncodedMessage, we need to ensure the value is fully decoded so that
651we don't hang to bytestrings if the calling process doesn't evaluate
652immediately. For UnencodedMessage we know (because the fingerprint comparison
653succeeds) that unsafeCoerce will not fail.
656-- | Remove any message from the queue
657matchUnknown :: Process b -> Match b
658matchUnknown p = Match $ MatchMsg (const (Just p))
660-- | Receives messages and forwards them to @pid@ if @p msg == True@.
661delegate :: ProcessId -> (Message -> Bool) -> Process ()
664 matchAny (\m -> case (p m) of
665 True -> forward m pid
670-- | A straight relay that forwards all messages to the supplied pid.
671relay :: ProcessId -> Process ()
672relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
674-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
675-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
676-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
677-- fails, no action is taken and the proxy process will continue running.
678proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
682 next <- handleMessage m proc
684 Just True -> forward m pid
685 Just False -> return () -- explicitly ignored
686 Nothing -> return ()) -- un-routable
690--------------------------------------------------------------------------------
691-- Process management --
692--------------------------------------------------------------------------------
694-- | Thrown by 'terminate'
695data ProcessTerminationException = ProcessTerminationException
696 deriving (Show, Typeable)
698instance Exception ProcessTerminationException
700-- | Terminate immediately (throws a ProcessTerminationException)
701terminate :: Process a
702terminate = liftIO $ throwIO ProcessTerminationException
705-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
706die :: Serializable a => a -> Process b
708-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
709-- could be decoded by a handler passed to 'catchExit', re-thrown or even
710-- passed to another process via the tracing mechanism.
712 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
714-- | Forceful request to kill a process. Where 'exit' provides an exception
715-- that can be caught and handled, 'kill' throws an unexposed exception type
716-- which cannot be handled explicitly (by type).
717kill :: ProcessId -> String -> Process ()
718-- NOTE: We send the message to our local node controller, which will then
719-- forward it to a remote node controller (if applicable). Sending it directly
720-- to a remote node controller means that that the message may overtake a
721-- 'monitor' or 'link' request.
722kill them reason = sendCtrlMsg Nothing (Kill them reason)
724-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
725-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
726-- manner can be handled using the 'catchExit' family of functions.
727exit :: Serializable a => ProcessId -> a -> Process ()
728-- NOTE: We send the message to our local node controller, which will then
729-- forward it to a remote node controller (if applicable). Sending it directly
730-- to a remote node controller means that that the message may overtake a
731-- 'monitor' or 'link' request.
732exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
734-- | Catches 'ProcessExitException'. The handler will not be applied unless its
735-- type matches the encoded data stored in the exception (see the /reason/
736-- argument given to the 'exit' primitive). If the handler cannot be applied,
737-- the exception will be re-thrown.
739-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
740-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
741catchExit :: forall a b . (Show a, Serializable a)
743 -> (ProcessId -> a -> Process b)
745catchExit act exitHandler = catch act handleExit
747 handleExit ex@(ProcessExitException from msg) =
748 if messageFingerprint msg == fingerprint (undefined :: a)
749 then exitHandler from decoded
750 else liftIO $ throwIO ex
753 -- Make sure the value is fully decoded so that we don't hang to
754 -- bytestrings if the caller doesn't use the value immediately
755 !decoded = decode (messageEncoding msg)
757-- | Lift 'Control.Exception.catches' (almost).
759-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
760-- message, a handler must accept inputs of the expected type. In order to
761-- handle a list of potentially different handlers (and therefore input types),
762-- a handler passed to 'catchesExit' must accept 'Message' and return
763-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
765-- See 'maybeHandleMessage' and 'Message' for more details.
766catchesExit :: Process b
767 -> [(ProcessId -> Message -> (Process (Maybe b)))]
769catchesExit act handlers = catch act ((flip handleExit) handlers)
771 handleExit :: ProcessExitException
772 -> [(ProcessId -> Message -> Process (Maybe b))]
774 handleExit ex [] = liftIO $ throwIO ex
775 handleExit ex@(ProcessExitException from msg) (h:hs) = do
778 Nothing -> handleExit ex hs
781-- | Our own process ID
782getSelfPid :: Process ProcessId
783getSelfPid = processId <$> ask
785-- | Get the node ID of our local node
786getSelfNode :: Process NodeId
787getSelfNode = localNodeId . processNode <$> ask
790-- | Get statistics about the specified node
791getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
793 selfNode <- getSelfNode
795 then Right `fmap` getLocalNodeStats -- optimisation
796 else getNodeStatsRemote selfNode
798 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
799 getNodeStatsRemote selfNode = do
800 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
801 bracket (monitorNode nid) unmonitor $ \mRef ->
802 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
803 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
804 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
807-- | Get statistics about our local node
808getLocalNodeStats :: Process NodeStats
809getLocalNodeStats = do
811 sendCtrlMsg Nothing $ GetNodeStats self
812 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
814-- | Get information about the specified process
815getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
817 let them = processNodeId pid in do
819 dest <- mkNode them us
820 sendCtrlMsg dest $ GetInfo pid
822 match (\(p :: ProcessInfo) -> return $ Just p)
823 , match (\(_ :: ProcessInfoNone) -> return Nothing)
825 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
826 mkNode them us = case them == us of
827 True -> return Nothing
828 _ -> return $ Just them
830--------------------------------------------------------------------------------
831-- Monitoring and linking --
832--------------------------------------------------------------------------------
834-- | Link to a remote process (asynchronous)
836-- When process A links to process B (that is, process A calls
837-- @link pidB@) then an asynchronous exception will be thrown to process A
838-- when process B terminates (normally or abnormally), or when process A gets
839-- disconnected from process B. Although it is /technically/ possible to catch
840-- these exceptions, chances are if you find yourself trying to do so you should
841-- probably be using 'monitor' rather than 'link'. In particular, code such as
843-- > link pidB -- Link to process B
844-- > expect -- Wait for a message from process B
845-- > unlink pidB -- Unlink again
847-- doesn't quite do what one might expect: if process B sends a message to
848-- process A, and /subsequently terminates/, then process A might or might not
849-- be terminated too, depending on whether the exception is thrown before or
850-- after the 'unlink' (i.e., this code has a race condition).
852-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
853-- call to 'link' has no effect.
855-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
856-- Linking makes no distinction between normal and abnormal termination of
857-- the remote process.
858link :: ProcessId -> Process ()
859link = sendCtrlMsg Nothing . Link . ProcessIdentifier
861-- | Monitor another process (asynchronous)
863-- When process A monitors process B (that is, process A calls
864-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
865-- when process B terminates (normally or abnormally), or when process A gets
866-- disconnected from process B. You receive this message like any other (using
867-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
868-- 'DiedDisconnect', etc.).
870-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
871-- multiple monitors are set up, multiple notifications will be delivered
872-- and monitors can be disabled individually using 'unmonitor'.
873monitor :: ProcessId -> Process MonitorRef
874monitor = monitor' . ProcessIdentifier
876-- | Establishes temporary monitoring of another process.
878-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
879-- of @code@. Note: although monitoring is no longer active when
880-- @withMonitor@ returns, there might still be unreceived monitor
881-- messages in the queue.
883withMonitor :: ProcessId -> Process a -> Process a
884withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
885 -- unmonitor blocks waiting for the response, so there's a possibility
886 -- that an exception might interrupt withMonitor before the unmonitor
887 -- has completed. I think that's better than making the unmonitor
892-- This is synchronous in the sense that once it returns you are guaranteed
893-- that no exception will be raised if the remote process dies. However, it is
894-- asynchronous in the sense that we do not wait for a response from the remote
896unlink :: ProcessId -> Process ()
899 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
903-- | Remove a node link
905-- This has the same synchronous/asynchronous nature as 'unlink'.
906unlinkNode :: NodeId -> Process ()
909 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
913-- | Remove a channel (send port) link
915-- This has the same synchronous/asynchronous nature as 'unlink'.
916unlinkPort :: SendPort a -> Process ()
918 unlinkPortAsync sport
919 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
925-- This has the same synchronous/asynchronous nature as 'unlink'.
926unmonitor :: MonitorRef -> Process ()
929 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
933--------------------------------------------------------------------------------
934-- Exception handling --
935--------------------------------------------------------------------------------
937-- | Lift 'Control.Exception.catch'
938catch :: Exception e => Process a -> (e -> Process a) -> Process a
941 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
943-- | Lift 'Control.Exception.try'
944try :: Exception e => Process a -> Process (Either e a)
947 liftIO $ Ex.try (runLocalProcess lproc p)
949-- | Lift 'Control.Exception.mask'
950mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
953 liftIO $ Ex.mask $ \restore ->
954 runLocalProcess lproc (p (liftRestore restore))
956 liftRestore :: (forall a. IO a -> IO a)
957 -> (forall a. Process a -> Process a)
958 liftRestore restoreIO = \p2 -> do
960 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
962-- | Lift 'Control.Exception.mask_'
963mask_ :: Process a -> Process a
966 liftIO $ Ex.mask_ $ runLocalProcess lproc p
968-- | Lift 'Control.Exception.onException'
969onException :: Process a -> Process b -> Process a
970onException p what = p `catch` \e -> do _ <- what
971 liftIO $ throwIO (e :: SomeException)
973-- | Lift 'Control.Exception.bracket'
974bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
975bracket before after thing =
976 mask $ \restore -> do
978 r <- restore (thing a) `onException` after a
982-- | Lift 'Control.Exception.bracket_'
983bracket_ :: Process a -> Process b -> Process c -> Process c
984bracket_ before after thing = bracket before (const after) (const thing)
986-- | Lift 'Control.Exception.finally'
987finally :: Process a -> Process b -> Process a
988finally a sequel = bracket_ (return ()) sequel a
990-- | You need this when using 'catches'
991data Handler a = forall e . Exception e => Handler (e -> Process a)
993instance Functor Handler where
994 fmap f (Handler h) = Handler (fmap f . h)
996-- | Lift 'Control.Exception.catches'
997catches :: Process a -> [Handler a] -> Process a
998catches proc handlers = proc `catch` catchesHandler handlers
1000catchesHandler :: [Handler a] -> SomeException -> Process a
1001catchesHandler handlers e = foldr tryHandler (throw e) handlers
1002 where tryHandler (Handler handler) res
1003 = case fromException e of
1004 Just e' -> handler e'
1007--------------------------------------------------------------------------------
1009--------------------------------------------------------------------------------
1011-- | Like 'expect' but with a timeout
1012expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
1013expectTimeout n = receiveTimeout n [match return]
1015-- | Asynchronous version of 'spawn'
1017-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
1018spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
1019spawnAsync nid proc = do
1020 spawnRef <- getSpawnRef
1023 then sendCtrlMsg Nothing $ Spawn proc spawnRef
1024 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
1027-- | Monitor a node (asynchronous)
1028monitorNode :: NodeId -> Process MonitorRef
1030 monitor' . NodeIdentifier
1032-- | Monitor a typed channel (asynchronous)
1033monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1034monitorPort (SendPort cid) =
1035 monitor' (SendPortIdentifier cid)
1037-- | Remove a monitor (asynchronous)
1038unmonitorAsync :: MonitorRef -> Process ()
1040 sendCtrlMsg Nothing . Unmonitor
1042-- | Link to a node (asynchronous)
1043linkNode :: NodeId -> Process ()
1044linkNode = link' . NodeIdentifier
1046-- | Link to a channel (asynchronous)
1047linkPort :: SendPort a -> Process ()
1048linkPort (SendPort cid) =
1049 link' (SendPortIdentifier cid)
1051-- | Remove a process link (asynchronous)
1052unlinkAsync :: ProcessId -> Process ()
1054 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1056-- | Remove a node link (asynchronous)
1057unlinkNodeAsync :: NodeId -> Process ()
1059 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1061-- | Remove a channel (send port) link (asynchronous)
1062unlinkPortAsync :: SendPort a -> Process ()
1063unlinkPortAsync (SendPort cid) =
1064 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1066--------------------------------------------------------------------------------
1068--------------------------------------------------------------------------------
1072-- @say message@ sends a message (time, pid of the current process, message)
1073-- to the process registered as 'logger'. By default, this process simply
1074-- sends the string to 'stderr'. Individual Cloud Haskell backends might
1075-- replace this with a different logger process, however.
1076say :: String -> Process ()
1078 now <- liftIO getCurrentTime
1080 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
1082--------------------------------------------------------------------------------
1084--------------------------------------------------------------------------------
1086-- | Register a process with the local registry (synchronous). The name must not
1087-- already be registered. The process need not be on this node. A bad
1088-- registration will result in a 'ProcessRegistrationException'
1090-- The process to be registered does not have to be local itself.
1091register :: String -> ProcessId -> Process ()
1092register = registerImpl False
1094-- | Like 'register', but will replace an existing registration.
1095-- The name must already be registered.
1096reregister :: String -> ProcessId -> Process ()
1097reregister = registerImpl True
1099registerImpl :: Bool -> String -> ProcessId -> Process ()
1100registerImpl force label pid = do
1101 mynid <- getSelfNode
1102 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1104 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1105 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1108-- | Register a process with a remote registry (asynchronous).
1110-- The process to be registered does not have to live on the same remote node.
1111-- Reply wil come in the form of a 'RegisterReply' message
1113-- See comments in 'whereisRemoteAsync'
1114registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1115registerRemoteAsync nid label pid = do
1117 sendCtrlMsg (if nid == here then Nothing else Just nid)
1118 (Register label nid (Just pid) False)
1120reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1121reregisterRemoteAsync nid label pid =
1122 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1124-- | Remove a process from the local registry (asynchronous).
1125-- This version will wait until a response is gotten from the
1126-- management process. The name must already be registered.
1127unregister :: String -> Process ()
1128unregister label = do
1129 mynid <- getSelfNode
1130 sendCtrlMsg Nothing (Register label mynid Nothing False)
1132 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1133 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1136-- | Deal with the result from an attempted registration or unregistration
1137-- by throwing an exception if necessary
1138handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()
1139handleRegistrationReply label ok owner =
1141 liftIO $ throwIO $ ProcessRegistrationException label owner
1143-- | Remove a process from a remote registry (asynchronous).
1145-- Reply wil come in the form of a 'RegisterReply' message
1147-- See comments in 'whereisRemoteAsync'
1148unregisterRemoteAsync :: NodeId -> String -> Process ()
1149unregisterRemoteAsync nid label =
1150 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1152-- | Query the local process registry
1153whereis :: String -> Process (Maybe ProcessId)
1155 sendCtrlMsg Nothing (WhereIs label)
1156 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1157 (\(WhereIsReply _ mPid) -> return mPid)
1160-- | Query a remote process registry (asynchronous)
1162-- Reply will come in the form of a 'WhereIsReply' message.
1164-- There is currently no synchronous version of 'whereisRemoteAsync': if
1165-- you implement one yourself, be sure to take into account that the remote
1166-- node might die or get disconnect before it can respond (i.e. you should
1167-- use 'monitorNode' and take appropriate action when you receive a
1168-- 'NodeMonitorNotification').
1169whereisRemoteAsync :: NodeId -> String -> Process ()
1170whereisRemoteAsync nid label = do
1172 sendCtrlMsg (if nid == here then Nothing else Just nid) (WhereIs label)
1174-- | Named send to a process in the local registry (asynchronous)
1175nsend :: Serializable a => String -> a -> Process ()
1177 sendCtrlMsg Nothing (NamedSend label (createMessage msg))
1179-- | Named send to a process in the local registry (asynchronous).
1180-- This function makes /no/ attempt to serialize and (in the case when the
1181-- destination process resides on the same local node) therefore ensure that
1182-- the payload is fully evaluated before it is delivered.
1183unsafeNSend :: Serializable a => String -> a -> Process ()
1184unsafeNSend = Unsafe.nsend
1186-- | Named send to a process in a remote registry (asynchronous)
1187nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1188nsendRemote nid label msg = do
1190 if here == nid then nsend label msg
1191 else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1193-- | Named send to a process in a remote registry (asynchronous)
1194-- This function makes /no/ attempt to serialize and (in the case when the
1195-- destination process resides on the same local node) therefore ensure that
1196-- the payload is fully evaluated before it is delivered.
1197unsafeNSendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1198unsafeNSendRemote = Unsafe.nsendRemote
1200--------------------------------------------------------------------------------
1202--------------------------------------------------------------------------------
1204-- | Resolve a static value
1205unStatic :: Typeable a => Static a -> Process a
1207 rtable <- remoteTable . processNode <$> ask
1208 case Static.unstatic rtable static of
1209 Left err -> fail $ "Could not resolve static value: " ++ err
1212-- | Resolve a closure
1213unClosure :: Typeable a => Closure a -> Process a
1214unClosure closure = do
1215 rtable <- remoteTable . processNode <$> ask
1216 case Static.unclosure rtable closure of
1217 Left err -> fail $ "Could not resolve closure: " ++ err
1220--------------------------------------------------------------------------------
1222--------------------------------------------------------------------------------
1224-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1225-- message passing. However, when network connections get disrupted this
1226-- illusion cannot always be maintained. Once a network connection breaks (even
1227-- temporarily) no further communication on that connection will be possible.
1228-- For example, if process A sends a message to process B, and A is then
1229-- notified (by monitor notification) that it got disconnected from B, A will
1230-- not be able to send any further messages to B, /unless/ A explicitly
1231-- indicates that it is acceptable to attempt to reconnect to B using the
1232-- Cloud Haskell 'reconnect' primitive.
1234-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1235-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1236-- then receives a monitor notification that its connection to B has been lost,
1237-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1238-- and m3 but not m2.
1240-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1241-- to attempt to reconnect on the next send/. In particular, if no further
1242-- communication attempts are made to B then A can use reconnect to clean up
1243-- its connection to B.
1244reconnect :: ProcessId -> Process ()
1247 node <- processNode <$> ask
1248 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1250-- | Reconnect to a sendport. See 'reconnect' for more information.
1251reconnectPort :: SendPort a -> Process ()
1252reconnectPort them = do
1254 node <- processNode <$> ask
1255 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1257--------------------------------------------------------------------------------
1258-- Auxiliary functions --
1259--------------------------------------------------------------------------------
1261sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1263 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1265sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1266sendChanLocal spId msg =
1267 -- we *must* fully serialize/encode the message here, because
1268 -- attempting to use `unsafeCoerce' in the node controller
1269 -- won't work since we know nothing about the required type
1270 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1272getMonitorRefFor :: Identifier -> Process MonitorRef
1273getMonitorRefFor ident = do
1275 liftIO $ modifyMVar (processState proc) $ \st -> do
1276 let counter = st ^. monitorCounter
1277 return ( monitorCounter ^: (+ 1) $ st
1278 , MonitorRef ident counter
1281getSpawnRef :: Process SpawnRef
1284 liftIO $ modifyMVar (processState proc) $ \st -> do
1285 let counter = st ^. spawnCounter
1286 return ( spawnCounter ^: (+ 1) $ st
1290-- | Monitor a process/node/channel
1291monitor' :: Identifier -> Process MonitorRef
1293 monitorRef <- getMonitorRefFor ident
1294 sendCtrlMsg Nothing $ Monitor monitorRef
1297-- | Link to a process/node/channel
1298link' :: Identifier -> Process ()
1299link' = 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