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
92 , reregisterRemoteAsync
93 , unregisterRemoteAsync
99 -- * Exception handling
123 -- * Internal Exports
127#if ! MIN_VERSION_base(4,6,0)
128import Prelude hiding (catch)
131import Data.Binary (Binary(..), Put, Get, decode)
132import Data.Time.Clock (getCurrentTime, UTCTime(..))
133import Data.Time.Calendar (Day(..))
134import Data.Time.Format (formatTime)
135#if MIN_VERSION_time(1,5,0)
136import Data.Time.Format (defaultTimeLocale)
138import System.Locale (defaultTimeLocale)
140import System.Timeout (timeout)
141import Control.Monad (when, void)
142import Control.Monad.Reader (ask)
143import Control.Monad.IO.Class (liftIO)
144import Control.Monad.Catch
150import qualified Control.Monad.Catch as Catch
151import Control.Applicative
152import Control.Distributed.Process.Internal.StrictMVar
157import Control.Concurrent.STM
166import Control.Distributed.Process.Internal.CQueue
171import Control.Distributed.Process.Serializable (Serializable, fingerprint)
172import Data.Accessor ((^.), (^:), (^=))
173import Control.Distributed.Static
177import Data.Rank1Typeable (Typeable)
178import qualified Control.Distributed.Static as Static (unstatic, unclosure)
179import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
180import Control.Distributed.Process.Internal.Types
190 , NodeMonitorNotification(..)
191 , ProcessMonitorNotification(..)
201 , ProcessExitException(..)
204 , DidUnlinkProcess(..)
209 , ProcessRegistrationException(..)
211 , ProcessInfoNone(..)
215 , createUnencodedMessage
216 , ImplicitReconnect( NoImplicitReconnect)
221import Control.Distributed.Process.Internal.Messaging
228import Control.Distributed.Process.Management.Internal.Types
231import Control.Distributed.Process.Management.Internal.Trace.Types
234import Control.Distributed.Process.Internal.WeakTQueue
243--------------------------------------------------------------------------------
245--------------------------------------------------------------------------------
248send :: Serializable a => ProcessId -> a -> Process ()
249-- This requires a lookup on every send. If we want to avoid that we need to
250-- modify serializable to allow for stateful (IO) deserialization.
253 let us = processId proc
254 node = processNode proc
255 nodeId = localNodeId node
256 destNode = (processNodeId them) in do
257 case destNode == nodeId of
258 True -> sendLocal them msg
259 False -> liftIO $ sendMessage (processNode proc)
260 (ProcessIdentifier (processId proc))
261 (ProcessIdentifier them)
264 -- We do not fire the trace event until after the sending is complete;
265 -- In the remote case, 'sendMessage' can block in the networking stack.
266 liftIO $ traceEvent (localEventBus node)
267 (MxSent them us (createUnencodedMessage msg))
269-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
270-- and (in the case when the destination process resides on the same local
271-- node) therefore ensure that the payload is fully evaluated before it is
273unsafeSend :: Serializable a => ProcessId -> a -> Process ()
274unsafeSend = Unsafe.send
276-- | Send a message unreliably.
278-- Unlike 'send', this function is insensitive to 'reconnect'. It will
279-- try to send the message regardless of the history of connection failures
282-- Message passing with 'usend' is ordered for a given sender and receiver
283-- if the messages arrive at all.
285usend :: Serializable a => ProcessId -> a -> Process ()
288 let there = processNodeId them
290 then sendLocal them msg
291 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
294-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize
295-- the message when the destination process resides on the same local
296-- node. Therefore, a local receiver would need to be prepared to cope with any
297-- errors resulting from evaluation of the message.
298unsafeUSend :: Serializable a => ProcessId -> a -> Process ()
299unsafeUSend = Unsafe.usend
301-- | Wait for a message of a specific type
302expect :: forall a. Serializable a => Process a
303expect = receiveWait [match return]
305--------------------------------------------------------------------------------
307--------------------------------------------------------------------------------
309-- | Create a new typed channel, bound to the calling Process
311-- Note that the channel is bound to the lifecycle of the process that evaluates
312-- this function, such that when it dies/exits, the channel will no longer
313-- function, but will remain accessible. Thus reading from the ReceivePort will
314-- fail silently thereafter, blocking indefinitely (unless a timeout is used).
316newChan :: Serializable a => Process (SendPort a, ReceivePort a)
319 liftIO . modifyMVar (processState proc) $ \st -> do
320 let lcid = st ^. channelCounter
321 let cid = SendPortId { sendPortProcessId = processId proc
322 , sendPortLocalId = lcid
324 let sport = SendPort cid
325 chan <- liftIO newTQueueIO
326 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
327 let rport = ReceivePort $ readTQueue chan
328 let tch = TypedChannel chan'
329 return ( (channelCounter ^: (+ 1))
330 . (typedChannelWithId lcid ^= Just tch)
335 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
336 finalizer st lcid = modifyMVar_ st $
337 return . (typedChannelWithId lcid ^= Nothing)
339-- | Send a message on a typed channel
340sendChan :: Serializable a => SendPort a -> a -> Process ()
341sendChan (SendPort cid) msg = do
343 let node = localNodeId (processNode proc)
344 destNode = processNodeId (sendPortProcessId cid) in do
345 case destNode == node of
346 True -> sendChanLocal cid msg
348 liftIO $ sendBinary (processNode proc)
349 (ProcessIdentifier (processId proc))
350 (SendPortIdentifier cid)
354-- | Send a message on a typed channel. This function makes /no/ attempt to
355-- serialize and (in the case when the @ReceivePort@ resides on the same local
356-- node) therefore ensure that the payload is fully evaluated before it is
358unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
359unsafeSendChan = Unsafe.sendChan
361-- | Wait for a message on a typed channel
362receiveChan :: Serializable a => ReceivePort a -> Process a
363receiveChan = liftIO . atomically . receiveSTM
365-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
366-- non-blocking check for a message.
367receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
368receiveChanTimeout 0 ch = liftIO . atomically $
369 (Just <$> receiveSTM ch) `orElse` return Nothing
370receiveChanTimeout n ch = liftIO . timeout n . atomically $
373-- | Merge a list of typed channels.
375-- The result port is left-biased: if there are messages available on more
376-- than one port, the first available message is returned.
377mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
378mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
380-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
382mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
383mergePortsRR = \ps -> do
384 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
385 return $ ReceivePort (rr psVar)
389 rotate (x:xs) = xs ++ [x]
391 rr :: TVar [STM a] -> STM a
394 a <- foldr1 orElse ps
395 writeTVar psVar (rotate ps)
398--------------------------------------------------------------------------------
399-- Advanced messaging --
400--------------------------------------------------------------------------------
402-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
403newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
406-- | Test the matches in order against each message in the queue
407receiveWait :: [Match b] -> Process b
409 queue <- processQueue <$> ask
410 mProc <- liftIO $ dequeue queue Blocking (map unMatch ms)
413 Nothing -> die $ "System Invariant Violation: CQueue.hs returned `Nothing` "
414 ++ "in the absence of a timeout value."
416-- | Like 'receiveWait' but with a timeout.
418-- If the timeout is zero do a non-blocking check for matching messages. A
419-- non-zero timeout is applied only when waiting for incoming messages (that is,
420-- /after/ we have checked the messages that are already in the mailbox).
421receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
422receiveTimeout t ms = do
423 queue <- processQueue <$> ask
424 let blockSpec = if t == 0 then NonBlocking else Timeout t
425 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
427 Nothing -> return Nothing
428 Just proc -> Just <$> proc
430-- | Match on a typed channel
431matchChan :: ReceivePort a -> (a -> Process b) -> Match b
432matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
434-- | Match on an arbitrary STM action.
436-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
437-- actions with checks against our process' mailbox and/or any typed channel
438-- @ReceivePort@s we may hold.
440-- This allows us to process multiple input streams /along with our mailbox/,
441-- in just the same way that 'matchChan' supports checking both the mailbox
442-- /and/ an arbitrary set of typed channels in one atomic transaction.
444-- Note there are no ordering guarnatees with respect to these disparate input
447matchSTM :: STM a -> (a -> Process b) -> Match b
448matchSTM stm fn = Match $ MatchChan (fmap fn stm)
450-- | Match against any message of the right type
451match :: forall a b. Serializable a => (a -> Process b) -> Match b
452match = matchIf (const True)
454-- | Match against any message of the right type that satisfies a predicate
455matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
456matchIf c p = Match $ MatchMsg $ \msg ->
457 case messageFingerprint msg == fingerprint (undefined :: a) of
460 (UnencodedMessage _ m) ->
461 let m' = unsafeCoerce m :: a in
465 (EncodedMessage _ _) ->
466 if (c decoded) then Just (p decoded) else Nothing
469 -- Make sure the value is fully decoded so that we don't hang to
470 -- bytestrings when the process calling 'matchIf' doesn't process
471 -- the values immediately
472 !decoded = decode (messageEncoding msg)
474-- | Match against any message, regardless of the underlying (contained) type
475matchMessage :: (Message -> Process Message) -> Match Message
476matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
478-- | Match against any message (regardless of underlying type) that satisfies a predicate
479matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
480matchMessageIf c p = Match $ MatchMsg $ \msg ->
485-- | Forward a raw 'Message' to the given 'ProcessId'.
486forward :: Message -> ProcessId -> Process ()
489 let node = processNode proc
491 nid = localNodeId node
492 destNode = (processNodeId them) in do
493 case destNode == nid of
494 True -> sendCtrlMsg Nothing (LocalSend them msg)
495 False -> liftIO $ sendPayload (processNode proc)
496 (ProcessIdentifier (processId proc))
497 (ProcessIdentifier them)
499 (messageToPayload msg)
500 -- We do not fire the trace event until after the sending is complete;
501 -- In the remote case, 'sendMessage' can block in the networking stack.
502 liftIO $ traceEvent (localEventBus node)
505-- | Forward a raw 'Message' to the given 'ProcessId'.
507-- Unlike 'forward', this function is insensitive to 'reconnect'. It will
508-- try to send the message regardless of the history of connection failures
510uforward :: Message -> ProcessId -> Process ()
511uforward msg them = do
513 let node = processNode proc
515 nid = localNodeId node
516 destNode = (processNodeId them) in do
517 case destNode == nid of
518 True -> sendCtrlMsg Nothing (LocalSend them msg)
519 False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)
521 -- We do not fire the trace event until after the sending is complete;
522 -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
523 liftIO $ traceEvent (localEventBus node)
526-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
527-- 'Serializable' - like the datum they contain - but also note, deserialising
528-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
529-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
534-- send self (wrapMessage "blah")
535-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
536-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
537-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
540wrapMessage :: Serializable a => a -> Message
541wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
543-- | This is the /unsafe/ variant of 'wrapMessage'. See
544-- "Control.Distributed.Process.UnsafePrimitives" for details.
545unsafeWrapMessage :: Serializable a => a -> Message
546unsafeWrapMessage = Unsafe.wrapMessage
548-- [note Serializable UnencodedMessage]
549-- if we attempt to serialise an UnencodedMessage, it will be converted to an
550-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
551-- Primitives.hs for the gory details
553-- | Attempt to unwrap a raw 'Message'.
554-- If the type of the decoded message payload matches the expected type, the
555-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
558-- This expression, for example, will evaluate to @Nothing@
559-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
561-- Whereas this expression, will yield @Just "foo"@
562-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
564unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
566 case messageFingerprint msg == fingerprint (undefined :: a) of
567 False -> return Nothing :: m (Maybe a)
569 (UnencodedMessage _ ms) ->
570 let ms' = unsafeCoerce ms :: a
572 (EncodedMessage _ _) ->
573 return (Just (decoded))
575 decoded :: a -- note [decoding]
576 !decoded = decode (messageEncoding msg)
578-- | Attempt to handle a raw 'Message'.
579-- If the type of the message matches the type of the first argument to
580-- the supplied expression, then the message will be decoded and the expression
581-- evaluated against its value. If this runtime type checking fails however,
582-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
583-- evaluation proceeds, the resulting value with be wrapped with @Just@.
585-- Intended for use in `catchesExit` and `matchAny` primitives.
587handleMessage :: forall m a b. (Monad m, Serializable a)
588 => Message -> (a -> m b) -> m (Maybe b)
589handleMessage msg proc = handleMessageIf msg (const True) proc
591-- | Conditionally handle a raw 'Message'.
592-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
593-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
594-- for further information about runtime type checking.
596handleMessageIf :: forall m a b . (Monad m, Serializable a)
601handleMessageIf msg c proc = do
602 case messageFingerprint msg == fingerprint (undefined :: a) of
603 False -> return Nothing :: m (Maybe b)
605 (UnencodedMessage _ ms) ->
606 let ms' = unsafeCoerce ms :: a in
608 True -> do { r <- proc ms'; return (Just r) }
609 False -> return Nothing :: m (Maybe b)
610 (EncodedMessage _ _) ->
612 True -> do { r <- proc (decoded :: a); return (Just r) }
613 False -> return Nothing :: m (Maybe b)
615 decoded :: a -- note [decoding]
616 !decoded = decode (messageEncoding msg)
618-- | As 'handleMessage' but ignores result, which is useful if you don't
619-- care whether or not the handler succeeded.
621handleMessage_ :: forall m a . (Monad m, Serializable a)
622 => Message -> (a -> m ()) -> m ()
623handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
625-- | Conditional version of 'handleMessage_'.
627handleMessageIf_ :: forall m a . (Monad m, Serializable a)
632handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
634-- | Match against an arbitrary message. 'matchAny' removes the first available
635-- message from the process mailbox. To handle arbitrary /raw/ messages once
636-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
638matchAny :: forall b. (Message -> Process b) -> Match b
639matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
641-- | Match against an arbitrary message. Intended for use with 'handleMessage'
642-- and 'unwrapMessage', this function /only/ removes a message from the process
643-- mailbox, /if/ the supplied condition matches. The success (or failure) of
644-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
645-- here, i.e., if the condition evaluates to @True@ then the message will
646-- be removed from the process mailbox and decoded, but that does /not/
647-- guarantee that an expression passed to @handleMessage@ will pass the
648-- runtime type checks and therefore be evaluated.
650matchAnyIf :: forall a b. (Serializable a)
652 -> (Message -> Process b)
654matchAnyIf c p = Match $ MatchMsg $ \msg ->
655 case messageFingerprint msg == fingerprint (undefined :: a) of
656 True | check -> Just (p msg)
661 (EncodedMessage _ _) -> c decoded
662 (UnencodedMessage _ m') -> c (unsafeCoerce m')
664 decoded :: a -- note [decoding]
665 !decoded = decode (messageEncoding msg)
669For an EncodedMessage, we need to ensure the value is fully decoded so that
670we don't hang to bytestrings if the calling process doesn't evaluate
671immediately. For UnencodedMessage we know (because the fingerprint comparison
672succeeds) that unsafeCoerce will not fail.
675-- | Remove any message from the queue
676matchUnknown :: Process b -> Match b
677matchUnknown p = Match $ MatchMsg (const (Just p))
679-- | Receives messages and forwards them to @pid@ if @p msg == True@.
680delegate :: ProcessId -> (Message -> Bool) -> Process ()
683 matchAny (\m -> case (p m) of
684 True -> forward m pid
689-- | A straight relay that forwards all messages to the supplied pid.
690relay :: ProcessId -> Process ()
691relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
693-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
694-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
695-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
696-- fails, no action is taken and the proxy process will continue running.
697proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
701 next <- handleMessage m proc
703 Just True -> forward m pid
704 Just False -> return () -- explicitly ignored
705 Nothing -> return ()) -- un-routable
709--------------------------------------------------------------------------------
710-- Process management --
711--------------------------------------------------------------------------------
713-- | Thrown by 'terminate'
714data ProcessTerminationException = ProcessTerminationException
715 deriving (Show, Typeable)
717instance Exception ProcessTerminationException
719-- | Terminate immediately (throws a ProcessTerminationException)
720terminate :: Process a
721terminate = throwM ProcessTerminationException
724-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
725die :: Serializable a => a -> Process b
727-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
728-- could be decoded by a handler passed to 'catchExit', re-thrown or even
729-- passed to another process via the tracing mechanism.
731 throwM (ProcessExitException pid (createMessage reason))
733-- | Forceful request to kill a process. Where 'exit' provides an exception
734-- that can be caught and handled, 'kill' throws an unexposed exception type
735-- which cannot be handled explicitly (by type).
736kill :: ProcessId -> String -> Process ()
737-- NOTE: We send the message to our local node controller, which will then
738-- forward it to a remote node controller (if applicable). Sending it directly
739-- to a remote node controller means that that the message may overtake a
740-- 'monitor' or 'link' request.
741kill them reason = sendCtrlMsg Nothing (Kill them reason)
743-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
744-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
745-- manner can be handled using the 'catchExit' family of functions.
746exit :: Serializable a => ProcessId -> a -> Process ()
747-- NOTE: We send the message to our local node controller, which will then
748-- forward it to a remote node controller (if applicable). Sending it directly
749-- to a remote node controller means that that the message may overtake a
750-- 'monitor' or 'link' request.
751exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
753-- | Catches 'ProcessExitException'. The handler will not be applied unless its
754-- type matches the encoded data stored in the exception (see the /reason/
755-- argument given to the 'exit' primitive). If the handler cannot be applied,
756-- the exception will be re-thrown.
758-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
759-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
760catchExit :: forall a b . (Show a, Serializable a)
762 -> (ProcessId -> a -> Process b)
764catchExit act exitHandler = Catch.catch act handleExit
766 handleExit ex@(ProcessExitException from msg) =
767 if messageFingerprint msg == fingerprint (undefined :: a)
768 then exitHandler from decoded
772 -- Make sure the value is fully decoded so that we don't hang to
773 -- bytestrings if the caller doesn't use the value immediately
774 !decoded = decode (messageEncoding msg)
776-- | Lift 'Control.Exception.catches' (almost).
778-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
779-- message, a handler must accept inputs of the expected type. In order to
780-- handle a list of potentially different handlers (and therefore input types),
781-- a handler passed to 'catchesExit' must accept 'Message' and return
782-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
784-- See 'maybeHandleMessage' and 'Message' for more details.
785catchesExit :: Process b
786 -> [(ProcessId -> Message -> (Process (Maybe b)))]
788catchesExit act handlers = Catch.catch act ((flip handleExit) handlers)
790 handleExit :: ProcessExitException
791 -> [(ProcessId -> Message -> Process (Maybe b))]
793 handleExit ex [] = throwM ex
794 handleExit ex@(ProcessExitException from msg) (h:hs) = do
797 Nothing -> handleExit ex hs
800-- | Our own process ID
801getSelfPid :: Process ProcessId
802getSelfPid = processId <$> ask
804-- | Get the node ID of our local node
805getSelfNode :: Process NodeId
806getSelfNode = localNodeId . processNode <$> ask
809-- | Get statistics about the specified node
810getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
812 selfNode <- getSelfNode
814 then Right `fmap` getLocalNodeStats -- optimisation
815 else getNodeStatsRemote selfNode
817 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
818 getNodeStatsRemote selfNode = do
819 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
820 bracket (monitorNode nid) unmonitor $ \mRef ->
821 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
822 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
823 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
826-- | Get statistics about our local node
827getLocalNodeStats :: Process NodeStats
828getLocalNodeStats = do
830 sendCtrlMsg Nothing $ GetNodeStats self
831 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
833-- | Get information about the specified process
834getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
836 let them = processNodeId pid in do
838 dest <- mkNode them us
839 sendCtrlMsg dest $ GetInfo pid
841 match (\(p :: ProcessInfo) -> return $ Just p)
842 , match (\(_ :: ProcessInfoNone) -> return Nothing)
844 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
845 mkNode them us = case them == us of
846 True -> return Nothing
847 _ -> return $ Just them
849--------------------------------------------------------------------------------
850-- Monitoring and linking --
851--------------------------------------------------------------------------------
853-- | Link to a remote process (asynchronous)
855-- When process A links to process B (that is, process A calls
856-- @link pidB@) then an asynchronous exception will be thrown to process A
857-- when process B terminates (normally or abnormally), or when process A gets
858-- disconnected from process B. Although it is /technically/ possible to catch
859-- these exceptions, chances are if you find yourself trying to do so you should
860-- probably be using 'monitor' rather than 'link'. In particular, code such as
862-- > link pidB -- Link to process B
863-- > expect -- Wait for a message from process B
864-- > unlink pidB -- Unlink again
866-- doesn't quite do what one might expect: if process B sends a message to
867-- process A, and /subsequently terminates/, then process A might or might not
868-- be terminated too, depending on whether the exception is thrown before or
869-- after the 'unlink' (i.e., this code has a race condition).
871-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
872-- call to 'link' has no effect.
874-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
875-- Linking makes no distinction between normal and abnormal termination of
876-- the remote process.
877link :: ProcessId -> Process ()
878link = sendCtrlMsg Nothing . Link . ProcessIdentifier
880-- | Monitor another process (asynchronous)
882-- When process A monitors process B (that is, process A calls
883-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
884-- when process B terminates (normally or abnormally), or when process A gets
885-- disconnected from process B. You receive this message like any other (using
886-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
887-- 'DiedDisconnect', etc.).
889-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
890-- multiple monitors are set up, multiple notifications will be delivered
891-- and monitors can be disabled individually using 'unmonitor'.
892monitor :: ProcessId -> Process MonitorRef
893monitor = monitor' . ProcessIdentifier
895-- | Establishes temporary monitoring of another process.
897-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
898-- of @code@. Note: although monitoring is no longer active when
899-- @withMonitor@ returns, there might still be unreceived monitor
900-- messages in the queue.
902withMonitor :: ProcessId -> (MonitorRef -> Process a) -> Process a
903withMonitor pid = bracket (monitor pid) unmonitor
905-- | Establishes temporary monitoring of another process.
907-- @withMonitor_ pid code@ sets up monitoring of @pid@ for the duration
908-- of @code@. Note: although monitoring is no longer active when
909-- @withMonitor_@ returns, there might still be unreceived monitor
910-- messages in the queue.
913withMonitor_ :: ProcessId -> Process a -> Process a
914withMonitor_ p = withMonitor p . const
918-- This is synchronous in the sense that once it returns you are guaranteed
919-- that no exception will be raised if the remote process dies. However, it is
920-- asynchronous in the sense that we do not wait for a response from the remote
922unlink :: ProcessId -> Process ()
925 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
929-- | Remove a node link
931-- This has the same synchronous/asynchronous nature as 'unlink'.
932unlinkNode :: NodeId -> Process ()
935 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
939-- | Remove a channel (send port) link
941-- This has the same synchronous/asynchronous nature as 'unlink'.
942unlinkPort :: SendPort a -> Process ()
944 unlinkPortAsync sport
945 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
951-- This has the same synchronous/asynchronous nature as 'unlink'.
953-- ProcessMonitorNotification messages for the given MonitorRef are removed from
955unmonitor :: MonitorRef -> Process ()
958 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
961 -- Discard the notification if any. With the current NC implementation at most
962 -- one notification is in the mailbox for any given ref.
963 void $ receiveTimeout 0
964 [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref' == ref)
968--------------------------------------------------------------------------------
969-- Exception handling --
970--------------------------------------------------------------------------------
972-- | Lift 'Control.Exception.catch'
973catch :: Exception e => Process a -> (e -> Process a) -> Process a
975{-# DEPRECATED catch "Use Control.Monad.Catch.catch instead" #-}
977-- | Lift 'Control.Exception.try'
978try :: Exception e => Process a -> Process (Either e a)
980{-# DEPRECATED try "Use Control.Monad.Catch.try instead" #-}
982-- | Lift 'Control.Exception.mask'
983mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
985{-# DEPRECATED mask "Use Control.Monad.Catch.mask_ instead" #-}
987-- | Lift 'Control.Exception.mask_'
988mask_ :: Process a -> Process a
990{-# DEPRECATED mask_ "Use Control.Monad.Catch.mask_ instead" #-}
992-- | Lift 'Control.Exception.onException'
993onException :: Process a -> Process b -> Process a
994onException = Catch.onException
995{-# DEPRECATED onException "Use Control.Monad.Catch.onException instead" #-}
997-- | Lift 'Control.Exception.bracket'
998bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
999bracket = Catch.bracket
1000{-# DEPRECATED bracket "Use Control.Monad.Catch.bracket instead" #-}
1002-- | Lift 'Control.Exception.bracket_'
1003bracket_ :: Process a -> Process b -> Process c -> Process c
1004bracket_ = Catch.bracket_
1005{-# DEPRECATED bracket_ "Use Control.Monad.Catch.bracket_ instead" #-}
1007-- | Lift 'Control.Exception.finally'
1008finally :: Process a -> Process b -> Process a
1009finally = Catch.finally
1010{-# DEPRECATED finally "Use Control.Monad.Catch.finally instead" #-}
1012-- | You need this when using 'catches'
1013data Handler a = forall e . Exception e => Handler (e -> Process a)
1015instance Functor Handler where
1016 fmap f (Handler h) = Handler (fmap f . h)
1018-- | Lift 'Control.Exception.catches'
1019catches :: Process a -> [Handler a] -> Process a
1020catches proc handlers = proc `Catch.catch` catchesHandler handlers
1022catchesHandler :: [Handler a] -> SomeException -> Process a
1023catchesHandler handlers e = foldr tryHandler (throwM e) handlers
1024 where tryHandler (Handler handler) res
1025 = case fromException e of
1026 Just e' -> handler e'
1029--------------------------------------------------------------------------------
1031--------------------------------------------------------------------------------
1033-- | Like 'expect' but with a timeout
1034expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
1035expectTimeout n = receiveTimeout n [match return]
1037-- | Asynchronous version of 'spawn'
1039-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
1040spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
1041spawnAsync nid proc = do
1042 spawnRef <- getSpawnRef
1045 then sendCtrlMsg Nothing $ Spawn proc spawnRef
1046 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
1049-- | Monitor a node (asynchronous)
1050monitorNode :: NodeId -> Process MonitorRef
1052 monitor' . NodeIdentifier
1054-- | Monitor a typed channel (asynchronous)
1055monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1056monitorPort (SendPort cid) =
1057 monitor' (SendPortIdentifier cid)
1059-- | Remove a monitor (asynchronous)
1060unmonitorAsync :: MonitorRef -> Process ()
1062 sendCtrlMsg Nothing . Unmonitor
1064-- | Link to a node (asynchronous)
1065linkNode :: NodeId -> Process ()
1066linkNode = link' . NodeIdentifier
1068-- | Link to a channel (asynchronous)
1069linkPort :: SendPort a -> Process ()
1070linkPort (SendPort cid) =
1071 link' (SendPortIdentifier cid)
1073-- | Remove a process link (asynchronous)
1074unlinkAsync :: ProcessId -> Process ()
1076 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1078-- | Remove a node link (asynchronous)
1079unlinkNodeAsync :: NodeId -> Process ()
1081 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1083-- | Remove a channel (send port) link (asynchronous)
1084unlinkPortAsync :: SendPort a -> Process ()
1085unlinkPortAsync (SendPort cid) =
1086 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1088--------------------------------------------------------------------------------
1090--------------------------------------------------------------------------------
1092data SayMessage = SayMessage { sayTime :: UTCTime
1093 , sayProcess :: ProcessId
1094 , sayMessage :: String }
1097-- There is sadly no Show UTCTime instance
1098instance Show SayMessage where
1101 $ showString "SayMessage "
1102 . showString (formatTime defaultTimeLocale "%c" (sayTime msg))
1104 . showsPrec 11 (sayProcess msg) . showChar ' '
1105 . showsPrec 11 (sayMessage msg) . showChar ' '
1107instance Binary SayMessage where
1109 putUTCTime (sayTime s)
1112 get = SayMessage <$> getUTCTime <*> get <*> get
1114-- Sadly there is no Binary UTCTime instance
1115putUTCTime :: UTCTime -> Put
1116putUTCTime (UTCTime (ModifiedJulianDay day) tod) = do
1118 put (toRational tod)
1120getUTCTime :: Get UTCTime
1124 return $! UTCTime (ModifiedJulianDay day)
1129-- @say message@ sends a message of type 'SayMessage' with the current time and
1130-- 'ProcessId' of the current process to the process registered as @logger@. By
1131-- default, this process simply sends the string to @stderr@. Individual Cloud
1132-- Haskell backends might replace this with a different logger process, however.
1133say :: String -> Process ()
1135 now <- liftIO getCurrentTime
1137 nsend "logger" (SayMessage now us string)
1139--------------------------------------------------------------------------------
1141--------------------------------------------------------------------------------
1143-- | Register a process with the local registry (synchronous). The name must not
1144-- already be registered. The process need not be on this node. A bad
1145-- registration will result in a 'ProcessRegistrationException'
1147-- The process to be registered does not have to be local itself.
1148register :: String -> ProcessId -> Process ()
1149register = registerImpl False
1151-- | Like 'register', but will replace an existing registration.
1152-- The name must already be registered.
1153reregister :: String -> ProcessId -> Process ()
1154reregister = registerImpl True
1156registerImpl :: Bool -> String -> ProcessId -> Process ()
1157registerImpl force label pid = do
1158 mynid <- getSelfNode
1159 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1161 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1162 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1165-- | Register a process with a remote registry (asynchronous).
1167-- The process to be registered does not have to live on the same remote node.
1168-- Reply wil come in the form of a 'RegisterReply' message
1170-- See comments in 'whereisRemoteAsync'
1171registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1172registerRemoteAsync nid label pid = do
1174 sendCtrlMsg (if nid == here then Nothing else Just nid)
1175 (Register label nid (Just pid) False)
1177reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1178reregisterRemoteAsync nid label pid =
1179 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1181-- | Remove a process from the local registry (asynchronous).
1182-- This version will wait until a response is gotten from the
1183-- management process. The name must already be registered.
1184unregister :: String -> Process ()
1185unregister label = do
1186 mynid <- getSelfNode
1187 sendCtrlMsg Nothing (Register label mynid Nothing False)
1189 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1190 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1193-- | Deal with the result from an attempted registration or unregistration
1194-- by throwing an exception if necessary
1195handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()
1196handleRegistrationReply label ok owner =
1198 throwM $ ProcessRegistrationException label owner
1200-- | Remove a process from a remote registry (asynchronous).
1202-- Reply wil come in the form of a 'RegisterReply' message
1204-- See comments in 'whereisRemoteAsync'
1205unregisterRemoteAsync :: NodeId -> String -> Process ()
1206unregisterRemoteAsync nid label =
1207 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1209-- | Query the local process registry
1210whereis :: String -> Process (Maybe ProcessId)
1212 sendCtrlMsg Nothing (WhereIs label)
1213 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1214 (\(WhereIsReply _ mPid) -> return mPid)
1217-- | Query a remote process registry (asynchronous)
1219-- Reply will come in the form of a 'WhereIsReply' message.
1221-- There is currently no synchronous version of 'whereisRemoteAsync': if
1222-- you implement one yourself, be sure to take into account that the remote
1223-- node might die or get disconnect before it can respond (i.e. you should
1224-- use 'monitorNode' and take appropriate action when you receive a
1225-- 'NodeMonitorNotification').
1226whereisRemoteAsync :: NodeId -> String -> Process ()
1227whereisRemoteAsync nid label = do
1229 sendCtrlMsg (if nid == here then Nothing else Just nid) (WhereIs label)
1231-- | Named send to a process in the local registry (asynchronous)
1232nsend :: Serializable a => String -> a -> Process ()
1234 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1236-- | Named send to a process in the local registry (asynchronous).
1237-- This function makes /no/ attempt to serialize and (in the case when the
1238-- destination process resides on the same local node) therefore ensure that
1239-- the payload is fully evaluated before it is delivered.
1240unsafeNSend :: Serializable a => String -> a -> Process ()
1241unsafeNSend = Unsafe.nsend
1243-- | Named send to a process in a remote registry (asynchronous)
1244nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1245nsendRemote nid label msg = do
1247 if here == nid then nsend label msg
1248 else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1250-- | Named send to a process in a remote registry (asynchronous)
1251-- This function makes /no/ attempt to serialize and (in the case when the
1252-- destination process resides on the same local node) therefore ensure that
1253-- the payload is fully evaluated before it is delivered.
1254unsafeNSendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1255unsafeNSendRemote = Unsafe.nsendRemote
1257--------------------------------------------------------------------------------
1259--------------------------------------------------------------------------------
1261-- | Resolve a static value
1262unStatic :: Typeable a => Static a -> Process a
1264 rtable <- remoteTable . processNode <$> ask
1265 case Static.unstatic rtable static of
1266 Left err -> fail $ "Could not resolve static value: " ++ err
1269-- | Resolve a closure
1270unClosure :: Typeable a => Closure a -> Process a
1271unClosure closure = do
1272 rtable <- remoteTable . processNode <$> ask
1273 case Static.unclosure rtable closure of
1274 Left err -> fail $ "Could not resolve closure: " ++ err
1277--------------------------------------------------------------------------------
1279--------------------------------------------------------------------------------
1281-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1282-- message passing. However, when network connections get disrupted this
1283-- illusion cannot always be maintained. Once a network connection breaks (even
1284-- temporarily) no further communication on that connection will be possible.
1285-- For example, if process A sends a message to process B, and A is then
1286-- notified (by monitor notification) that it got disconnected from B, A will
1287-- not be able to send any further messages to B, /unless/ A explicitly
1288-- indicates that it is acceptable to attempt to reconnect to B using the
1289-- Cloud Haskell 'reconnect' primitive.
1291-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1292-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1293-- then receives a monitor notification that its connection to B has been lost,
1294-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1295-- and m3 but not m2.
1297-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1298-- to attempt to reconnect on the next send/. In particular, if no further
1299-- communication attempts are made to B then A can use reconnect to clean up
1300-- its connection to B.
1301reconnect :: ProcessId -> Process ()
1304 node <- processNode <$> ask
1305 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1307-- | Reconnect to a sendport. See 'reconnect' for more information.
1308reconnectPort :: SendPort a -> Process ()
1309reconnectPort them = do
1311 node <- processNode <$> ask
1312 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1314--------------------------------------------------------------------------------
1315-- Auxiliary functions --
1316--------------------------------------------------------------------------------
1318sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1320 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1322sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1323sendChanLocal spId msg =
1324 -- we *must* fully serialize/encode the message here, because
1325 -- attempting to use `unsafeCoerce' in the node controller
1326 -- won't work since we know nothing about the required type
1327 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1329getMonitorRefFor :: Identifier -> Process MonitorRef
1330getMonitorRefFor ident = do
1332 liftIO $ modifyMVar (processState proc) $ \st -> do
1333 let counter = st ^. monitorCounter
1334 return ( monitorCounter ^: (+ 1) $ st
1335 , MonitorRef ident counter
1338getSpawnRef :: Process SpawnRef
1341 liftIO $ modifyMVar (processState proc) $ \st -> do
1342 let counter = st ^. spawnCounter
1343 return ( spawnCounter ^: (+ 1) $ st
1347-- | Monitor a process/node/channel
1348monitor' :: Identifier -> Process MonitorRef
1350 monitorRef <- getMonitorRefFor ident
1351 sendCtrlMsg Nothing $ Monitor monitorRef
1354-- | Link to a process/node/channel
1355link' :: Identifier -> Process ()
1356link' = 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