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
310newChan :: Serializable a => Process (SendPort a, ReceivePort a)
313 liftIO . modifyMVar (processState proc) $ \st -> do
314 let lcid = st ^. channelCounter
315 let cid = SendPortId { sendPortProcessId = processId proc
316 , sendPortLocalId = lcid
318 let sport = SendPort cid
319 chan <- liftIO newTQueueIO
320 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
321 let rport = ReceivePort $ readTQueue chan
322 let tch = TypedChannel chan'
323 return ( (channelCounter ^: (+ 1))
324 . (typedChannelWithId lcid ^= Just tch)
329 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
330 finalizer st lcid = modifyMVar_ st $
331 return . (typedChannelWithId lcid ^= Nothing)
333-- | Send a message on a typed channel
334sendChan :: Serializable a => SendPort a -> a -> Process ()
335sendChan (SendPort cid) msg = do
337 let node = localNodeId (processNode proc)
338 destNode = processNodeId (sendPortProcessId cid) in do
339 case destNode == node of
340 True -> sendChanLocal cid msg
342 liftIO $ sendBinary (processNode proc)
343 (ProcessIdentifier (processId proc))
344 (SendPortIdentifier cid)
348-- | Send a message on a typed channel. This function makes /no/ attempt to
349-- serialize and (in the case when the @ReceivePort@ resides on the same local
350-- node) therefore ensure that the payload is fully evaluated before it is
352unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
353unsafeSendChan = Unsafe.sendChan
355-- | Wait for a message on a typed channel
356receiveChan :: Serializable a => ReceivePort a -> Process a
357receiveChan = liftIO . atomically . receiveSTM
359-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
360-- non-blocking check for a message.
361receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
362receiveChanTimeout 0 ch = liftIO . atomically $
363 (Just <$> receiveSTM ch) `orElse` return Nothing
364receiveChanTimeout n ch = liftIO . timeout n . atomically $
367-- | Merge a list of typed channels.
369-- The result port is left-biased: if there are messages available on more
370-- than one port, the first available message is returned.
371mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
372mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
374-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
376mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
377mergePortsRR = \ps -> do
378 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
379 return $ ReceivePort (rr psVar)
383 rotate (x:xs) = xs ++ [x]
385 rr :: TVar [STM a] -> STM a
388 a <- foldr1 orElse ps
389 writeTVar psVar (rotate ps)
392--------------------------------------------------------------------------------
393-- Advanced messaging --
394--------------------------------------------------------------------------------
396-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
397newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
400-- | Test the matches in order against each message in the queue
401receiveWait :: [Match b] -> Process b
403 queue <- processQueue <$> ask
404 mProc <- liftIO $ dequeue queue Blocking (map unMatch ms)
407 Nothing -> die $ "System Invariant Violation: CQueue.hs returned `Nothing` "
408 ++ "in the absence of a timeout value."
410-- | Like 'receiveWait' but with a timeout.
412-- If the timeout is zero do a non-blocking check for matching messages. A
413-- non-zero timeout is applied only when waiting for incoming messages (that is,
414-- /after/ we have checked the messages that are already in the mailbox).
415receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
416receiveTimeout t ms = do
417 queue <- processQueue <$> ask
418 let blockSpec = if t == 0 then NonBlocking else Timeout t
419 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
421 Nothing -> return Nothing
422 Just proc -> Just <$> proc
424-- | Match on a typed channel
425matchChan :: ReceivePort a -> (a -> Process b) -> Match b
426matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
428-- | Match on an arbitrary STM action.
430-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
431-- actions with checks against our process' mailbox and/or any typed channel
432-- @ReceivePort@s we may hold.
434-- This allows us to process multiple input streams /along with our mailbox/,
435-- in just the same way that 'matchChan' supports checking both the mailbox
436-- /and/ an arbitrary set of typed channels in one atomic transaction.
438-- Note there are no ordering guarnatees with respect to these disparate input
441matchSTM :: STM a -> (a -> Process b) -> Match b
442matchSTM stm fn = Match $ MatchChan (fmap fn stm)
444-- | Match against any message of the right type
445match :: forall a b. Serializable a => (a -> Process b) -> Match b
446match = matchIf (const True)
448-- | Match against any message of the right type that satisfies a predicate
449matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
450matchIf c p = Match $ MatchMsg $ \msg ->
451 case messageFingerprint msg == fingerprint (undefined :: a) of
454 (UnencodedMessage _ m) ->
455 let m' = unsafeCoerce m :: a in
459 (EncodedMessage _ _) ->
460 if (c decoded) then Just (p decoded) else Nothing
463 -- Make sure the value is fully decoded so that we don't hang to
464 -- bytestrings when the process calling 'matchIf' doesn't process
465 -- the values immediately
466 !decoded = decode (messageEncoding msg)
468-- | Match against any message, regardless of the underlying (contained) type
469matchMessage :: (Message -> Process Message) -> Match Message
470matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
472-- | Match against any message (regardless of underlying type) that satisfies a predicate
473matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
474matchMessageIf c p = Match $ MatchMsg $ \msg ->
479-- | Forward a raw 'Message' to the given 'ProcessId'.
480forward :: Message -> ProcessId -> Process ()
483 let node = processNode proc
485 nid = localNodeId node
486 destNode = (processNodeId them) in do
487 case destNode == nid of
488 True -> sendCtrlMsg Nothing (LocalSend them msg)
491 sendPayload (processNode proc) (ProcessIdentifier (processId proc)) (ProcessIdentifier them) NoImplicitReconnect (messageToPayload msg)
492 -- We do not fire the trace event until after the sending is complete;
493 -- In this remote case, 'sendMessage' can block in the networking stack.
494 liftIO $ traceEvent (localEventBus node)
497-- | Forward a raw 'Message' to the given 'ProcessId'.
499-- Unlike 'forward', this function is insensitive to 'reconnect'. It will
500-- try to send the message regardless of the history of connection failures
502uforward :: Message -> ProcessId -> Process ()
503uforward msg them = do
505 let node = processNode proc
507 nid = localNodeId node
508 destNode = (processNodeId them) in do
509 case destNode == nid of
510 True -> sendCtrlMsg Nothing (LocalSend them msg)
512 sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them) msg
513 -- We do not fire the trace event until after the sending is complete;
514 -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
515 liftIO $ traceEvent (localEventBus node)
518-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
519-- 'Serializable' - like the datum they contain - but also note, deserialising
520-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
521-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
526-- send self (wrapMessage "blah")
527-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
528-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
529-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
532wrapMessage :: Serializable a => a -> Message
533wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
535-- | This is the /unsafe/ variant of 'wrapMessage'. See
536-- "Control.Distributed.Process.UnsafePrimitives" for details.
537unsafeWrapMessage :: Serializable a => a -> Message
538unsafeWrapMessage = Unsafe.wrapMessage
540-- [note Serializable UnencodedMessage]
541-- if we attempt to serialise an UnencodedMessage, it will be converted to an
542-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
543-- Primitives.hs for the gory details
545-- | Attempt to unwrap a raw 'Message'.
546-- If the type of the decoded message payload matches the expected type, the
547-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
550-- This expression, for example, will evaluate to @Nothing@
551-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
553-- Whereas this expression, will yield @Just "foo"@
554-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
556unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
558 case messageFingerprint msg == fingerprint (undefined :: a) of
559 False -> return Nothing :: m (Maybe a)
561 (UnencodedMessage _ ms) ->
562 let ms' = unsafeCoerce ms :: a
564 (EncodedMessage _ _) ->
565 return (Just (decoded))
567 decoded :: a -- note [decoding]
568 !decoded = decode (messageEncoding msg)
570-- | Attempt to handle a raw 'Message'.
571-- If the type of the message matches the type of the first argument to
572-- the supplied expression, then the message will be decoded and the expression
573-- evaluated against its value. If this runtime type checking fails however,
574-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
575-- evaluation proceeds, the resulting value with be wrapped with @Just@.
577-- Intended for use in `catchesExit` and `matchAny` primitives.
579handleMessage :: forall m a b. (Monad m, Serializable a)
580 => Message -> (a -> m b) -> m (Maybe b)
581handleMessage msg proc = handleMessageIf msg (const True) proc
583-- | Conditionally handle a raw 'Message'.
584-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
585-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
586-- for further information about runtime type checking.
588handleMessageIf :: forall m a b . (Monad m, Serializable a)
593handleMessageIf msg c proc = do
594 case messageFingerprint msg == fingerprint (undefined :: a) of
595 False -> return Nothing :: m (Maybe b)
597 (UnencodedMessage _ ms) ->
598 let ms' = unsafeCoerce ms :: a in
600 True -> do { r <- proc ms'; return (Just r) }
601 False -> return Nothing :: m (Maybe b)
602 (EncodedMessage _ _) ->
604 True -> do { r <- proc (decoded :: a); return (Just r) }
605 False -> return Nothing :: m (Maybe b)
607 decoded :: a -- note [decoding]
608 !decoded = decode (messageEncoding msg)
610-- | As 'handleMessage' but ignores result, which is useful if you don't
611-- care whether or not the handler succeeded.
613handleMessage_ :: forall m a . (Monad m, Serializable a)
614 => Message -> (a -> m ()) -> m ()
615handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
617-- | Conditional version of 'handleMessage_'.
619handleMessageIf_ :: forall m a . (Monad m, Serializable a)
624handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
626-- | Match against an arbitrary message. 'matchAny' removes the first available
627-- message from the process mailbox. To handle arbitrary /raw/ messages once
628-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
630matchAny :: forall b. (Message -> Process b) -> Match b
631matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
633-- | Match against an arbitrary message. Intended for use with 'handleMessage'
634-- and 'unwrapMessage', this function /only/ removes a message from the process
635-- mailbox, /if/ the supplied condition matches. The success (or failure) of
636-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
637-- here, i.e., if the condition evaluates to @True@ then the message will
638-- be removed from the process mailbox and decoded, but that does /not/
639-- guarantee that an expression passed to @handleMessage@ will pass the
640-- runtime type checks and therefore be evaluated.
642matchAnyIf :: forall a b. (Serializable a)
644 -> (Message -> Process b)
646matchAnyIf c p = Match $ MatchMsg $ \msg ->
647 case messageFingerprint msg == fingerprint (undefined :: a) of
648 True | check -> Just (p msg)
653 (EncodedMessage _ _) -> c decoded
654 (UnencodedMessage _ m') -> c (unsafeCoerce m')
656 decoded :: a -- note [decoding]
657 !decoded = decode (messageEncoding msg)
661For an EncodedMessage, we need to ensure the value is fully decoded so that
662we don't hang to bytestrings if the calling process doesn't evaluate
663immediately. For UnencodedMessage we know (because the fingerprint comparison
664succeeds) that unsafeCoerce will not fail.
667-- | Remove any message from the queue
668matchUnknown :: Process b -> Match b
669matchUnknown p = Match $ MatchMsg (const (Just p))
671-- | Receives messages and forwards them to @pid@ if @p msg == True@.
672delegate :: ProcessId -> (Message -> Bool) -> Process ()
675 matchAny (\m -> case (p m) of
676 True -> forward m pid
681-- | A straight relay that forwards all messages to the supplied pid.
682relay :: ProcessId -> Process ()
683relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
685-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
686-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
687-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
688-- fails, no action is taken and the proxy process will continue running.
689proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
693 next <- handleMessage m proc
695 Just True -> forward m pid
696 Just False -> return () -- explicitly ignored
697 Nothing -> return ()) -- un-routable
701--------------------------------------------------------------------------------
702-- Process management --
703--------------------------------------------------------------------------------
705-- | Thrown by 'terminate'
706data ProcessTerminationException = ProcessTerminationException
707 deriving (Show, Typeable)
709instance Exception ProcessTerminationException
711-- | Terminate immediately (throws a ProcessTerminationException)
712terminate :: Process a
713terminate = throwM ProcessTerminationException
716-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
717die :: Serializable a => a -> Process b
719-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
720-- could be decoded by a handler passed to 'catchExit', re-thrown or even
721-- passed to another process via the tracing mechanism.
723 throwM (ProcessExitException pid (createMessage reason))
725-- | Forceful request to kill a process. Where 'exit' provides an exception
726-- that can be caught and handled, 'kill' throws an unexposed exception type
727-- which cannot be handled explicitly (by type).
728kill :: ProcessId -> String -> Process ()
729-- NOTE: We send the message to our local node controller, which will then
730-- forward it to a remote node controller (if applicable). Sending it directly
731-- to a remote node controller means that that the message may overtake a
732-- 'monitor' or 'link' request.
733kill them reason = sendCtrlMsg Nothing (Kill them reason)
735-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
736-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
737-- manner can be handled using the 'catchExit' family of functions.
738exit :: Serializable a => ProcessId -> a -> Process ()
739-- NOTE: We send the message to our local node controller, which will then
740-- forward it to a remote node controller (if applicable). Sending it directly
741-- to a remote node controller means that that the message may overtake a
742-- 'monitor' or 'link' request.
743exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
745-- | Catches 'ProcessExitException'. The handler will not be applied unless its
746-- type matches the encoded data stored in the exception (see the /reason/
747-- argument given to the 'exit' primitive). If the handler cannot be applied,
748-- the exception will be re-thrown.
750-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
751-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
752catchExit :: forall a b . (Show a, Serializable a)
754 -> (ProcessId -> a -> Process b)
756catchExit act exitHandler = Catch.catch act handleExit
758 handleExit ex@(ProcessExitException from msg) =
759 if messageFingerprint msg == fingerprint (undefined :: a)
760 then exitHandler from decoded
764 -- Make sure the value is fully decoded so that we don't hang to
765 -- bytestrings if the caller doesn't use the value immediately
766 !decoded = decode (messageEncoding msg)
768-- | Lift 'Control.Exception.catches' (almost).
770-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
771-- message, a handler must accept inputs of the expected type. In order to
772-- handle a list of potentially different handlers (and therefore input types),
773-- a handler passed to 'catchesExit' must accept 'Message' and return
774-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
776-- See 'maybeHandleMessage' and 'Message' for more details.
777catchesExit :: Process b
778 -> [(ProcessId -> Message -> (Process (Maybe b)))]
780catchesExit act handlers = Catch.catch act ((flip handleExit) handlers)
782 handleExit :: ProcessExitException
783 -> [(ProcessId -> Message -> Process (Maybe b))]
785 handleExit ex [] = throwM ex
786 handleExit ex@(ProcessExitException from msg) (h:hs) = do
789 Nothing -> handleExit ex hs
792-- | Our own process ID
793getSelfPid :: Process ProcessId
794getSelfPid = processId <$> ask
796-- | Get the node ID of our local node
797getSelfNode :: Process NodeId
798getSelfNode = localNodeId . processNode <$> ask
801-- | Get statistics about the specified node
802getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
804 selfNode <- getSelfNode
806 then Right `fmap` getLocalNodeStats -- optimisation
807 else getNodeStatsRemote selfNode
809 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
810 getNodeStatsRemote selfNode = do
811 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
812 bracket (monitorNode nid) unmonitor $ \mRef ->
813 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
814 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
815 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
818-- | Get statistics about our local node
819getLocalNodeStats :: Process NodeStats
820getLocalNodeStats = do
822 sendCtrlMsg Nothing $ GetNodeStats self
823 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
825-- | Get information about the specified process
826getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
828 let them = processNodeId pid in do
830 dest <- mkNode them us
831 sendCtrlMsg dest $ GetInfo pid
833 match (\(p :: ProcessInfo) -> return $ Just p)
834 , match (\(_ :: ProcessInfoNone) -> return Nothing)
836 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
837 mkNode them us = case them == us of
838 True -> return Nothing
839 _ -> return $ Just them
841--------------------------------------------------------------------------------
842-- Monitoring and linking --
843--------------------------------------------------------------------------------
845-- | Link to a remote process (asynchronous)
847-- When process A links to process B (that is, process A calls
848-- @link pidB@) then an asynchronous exception will be thrown to process A
849-- when process B terminates (normally or abnormally), or when process A gets
850-- disconnected from process B. Although it is /technically/ possible to catch
851-- these exceptions, chances are if you find yourself trying to do so you should
852-- probably be using 'monitor' rather than 'link'. In particular, code such as
854-- > link pidB -- Link to process B
855-- > expect -- Wait for a message from process B
856-- > unlink pidB -- Unlink again
858-- doesn't quite do what one might expect: if process B sends a message to
859-- process A, and /subsequently terminates/, then process A might or might not
860-- be terminated too, depending on whether the exception is thrown before or
861-- after the 'unlink' (i.e., this code has a race condition).
863-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
864-- call to 'link' has no effect.
866-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
867-- Linking makes no distinction between normal and abnormal termination of
868-- the remote process.
869link :: ProcessId -> Process ()
870link = sendCtrlMsg Nothing . Link . ProcessIdentifier
872-- | Monitor another process (asynchronous)
874-- When process A monitors process B (that is, process A calls
875-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
876-- when process B terminates (normally or abnormally), or when process A gets
877-- disconnected from process B. You receive this message like any other (using
878-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
879-- 'DiedDisconnect', etc.).
881-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
882-- multiple monitors are set up, multiple notifications will be delivered
883-- and monitors can be disabled individually using 'unmonitor'.
884monitor :: ProcessId -> Process MonitorRef
885monitor = monitor' . ProcessIdentifier
887-- | Establishes temporary monitoring of another process.
889-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
890-- of @code@. Note: although monitoring is no longer active when
891-- @withMonitor@ returns, there might still be unreceived monitor
892-- messages in the queue.
894withMonitor :: ProcessId -> (MonitorRef -> Process a) -> Process a
895withMonitor pid = bracket (monitor pid) unmonitor
897-- | Establishes temporary monitoring of another process.
899-- @withMonitor_ pid code@ sets up monitoring of @pid@ for the duration
900-- of @code@. Note: although monitoring is no longer active when
901-- @withMonitor_@ returns, there might still be unreceived monitor
902-- messages in the queue.
905withMonitor_ :: ProcessId -> Process a -> Process a
906withMonitor_ p = withMonitor p . const
910-- This is synchronous in the sense that once it returns you are guaranteed
911-- that no exception will be raised if the remote process dies. However, it is
912-- asynchronous in the sense that we do not wait for a response from the remote
914unlink :: ProcessId -> Process ()
917 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
921-- | Remove a node link
923-- This has the same synchronous/asynchronous nature as 'unlink'.
924unlinkNode :: NodeId -> Process ()
927 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
931-- | Remove a channel (send port) link
933-- This has the same synchronous/asynchronous nature as 'unlink'.
934unlinkPort :: SendPort a -> Process ()
936 unlinkPortAsync sport
937 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
943-- This has the same synchronous/asynchronous nature as 'unlink'.
945-- ProcessMonitorNotification messages for the given MonitorRef are removed from
947unmonitor :: MonitorRef -> Process ()
950 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
953 -- Discard the notification if any. With the current NC implementation at most
954 -- one notification is in the mailbox for any given ref.
955 void $ receiveTimeout 0
956 [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref' == ref)
960--------------------------------------------------------------------------------
961-- Exception handling --
962--------------------------------------------------------------------------------
964-- | Lift 'Control.Exception.catch'
965catch :: Exception e => Process a -> (e -> Process a) -> Process a
967{-# DEPRECATED catch "Use Control.Monad.Catch.catch instead" #-}
969-- | Lift 'Control.Exception.try'
970try :: Exception e => Process a -> Process (Either e a)
972{-# DEPRECATED try "Use Control.Monad.Catch.try instead" #-}
974-- | Lift 'Control.Exception.mask'
975mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
977{-# DEPRECATED mask "Use Control.Monad.Catch.mask_ instead" #-}
979-- | Lift 'Control.Exception.mask_'
980mask_ :: Process a -> Process a
982{-# DEPRECATED mask_ "Use Control.Monad.Catch.mask_ instead" #-}
984-- | Lift 'Control.Exception.onException'
985onException :: Process a -> Process b -> Process a
986onException = Catch.onException
987{-# DEPRECATED onException "Use Control.Monad.Catch.onException instead" #-}
989-- | Lift 'Control.Exception.bracket'
990bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
991bracket = Catch.bracket
992{-# DEPRECATED bracket "Use Control.Monad.Catch.bracket instead" #-}
994-- | Lift 'Control.Exception.bracket_'
995bracket_ :: Process a -> Process b -> Process c -> Process c
996bracket_ = Catch.bracket_
997{-# DEPRECATED bracket_ "Use Control.Monad.Catch.bracket_ instead" #-}
999-- | Lift 'Control.Exception.finally'
1000finally :: Process a -> Process b -> Process a
1001finally = Catch.finally
1002{-# DEPRECATED finally "Use Control.Monad.Catch.finally instead" #-}
1004-- | You need this when using 'catches'
1005data Handler a = forall e . Exception e => Handler (e -> Process a)
1007instance Functor Handler where
1008 fmap f (Handler h) = Handler (fmap f . h)
1010-- | Lift 'Control.Exception.catches'
1011catches :: Process a -> [Handler a] -> Process a
1012catches proc handlers = proc `Catch.catch` catchesHandler handlers
1014catchesHandler :: [Handler a] -> SomeException -> Process a
1015catchesHandler handlers e = foldr tryHandler (throwM e) handlers
1016 where tryHandler (Handler handler) res
1017 = case fromException e of
1018 Just e' -> handler e'
1021--------------------------------------------------------------------------------
1023--------------------------------------------------------------------------------
1025-- | Like 'expect' but with a timeout
1026expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
1027expectTimeout n = receiveTimeout n [match return]
1029-- | Asynchronous version of 'spawn'
1031-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
1032spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
1033spawnAsync nid proc = do
1034 spawnRef <- getSpawnRef
1037 then sendCtrlMsg Nothing $ Spawn proc spawnRef
1038 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
1041-- | Monitor a node (asynchronous)
1042monitorNode :: NodeId -> Process MonitorRef
1044 monitor' . NodeIdentifier
1046-- | Monitor a typed channel (asynchronous)
1047monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1048monitorPort (SendPort cid) =
1049 monitor' (SendPortIdentifier cid)
1051-- | Remove a monitor (asynchronous)
1052unmonitorAsync :: MonitorRef -> Process ()
1054 sendCtrlMsg Nothing . Unmonitor
1056-- | Link to a node (asynchronous)
1057linkNode :: NodeId -> Process ()
1058linkNode = link' . NodeIdentifier
1060-- | Link to a channel (asynchronous)
1061linkPort :: SendPort a -> Process ()
1062linkPort (SendPort cid) =
1063 link' (SendPortIdentifier cid)
1065-- | Remove a process link (asynchronous)
1066unlinkAsync :: ProcessId -> Process ()
1068 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1070-- | Remove a node link (asynchronous)
1071unlinkNodeAsync :: NodeId -> Process ()
1073 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1075-- | Remove a channel (send port) link (asynchronous)
1076unlinkPortAsync :: SendPort a -> Process ()
1077unlinkPortAsync (SendPort cid) =
1078 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1080--------------------------------------------------------------------------------
1082--------------------------------------------------------------------------------
1084data SayMessage = SayMessage { sayTime :: UTCTime
1085 , sayProcess :: ProcessId
1086 , sayMessage :: String }
1089-- There is sadly no Show UTCTime instance
1090instance Show SayMessage where
1093 $ showString "SayMessage "
1094 . showString (formatTime defaultTimeLocale "%c" (sayTime msg))
1096 . showsPrec 11 (sayProcess msg) . showChar ' '
1097 . showsPrec 11 (sayMessage msg) . showChar ' '
1099instance Binary SayMessage where
1101 putUTCTime (sayTime s)
1104 get = SayMessage <$> getUTCTime <*> get <*> get
1106-- Sadly there is no Binary UTCTime instance
1107putUTCTime :: UTCTime -> Put
1108putUTCTime (UTCTime (ModifiedJulianDay day) tod) = do
1110 put (toRational tod)
1112getUTCTime :: Get UTCTime
1116 return $! UTCTime (ModifiedJulianDay day)
1121-- @say message@ sends a message of type 'SayMessage' with the current time and
1122-- 'ProcessId' of the current process to the process registered as @logger@. By
1123-- default, this process simply sends the string to @stderr@. Individual Cloud
1124-- Haskell backends might replace this with a different logger process, however.
1125say :: String -> Process ()
1127 now <- liftIO getCurrentTime
1129 nsend "logger" (SayMessage now us string)
1131--------------------------------------------------------------------------------
1133--------------------------------------------------------------------------------
1135-- | Register a process with the local registry (synchronous). The name must not
1136-- already be registered. The process need not be on this node. A bad
1137-- registration will result in a 'ProcessRegistrationException'
1139-- The process to be registered does not have to be local itself.
1140register :: String -> ProcessId -> Process ()
1141register = registerImpl False
1143-- | Like 'register', but will replace an existing registration.
1144-- The name must already be registered.
1145reregister :: String -> ProcessId -> Process ()
1146reregister = registerImpl True
1148registerImpl :: Bool -> String -> ProcessId -> Process ()
1149registerImpl force label pid = do
1150 mynid <- getSelfNode
1151 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1153 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1154 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1157-- | Register a process with a remote registry (asynchronous).
1159-- The process to be registered does not have to live on the same remote node.
1160-- Reply wil come in the form of a 'RegisterReply' message
1162-- See comments in 'whereisRemoteAsync'
1163registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1164registerRemoteAsync nid label pid = do
1166 sendCtrlMsg (if nid == here then Nothing else Just nid)
1167 (Register label nid (Just pid) False)
1169reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1170reregisterRemoteAsync nid label pid =
1171 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1173-- | Remove a process from the local registry (asynchronous).
1174-- This version will wait until a response is gotten from the
1175-- management process. The name must already be registered.
1176unregister :: String -> Process ()
1177unregister label = do
1178 mynid <- getSelfNode
1179 sendCtrlMsg Nothing (Register label mynid Nothing False)
1181 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1182 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1185-- | Deal with the result from an attempted registration or unregistration
1186-- by throwing an exception if necessary
1187handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()
1188handleRegistrationReply label ok owner =
1190 throwM $ ProcessRegistrationException label owner
1192-- | Remove a process from a remote registry (asynchronous).
1194-- Reply wil come in the form of a 'RegisterReply' message
1196-- See comments in 'whereisRemoteAsync'
1197unregisterRemoteAsync :: NodeId -> String -> Process ()
1198unregisterRemoteAsync nid label =
1199 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1201-- | Query the local process registry
1202whereis :: String -> Process (Maybe ProcessId)
1204 sendCtrlMsg Nothing (WhereIs label)
1205 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1206 (\(WhereIsReply _ mPid) -> return mPid)
1209-- | Query a remote process registry (asynchronous)
1211-- Reply will come in the form of a 'WhereIsReply' message.
1213-- There is currently no synchronous version of 'whereisRemoteAsync': if
1214-- you implement one yourself, be sure to take into account that the remote
1215-- node might die or get disconnect before it can respond (i.e. you should
1216-- use 'monitorNode' and take appropriate action when you receive a
1217-- 'NodeMonitorNotification').
1218whereisRemoteAsync :: NodeId -> String -> Process ()
1219whereisRemoteAsync nid label = do
1221 sendCtrlMsg (if nid == here then Nothing else Just nid) (WhereIs label)
1223-- | Named send to a process in the local registry (asynchronous)
1224nsend :: Serializable a => String -> a -> Process ()
1226 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1228-- | Named send to a process in the local registry (asynchronous).
1229-- This function makes /no/ attempt to serialize and (in the case when the
1230-- destination process resides on the same local node) therefore ensure that
1231-- the payload is fully evaluated before it is delivered.
1232unsafeNSend :: Serializable a => String -> a -> Process ()
1233unsafeNSend = Unsafe.nsend
1235-- | Named send to a process in a remote registry (asynchronous)
1236nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1237nsendRemote nid label msg = do
1239 if here == nid then nsend label msg
1240 else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1242-- | Named send to a process in a remote registry (asynchronous)
1243-- This function makes /no/ attempt to serialize and (in the case when the
1244-- destination process resides on the same local node) therefore ensure that
1245-- the payload is fully evaluated before it is delivered.
1246unsafeNSendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1247unsafeNSendRemote = Unsafe.nsendRemote
1249--------------------------------------------------------------------------------
1251--------------------------------------------------------------------------------
1253-- | Resolve a static value
1254unStatic :: Typeable a => Static a -> Process a
1256 rtable <- remoteTable . processNode <$> ask
1257 case Static.unstatic rtable static of
1258 Left err -> fail $ "Could not resolve static value: " ++ err
1261-- | Resolve a closure
1262unClosure :: Typeable a => Closure a -> Process a
1263unClosure closure = do
1264 rtable <- remoteTable . processNode <$> ask
1265 case Static.unclosure rtable closure of
1266 Left err -> fail $ "Could not resolve closure: " ++ err
1269--------------------------------------------------------------------------------
1271--------------------------------------------------------------------------------
1273-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1274-- message passing. However, when network connections get disrupted this
1275-- illusion cannot always be maintained. Once a network connection breaks (even
1276-- temporarily) no further communication on that connection will be possible.
1277-- For example, if process A sends a message to process B, and A is then
1278-- notified (by monitor notification) that it got disconnected from B, A will
1279-- not be able to send any further messages to B, /unless/ A explicitly
1280-- indicates that it is acceptable to attempt to reconnect to B using the
1281-- Cloud Haskell 'reconnect' primitive.
1283-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1284-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1285-- then receives a monitor notification that its connection to B has been lost,
1286-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1287-- and m3 but not m2.
1289-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1290-- to attempt to reconnect on the next send/. In particular, if no further
1291-- communication attempts are made to B then A can use reconnect to clean up
1292-- its connection to B.
1293reconnect :: ProcessId -> Process ()
1296 node <- processNode <$> ask
1297 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1299-- | Reconnect to a sendport. See 'reconnect' for more information.
1300reconnectPort :: SendPort a -> Process ()
1301reconnectPort them = do
1303 node <- processNode <$> ask
1304 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1306--------------------------------------------------------------------------------
1307-- Auxiliary functions --
1308--------------------------------------------------------------------------------
1310sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1312 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1314sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1315sendChanLocal spId msg =
1316 -- we *must* fully serialize/encode the message here, because
1317 -- attempting to use `unsafeCoerce' in the node controller
1318 -- won't work since we know nothing about the required type
1319 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1321getMonitorRefFor :: Identifier -> Process MonitorRef
1322getMonitorRefFor ident = do
1324 liftIO $ modifyMVar (processState proc) $ \st -> do
1325 let counter = st ^. monitorCounter
1326 return ( monitorCounter ^: (+ 1) $ st
1327 , MonitorRef ident counter
1330getSpawnRef :: Process SpawnRef
1333 liftIO $ modifyMVar (processState proc) $ \st -> do
1334 let counter = st ^. spawnCounter
1335 return ( spawnCounter ^: (+ 1) $ st
1339-- | Monitor a process/node/channel
1340monitor' :: Identifier -> Process MonitorRef
1342 monitorRef <- getMonitorRefFor ident
1343 sendCtrlMsg Nothing $ Monitor monitorRef
1346-- | Link to a process/node/channel
1347link' :: Identifier -> Process ()
1348link' = 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