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
311-- Note that the channel is bound to the lifecycle of the process that evaluated
312-- this function, such that if the calling process dies/exits, the channel will
313-- no longer be accessible, and reading from the ReceivePort will fail.
315newChan :: Serializable a => Process (SendPort a, ReceivePort a)
318 liftIO . modifyMVar (processState proc) $ \st -> do
319 let lcid = st ^. channelCounter
320 let cid = SendPortId { sendPortProcessId = processId proc
321 , sendPortLocalId = lcid
323 let sport = SendPort cid
324 chan <- liftIO newTQueueIO
325 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
326 let rport = ReceivePort $ readTQueue chan
327 let tch = TypedChannel chan'
328 return ( (channelCounter ^: (+ 1))
329 . (typedChannelWithId lcid ^= Just tch)
334 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
335 finalizer st lcid = modifyMVar_ st $
336 return . (typedChannelWithId lcid ^= Nothing)
338-- | Send a message on a typed channel
339sendChan :: Serializable a => SendPort a -> a -> Process ()
340sendChan (SendPort cid) msg = do
342 let node = localNodeId (processNode proc)
343 destNode = processNodeId (sendPortProcessId cid) in do
344 case destNode == node of
345 True -> sendChanLocal cid msg
347 liftIO $ sendBinary (processNode proc)
348 (ProcessIdentifier (processId proc))
349 (SendPortIdentifier cid)
353-- | Send a message on a typed channel. This function makes /no/ attempt to
354-- serialize and (in the case when the @ReceivePort@ resides on the same local
355-- node) therefore ensure that the payload is fully evaluated before it is
357unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
358unsafeSendChan = Unsafe.sendChan
360-- | Wait for a message on a typed channel
361receiveChan :: Serializable a => ReceivePort a -> Process a
362receiveChan = liftIO . atomically . receiveSTM
364-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
365-- non-blocking check for a message.
366receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
367receiveChanTimeout 0 ch = liftIO . atomically $
368 (Just <$> receiveSTM ch) `orElse` return Nothing
369receiveChanTimeout n ch = liftIO . timeout n . atomically $
372-- | Merge a list of typed channels.
374-- The result port is left-biased: if there are messages available on more
375-- than one port, the first available message is returned.
376mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
377mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
379-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
381mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
382mergePortsRR = \ps -> do
383 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
384 return $ ReceivePort (rr psVar)
388 rotate (x:xs) = xs ++ [x]
390 rr :: TVar [STM a] -> STM a
393 a <- foldr1 orElse ps
394 writeTVar psVar (rotate ps)
397--------------------------------------------------------------------------------
398-- Advanced messaging --
399--------------------------------------------------------------------------------
401-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
402newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
405-- | Test the matches in order against each message in the queue
406receiveWait :: [Match b] -> Process b
408 queue <- processQueue <$> ask
409 mProc <- liftIO $ dequeue queue Blocking (map unMatch ms)
412 Nothing -> die $ "System Invariant Violation: CQueue.hs returned `Nothing` "
413 ++ "in the absence of a timeout value."
415-- | Like 'receiveWait' but with a timeout.
417-- If the timeout is zero do a non-blocking check for matching messages. A
418-- non-zero timeout is applied only when waiting for incoming messages (that is,
419-- /after/ we have checked the messages that are already in the mailbox).
420receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
421receiveTimeout t ms = do
422 queue <- processQueue <$> ask
423 let blockSpec = if t == 0 then NonBlocking else Timeout t
424 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
426 Nothing -> return Nothing
427 Just proc -> Just <$> proc
429-- | Match on a typed channel
430matchChan :: ReceivePort a -> (a -> Process b) -> Match b
431matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
433-- | Match on an arbitrary STM action.
435-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
436-- actions with checks against our process' mailbox and/or any typed channel
437-- @ReceivePort@s we may hold.
439-- This allows us to process multiple input streams /along with our mailbox/,
440-- in just the same way that 'matchChan' supports checking both the mailbox
441-- /and/ an arbitrary set of typed channels in one atomic transaction.
443-- Note there are no ordering guarnatees with respect to these disparate input
446matchSTM :: STM a -> (a -> Process b) -> Match b
447matchSTM stm fn = Match $ MatchChan (fmap fn stm)
449-- | Match against any message of the right type
450match :: forall a b. Serializable a => (a -> Process b) -> Match b
451match = matchIf (const True)
453-- | Match against any message of the right type that satisfies a predicate
454matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
455matchIf c p = Match $ MatchMsg $ \msg ->
456 case messageFingerprint msg == fingerprint (undefined :: a) of
459 (UnencodedMessage _ m) ->
460 let m' = unsafeCoerce m :: a in
464 (EncodedMessage _ _) ->
465 if (c decoded) then Just (p decoded) else Nothing
468 -- Make sure the value is fully decoded so that we don't hang to
469 -- bytestrings when the process calling 'matchIf' doesn't process
470 -- the values immediately
471 !decoded = decode (messageEncoding msg)
473-- | Match against any message, regardless of the underlying (contained) type
474matchMessage :: (Message -> Process Message) -> Match Message
475matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
477-- | Match against any message (regardless of underlying type) that satisfies a predicate
478matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
479matchMessageIf c p = Match $ MatchMsg $ \msg ->
484-- | Forward a raw 'Message' to the given 'ProcessId'.
485forward :: Message -> ProcessId -> Process ()
488 let node = processNode proc
490 nid = localNodeId node
491 destNode = (processNodeId them) in do
492 case destNode == nid of
493 True -> sendCtrlMsg Nothing (LocalSend them msg)
494 False -> liftIO $ sendPayload (processNode proc)
495 (ProcessIdentifier (processId proc))
496 (ProcessIdentifier them)
498 (messageToPayload msg)
499 -- We do not fire the trace event until after the sending is complete;
500 -- In the remote case, 'sendMessage' can block in the networking stack.
501 liftIO $ traceEvent (localEventBus node)
504-- | Forward a raw 'Message' to the given 'ProcessId'.
506-- Unlike 'forward', this function is insensitive to 'reconnect'. It will
507-- try to send the message regardless of the history of connection failures
509uforward :: Message -> ProcessId -> Process ()
510uforward msg them = do
512 let node = processNode proc
514 nid = localNodeId node
515 destNode = (processNodeId them) in do
516 case destNode == nid of
517 True -> sendCtrlMsg Nothing (LocalSend them msg)
518 False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)
520 -- We do not fire the trace event until after the sending is complete;
521 -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
522 liftIO $ traceEvent (localEventBus node)
525-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
526-- 'Serializable' - like the datum they contain - but also note, deserialising
527-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
528-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
533-- send self (wrapMessage "blah")
534-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
535-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
536-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
539wrapMessage :: Serializable a => a -> Message
540wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
542-- | This is the /unsafe/ variant of 'wrapMessage'. See
543-- "Control.Distributed.Process.UnsafePrimitives" for details.
544unsafeWrapMessage :: Serializable a => a -> Message
545unsafeWrapMessage = Unsafe.wrapMessage
547-- [note Serializable UnencodedMessage]
548-- if we attempt to serialise an UnencodedMessage, it will be converted to an
549-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
550-- Primitives.hs for the gory details
552-- | Attempt to unwrap a raw 'Message'.
553-- If the type of the decoded message payload matches the expected type, the
554-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
557-- This expression, for example, will evaluate to @Nothing@
558-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
560-- Whereas this expression, will yield @Just "foo"@
561-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
563unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
565 case messageFingerprint msg == fingerprint (undefined :: a) of
566 False -> return Nothing :: m (Maybe a)
568 (UnencodedMessage _ ms) ->
569 let ms' = unsafeCoerce ms :: a
571 (EncodedMessage _ _) ->
572 return (Just (decoded))
574 decoded :: a -- note [decoding]
575 !decoded = decode (messageEncoding msg)
577-- | Attempt to handle a raw 'Message'.
578-- If the type of the message matches the type of the first argument to
579-- the supplied expression, then the message will be decoded and the expression
580-- evaluated against its value. If this runtime type checking fails however,
581-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
582-- evaluation proceeds, the resulting value with be wrapped with @Just@.
584-- Intended for use in `catchesExit` and `matchAny` primitives.
586handleMessage :: forall m a b. (Monad m, Serializable a)
587 => Message -> (a -> m b) -> m (Maybe b)
588handleMessage msg proc = handleMessageIf msg (const True) proc
590-- | Conditionally handle a raw 'Message'.
591-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
592-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
593-- for further information about runtime type checking.
595handleMessageIf :: forall m a b . (Monad m, Serializable a)
600handleMessageIf msg c proc = do
601 case messageFingerprint msg == fingerprint (undefined :: a) of
602 False -> return Nothing :: m (Maybe b)
604 (UnencodedMessage _ ms) ->
605 let ms' = unsafeCoerce ms :: a in
607 True -> do { r <- proc ms'; return (Just r) }
608 False -> return Nothing :: m (Maybe b)
609 (EncodedMessage _ _) ->
611 True -> do { r <- proc (decoded :: a); return (Just r) }
612 False -> return Nothing :: m (Maybe b)
614 decoded :: a -- note [decoding]
615 !decoded = decode (messageEncoding msg)
617-- | As 'handleMessage' but ignores result, which is useful if you don't
618-- care whether or not the handler succeeded.
620handleMessage_ :: forall m a . (Monad m, Serializable a)
621 => Message -> (a -> m ()) -> m ()
622handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
624-- | Conditional version of 'handleMessage_'.
626handleMessageIf_ :: forall m a . (Monad m, Serializable a)
631handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
633-- | Match against an arbitrary message. 'matchAny' removes the first available
634-- message from the process mailbox. To handle arbitrary /raw/ messages once
635-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
637matchAny :: forall b. (Message -> Process b) -> Match b
638matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
640-- | Match against an arbitrary message. Intended for use with 'handleMessage'
641-- and 'unwrapMessage', this function /only/ removes a message from the process
642-- mailbox, /if/ the supplied condition matches. The success (or failure) of
643-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
644-- here, i.e., if the condition evaluates to @True@ then the message will
645-- be removed from the process mailbox and decoded, but that does /not/
646-- guarantee that an expression passed to @handleMessage@ will pass the
647-- runtime type checks and therefore be evaluated.
649matchAnyIf :: forall a b. (Serializable a)
651 -> (Message -> Process b)
653matchAnyIf c p = Match $ MatchMsg $ \msg ->
654 case messageFingerprint msg == fingerprint (undefined :: a) of
655 True | check -> Just (p msg)
660 (EncodedMessage _ _) -> c decoded
661 (UnencodedMessage _ m') -> c (unsafeCoerce m')
663 decoded :: a -- note [decoding]
664 !decoded = decode (messageEncoding msg)
668For an EncodedMessage, we need to ensure the value is fully decoded so that
669we don't hang to bytestrings if the calling process doesn't evaluate
670immediately. For UnencodedMessage we know (because the fingerprint comparison
671succeeds) that unsafeCoerce will not fail.
674-- | Remove any message from the queue
675matchUnknown :: Process b -> Match b
676matchUnknown p = Match $ MatchMsg (const (Just p))
678-- | Receives messages and forwards them to @pid@ if @p msg == True@.
679delegate :: ProcessId -> (Message -> Bool) -> Process ()
682 matchAny (\m -> case (p m) of
683 True -> forward m pid
688-- | A straight relay that forwards all messages to the supplied pid.
689relay :: ProcessId -> Process ()
690relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
692-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
693-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
694-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
695-- fails, no action is taken and the proxy process will continue running.
696proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
700 next <- handleMessage m proc
702 Just True -> forward m pid
703 Just False -> return () -- explicitly ignored
704 Nothing -> return ()) -- un-routable
708--------------------------------------------------------------------------------
709-- Process management --
710--------------------------------------------------------------------------------
712-- | Thrown by 'terminate'
713data ProcessTerminationException = ProcessTerminationException
714 deriving (Show, Typeable)
716instance Exception ProcessTerminationException
718-- | Terminate immediately (throws a ProcessTerminationException)
719terminate :: Process a
720terminate = throwM ProcessTerminationException
723-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
724die :: Serializable a => a -> Process b
726-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
727-- could be decoded by a handler passed to 'catchExit', re-thrown or even
728-- passed to another process via the tracing mechanism.
730 throwM (ProcessExitException pid (createMessage reason))
732-- | Forceful request to kill a process. Where 'exit' provides an exception
733-- that can be caught and handled, 'kill' throws an unexposed exception type
734-- which cannot be handled explicitly (by type).
735kill :: ProcessId -> String -> Process ()
736-- NOTE: We send the message to our local node controller, which will then
737-- forward it to a remote node controller (if applicable). Sending it directly
738-- to a remote node controller means that that the message may overtake a
739-- 'monitor' or 'link' request.
740kill them reason = sendCtrlMsg Nothing (Kill them reason)
742-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
743-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
744-- manner can be handled using the 'catchExit' family of functions.
745exit :: Serializable a => ProcessId -> a -> Process ()
746-- NOTE: We send the message to our local node controller, which will then
747-- forward it to a remote node controller (if applicable). Sending it directly
748-- to a remote node controller means that that the message may overtake a
749-- 'monitor' or 'link' request.
750exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
752-- | Catches 'ProcessExitException'. The handler will not be applied unless its
753-- type matches the encoded data stored in the exception (see the /reason/
754-- argument given to the 'exit' primitive). If the handler cannot be applied,
755-- the exception will be re-thrown.
757-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
758-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
759catchExit :: forall a b . (Show a, Serializable a)
761 -> (ProcessId -> a -> Process b)
763catchExit act exitHandler = Catch.catch act handleExit
765 handleExit ex@(ProcessExitException from msg) =
766 if messageFingerprint msg == fingerprint (undefined :: a)
767 then exitHandler from decoded
771 -- Make sure the value is fully decoded so that we don't hang to
772 -- bytestrings if the caller doesn't use the value immediately
773 !decoded = decode (messageEncoding msg)
775-- | Lift 'Control.Exception.catches' (almost).
777-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
778-- message, a handler must accept inputs of the expected type. In order to
779-- handle a list of potentially different handlers (and therefore input types),
780-- a handler passed to 'catchesExit' must accept 'Message' and return
781-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
783-- See 'maybeHandleMessage' and 'Message' for more details.
784catchesExit :: Process b
785 -> [(ProcessId -> Message -> (Process (Maybe b)))]
787catchesExit act handlers = Catch.catch act ((flip handleExit) handlers)
789 handleExit :: ProcessExitException
790 -> [(ProcessId -> Message -> Process (Maybe b))]
792 handleExit ex [] = throwM ex
793 handleExit ex@(ProcessExitException from msg) (h:hs) = do
796 Nothing -> handleExit ex hs
799-- | Our own process ID
800getSelfPid :: Process ProcessId
801getSelfPid = processId <$> ask
803-- | Get the node ID of our local node
804getSelfNode :: Process NodeId
805getSelfNode = localNodeId . processNode <$> ask
808-- | Get statistics about the specified node
809getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
811 selfNode <- getSelfNode
813 then Right `fmap` getLocalNodeStats -- optimisation
814 else getNodeStatsRemote selfNode
816 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
817 getNodeStatsRemote selfNode = do
818 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
819 bracket (monitorNode nid) unmonitor $ \mRef ->
820 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
821 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
822 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
825-- | Get statistics about our local node
826getLocalNodeStats :: Process NodeStats
827getLocalNodeStats = do
829 sendCtrlMsg Nothing $ GetNodeStats self
830 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
832-- | Get information about the specified process
833getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
835 let them = processNodeId pid in do
837 dest <- mkNode them us
838 sendCtrlMsg dest $ GetInfo pid
840 match (\(p :: ProcessInfo) -> return $ Just p)
841 , match (\(_ :: ProcessInfoNone) -> return Nothing)
843 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
844 mkNode them us = case them == us of
845 True -> return Nothing
846 _ -> return $ Just them
848--------------------------------------------------------------------------------
849-- Monitoring and linking --
850--------------------------------------------------------------------------------
852-- | Link to a remote process (asynchronous)
854-- When process A links to process B (that is, process A calls
855-- @link pidB@) then an asynchronous exception will be thrown to process A
856-- when process B terminates (normally or abnormally), or when process A gets
857-- disconnected from process B. Although it is /technically/ possible to catch
858-- these exceptions, chances are if you find yourself trying to do so you should
859-- probably be using 'monitor' rather than 'link'. In particular, code such as
861-- > link pidB -- Link to process B
862-- > expect -- Wait for a message from process B
863-- > unlink pidB -- Unlink again
865-- doesn't quite do what one might expect: if process B sends a message to
866-- process A, and /subsequently terminates/, then process A might or might not
867-- be terminated too, depending on whether the exception is thrown before or
868-- after the 'unlink' (i.e., this code has a race condition).
870-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
871-- call to 'link' has no effect.
873-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
874-- Linking makes no distinction between normal and abnormal termination of
875-- the remote process.
876link :: ProcessId -> Process ()
877link = sendCtrlMsg Nothing . Link . ProcessIdentifier
879-- | Monitor another process (asynchronous)
881-- When process A monitors process B (that is, process A calls
882-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
883-- when process B terminates (normally or abnormally), or when process A gets
884-- disconnected from process B. You receive this message like any other (using
885-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
886-- 'DiedDisconnect', etc.).
888-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
889-- multiple monitors are set up, multiple notifications will be delivered
890-- and monitors can be disabled individually using 'unmonitor'.
891monitor :: ProcessId -> Process MonitorRef
892monitor = monitor' . ProcessIdentifier
894-- | Establishes temporary monitoring of another process.
896-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
897-- of @code@. Note: although monitoring is no longer active when
898-- @withMonitor@ returns, there might still be unreceived monitor
899-- messages in the queue.
901withMonitor :: ProcessId -> (MonitorRef -> Process a) -> Process a
902withMonitor pid = bracket (monitor pid) unmonitor
904-- | Establishes temporary monitoring of another process.
906-- @withMonitor_ pid code@ sets up monitoring of @pid@ for the duration
907-- of @code@. Note: although monitoring is no longer active when
908-- @withMonitor_@ returns, there might still be unreceived monitor
909-- messages in the queue.
912withMonitor_ :: ProcessId -> Process a -> Process a
913withMonitor_ p = withMonitor p . const
917-- This is synchronous in the sense that once it returns you are guaranteed
918-- that no exception will be raised if the remote process dies. However, it is
919-- asynchronous in the sense that we do not wait for a response from the remote
921unlink :: ProcessId -> Process ()
924 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
928-- | Remove a node link
930-- This has the same synchronous/asynchronous nature as 'unlink'.
931unlinkNode :: NodeId -> Process ()
934 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
938-- | Remove a channel (send port) link
940-- This has the same synchronous/asynchronous nature as 'unlink'.
941unlinkPort :: SendPort a -> Process ()
943 unlinkPortAsync sport
944 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
950-- This has the same synchronous/asynchronous nature as 'unlink'.
952-- ProcessMonitorNotification messages for the given MonitorRef are removed from
954unmonitor :: MonitorRef -> Process ()
957 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
960 -- Discard the notification if any. With the current NC implementation at most
961 -- one notification is in the mailbox for any given ref.
962 void $ receiveTimeout 0
963 [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref' == ref)
967--------------------------------------------------------------------------------
968-- Exception handling --
969--------------------------------------------------------------------------------
971-- | Lift 'Control.Exception.catch'
972catch :: Exception e => Process a -> (e -> Process a) -> Process a
974{-# DEPRECATED catch "Use Control.Monad.Catch.catch instead" #-}
976-- | Lift 'Control.Exception.try'
977try :: Exception e => Process a -> Process (Either e a)
979{-# DEPRECATED try "Use Control.Monad.Catch.try instead" #-}
981-- | Lift 'Control.Exception.mask'
982mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
984{-# DEPRECATED mask "Use Control.Monad.Catch.mask_ instead" #-}
986-- | Lift 'Control.Exception.mask_'
987mask_ :: Process a -> Process a
989{-# DEPRECATED mask_ "Use Control.Monad.Catch.mask_ instead" #-}
991-- | Lift 'Control.Exception.onException'
992onException :: Process a -> Process b -> Process a
993onException = Catch.onException
994{-# DEPRECATED onException "Use Control.Monad.Catch.onException instead" #-}
996-- | Lift 'Control.Exception.bracket'
997bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
998bracket = Catch.bracket
999{-# DEPRECATED bracket "Use Control.Monad.Catch.bracket instead" #-}
1001-- | Lift 'Control.Exception.bracket_'
1002bracket_ :: Process a -> Process b -> Process c -> Process c
1003bracket_ = Catch.bracket_
1004{-# DEPRECATED bracket_ "Use Control.Monad.Catch.bracket_ instead" #-}
1006-- | Lift 'Control.Exception.finally'
1007finally :: Process a -> Process b -> Process a
1008finally = Catch.finally
1009{-# DEPRECATED finally "Use Control.Monad.Catch.finally instead" #-}
1011-- | You need this when using 'catches'
1012data Handler a = forall e . Exception e => Handler (e -> Process a)
1014instance Functor Handler where
1015 fmap f (Handler h) = Handler (fmap f . h)
1017-- | Lift 'Control.Exception.catches'
1018catches :: Process a -> [Handler a] -> Process a
1019catches proc handlers = proc `Catch.catch` catchesHandler handlers
1021catchesHandler :: [Handler a] -> SomeException -> Process a
1022catchesHandler handlers e = foldr tryHandler (throwM e) handlers
1023 where tryHandler (Handler handler) res
1024 = case fromException e of
1025 Just e' -> handler e'
1028--------------------------------------------------------------------------------
1030--------------------------------------------------------------------------------
1032-- | Like 'expect' but with a timeout
1033expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
1034expectTimeout n = receiveTimeout n [match return]
1036-- | Asynchronous version of 'spawn'
1038-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
1039spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
1040spawnAsync nid proc = do
1041 spawnRef <- getSpawnRef
1044 then sendCtrlMsg Nothing $ Spawn proc spawnRef
1045 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
1048-- | Monitor a node (asynchronous)
1049monitorNode :: NodeId -> Process MonitorRef
1051 monitor' . NodeIdentifier
1053-- | Monitor a typed channel (asynchronous)
1054monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1055monitorPort (SendPort cid) =
1056 monitor' (SendPortIdentifier cid)
1058-- | Remove a monitor (asynchronous)
1059unmonitorAsync :: MonitorRef -> Process ()
1061 sendCtrlMsg Nothing . Unmonitor
1063-- | Link to a node (asynchronous)
1064linkNode :: NodeId -> Process ()
1065linkNode = link' . NodeIdentifier
1067-- | Link to a channel (asynchronous)
1068linkPort :: SendPort a -> Process ()
1069linkPort (SendPort cid) =
1070 link' (SendPortIdentifier cid)
1072-- | Remove a process link (asynchronous)
1073unlinkAsync :: ProcessId -> Process ()
1075 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1077-- | Remove a node link (asynchronous)
1078unlinkNodeAsync :: NodeId -> Process ()
1080 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1082-- | Remove a channel (send port) link (asynchronous)
1083unlinkPortAsync :: SendPort a -> Process ()
1084unlinkPortAsync (SendPort cid) =
1085 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1087--------------------------------------------------------------------------------
1089--------------------------------------------------------------------------------
1091data SayMessage = SayMessage { sayTime :: UTCTime
1092 , sayProcess :: ProcessId
1093 , sayMessage :: String }
1096-- There is sadly no Show UTCTime instance
1097instance Show SayMessage where
1100 $ showString "SayMessage "
1101 . showString (formatTime defaultTimeLocale "%c" (sayTime msg))
1103 . showsPrec 11 (sayProcess msg) . showChar ' '
1104 . showsPrec 11 (sayMessage msg) . showChar ' '
1106instance Binary SayMessage where
1108 putUTCTime (sayTime s)
1111 get = SayMessage <$> getUTCTime <*> get <*> get
1113-- Sadly there is no Binary UTCTime instance
1114putUTCTime :: UTCTime -> Put
1115putUTCTime (UTCTime (ModifiedJulianDay day) tod) = do
1117 put (toRational tod)
1119getUTCTime :: Get UTCTime
1123 return $! UTCTime (ModifiedJulianDay day)
1128-- @say message@ sends a message of type 'SayMessage' with the current time and
1129-- 'ProcessId' of the current process to the process registered as @logger@. By
1130-- default, this process simply sends the string to @stderr@. Individual Cloud
1131-- Haskell backends might replace this with a different logger process, however.
1132say :: String -> Process ()
1134 now <- liftIO getCurrentTime
1136 nsend "logger" (SayMessage now us string)
1138--------------------------------------------------------------------------------
1140--------------------------------------------------------------------------------
1142-- | Register a process with the local registry (synchronous). The name must not
1143-- already be registered. The process need not be on this node. A bad
1144-- registration will result in a 'ProcessRegistrationException'
1146-- The process to be registered does not have to be local itself.
1147register :: String -> ProcessId -> Process ()
1148register = registerImpl False
1150-- | Like 'register', but will replace an existing registration.
1151-- The name must already be registered.
1152reregister :: String -> ProcessId -> Process ()
1153reregister = registerImpl True
1155registerImpl :: Bool -> String -> ProcessId -> Process ()
1156registerImpl force label pid = do
1157 mynid <- getSelfNode
1158 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1160 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1161 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1164-- | Register a process with a remote registry (asynchronous).
1166-- The process to be registered does not have to live on the same remote node.
1167-- Reply wil come in the form of a 'RegisterReply' message
1169-- See comments in 'whereisRemoteAsync'
1170registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1171registerRemoteAsync nid label pid = do
1173 sendCtrlMsg (if nid == here then Nothing else Just nid)
1174 (Register label nid (Just pid) False)
1176reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1177reregisterRemoteAsync nid label pid =
1178 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1180-- | Remove a process from the local registry (asynchronous).
1181-- This version will wait until a response is gotten from the
1182-- management process. The name must already be registered.
1183unregister :: String -> Process ()
1184unregister label = do
1185 mynid <- getSelfNode
1186 sendCtrlMsg Nothing (Register label mynid Nothing False)
1188 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1189 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1192-- | Deal with the result from an attempted registration or unregistration
1193-- by throwing an exception if necessary
1194handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()
1195handleRegistrationReply label ok owner =
1197 throwM $ ProcessRegistrationException label owner
1199-- | Remove a process from a remote registry (asynchronous).
1201-- Reply wil come in the form of a 'RegisterReply' message
1203-- See comments in 'whereisRemoteAsync'
1204unregisterRemoteAsync :: NodeId -> String -> Process ()
1205unregisterRemoteAsync nid label =
1206 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1208-- | Query the local process registry
1209whereis :: String -> Process (Maybe ProcessId)
1211 sendCtrlMsg Nothing (WhereIs label)
1212 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1213 (\(WhereIsReply _ mPid) -> return mPid)
1216-- | Query a remote process registry (asynchronous)
1218-- Reply will come in the form of a 'WhereIsReply' message.
1220-- There is currently no synchronous version of 'whereisRemoteAsync': if
1221-- you implement one yourself, be sure to take into account that the remote
1222-- node might die or get disconnect before it can respond (i.e. you should
1223-- use 'monitorNode' and take appropriate action when you receive a
1224-- 'NodeMonitorNotification').
1225whereisRemoteAsync :: NodeId -> String -> Process ()
1226whereisRemoteAsync nid label = do
1228 sendCtrlMsg (if nid == here then Nothing else Just nid) (WhereIs label)
1230-- | Named send to a process in the local registry (asynchronous)
1231nsend :: Serializable a => String -> a -> Process ()
1233 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1235-- | Named send to a process in the local registry (asynchronous).
1236-- This function makes /no/ attempt to serialize and (in the case when the
1237-- destination process resides on the same local node) therefore ensure that
1238-- the payload is fully evaluated before it is delivered.
1239unsafeNSend :: Serializable a => String -> a -> Process ()
1240unsafeNSend = Unsafe.nsend
1242-- | Named send to a process in a remote registry (asynchronous)
1243nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1244nsendRemote nid label msg = do
1246 if here == nid then nsend label msg
1247 else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1249-- | Named send to a process in a remote registry (asynchronous)
1250-- This function makes /no/ attempt to serialize and (in the case when the
1251-- destination process resides on the same local node) therefore ensure that
1252-- the payload is fully evaluated before it is delivered.
1253unsafeNSendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1254unsafeNSendRemote = Unsafe.nsendRemote
1256--------------------------------------------------------------------------------
1258--------------------------------------------------------------------------------
1260-- | Resolve a static value
1261unStatic :: Typeable a => Static a -> Process a
1263 rtable <- remoteTable . processNode <$> ask
1264 case Static.unstatic rtable static of
1265 Left err -> fail $ "Could not resolve static value: " ++ err
1268-- | Resolve a closure
1269unClosure :: Typeable a => Closure a -> Process a
1270unClosure closure = do
1271 rtable <- remoteTable . processNode <$> ask
1272 case Static.unclosure rtable closure of
1273 Left err -> fail $ "Could not resolve closure: " ++ err
1276--------------------------------------------------------------------------------
1278--------------------------------------------------------------------------------
1280-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1281-- message passing. However, when network connections get disrupted this
1282-- illusion cannot always be maintained. Once a network connection breaks (even
1283-- temporarily) no further communication on that connection will be possible.
1284-- For example, if process A sends a message to process B, and A is then
1285-- notified (by monitor notification) that it got disconnected from B, A will
1286-- not be able to send any further messages to B, /unless/ A explicitly
1287-- indicates that it is acceptable to attempt to reconnect to B using the
1288-- Cloud Haskell 'reconnect' primitive.
1290-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1291-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1292-- then receives a monitor notification that its connection to B has been lost,
1293-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1294-- and m3 but not m2.
1296-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1297-- to attempt to reconnect on the next send/. In particular, if no further
1298-- communication attempts are made to B then A can use reconnect to clean up
1299-- its connection to B.
1300reconnect :: ProcessId -> Process ()
1303 node <- processNode <$> ask
1304 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1306-- | Reconnect to a sendport. See 'reconnect' for more information.
1307reconnectPort :: SendPort a -> Process ()
1308reconnectPort them = do
1310 node <- processNode <$> ask
1311 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1313--------------------------------------------------------------------------------
1314-- Auxiliary functions --
1315--------------------------------------------------------------------------------
1317sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1319 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1321sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1322sendChanLocal spId msg =
1323 -- we *must* fully serialize/encode the message here, because
1324 -- attempting to use `unsafeCoerce' in the node controller
1325 -- won't work since we know nothing about the required type
1326 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1328getMonitorRefFor :: Identifier -> Process MonitorRef
1329getMonitorRefFor ident = do
1331 liftIO $ modifyMVar (processState proc) $ \st -> do
1332 let counter = st ^. monitorCounter
1333 return ( monitorCounter ^: (+ 1) $ st
1334 , MonitorRef ident counter
1337getSpawnRef :: Process SpawnRef
1340 liftIO $ modifyMVar (processState proc) $ \st -> do
1341 let counter = st ^. spawnCounter
1342 return ( spawnCounter ^: (+ 1) $ st
1346-- | Monitor a process/node/channel
1347monitor' :: Identifier -> Process MonitorRef
1349 monitorRef <- getMonitorRefFor ident
1350 sendCtrlMsg Nothing $ Monitor monitorRef
1353-- | Link to a process/node/channel
1354link' :: Identifier -> Process ()
1355link' = 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