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
91 , reregisterRemoteAsync
92 , unregisterRemoteAsync
98 -- * Exception handling
122 -- * Internal Exports
126#if ! MIN_VERSION_base(4,6,0)
127import Prelude hiding (catch)
130import Data.Binary (Binary(..), Put, Get, decode)
131import Data.Time.Clock (getCurrentTime, UTCTime(..))
132import Data.Time.Calendar (Day(..))
133import Data.Time.Format (formatTime)
134#if MIN_VERSION_time(1,5,0)
135import Data.Time.Format (defaultTimeLocale)
137import System.Locale (defaultTimeLocale)
139import System.Timeout (timeout)
140import Control.Monad (when, void)
141import Control.Monad.Reader (ask)
142import Control.Monad.IO.Class (liftIO)
143import Control.Monad.Catch
149import qualified Control.Monad.Catch as Catch
150import Control.Applicative
151import Control.Distributed.Process.Internal.StrictMVar
156import Control.Concurrent.STM
165import Control.Distributed.Process.Internal.CQueue
170import Control.Distributed.Process.Serializable (Serializable, fingerprint)
171import Data.Accessor ((^.), (^:), (^=))
172import Control.Distributed.Static
176import Data.Rank1Typeable (Typeable)
177import qualified Control.Distributed.Static as Static (unstatic, unclosure)
178import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
179import Control.Distributed.Process.Internal.Types
189 , NodeMonitorNotification(..)
190 , ProcessMonitorNotification(..)
200 , ProcessExitException(..)
203 , DidUnlinkProcess(..)
208 , ProcessRegistrationException(..)
210 , ProcessInfoNone(..)
214 , createUnencodedMessage
215 , ImplicitReconnect( NoImplicitReconnect)
220import Control.Distributed.Process.Internal.Messaging
227import 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 -- see note [trace MxSent]
265 liftIO $ traceEvent (localEventBus node)
266 MxSent { whichProcess = us
267 , whereTo = ProcId them
268 , message = (createUnencodedMessage msg)
271-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
272-- and (in the case when the destination process resides on the same local
273-- node) therefore ensure that the payload is fully evaluated before it is
275unsafeSend :: Serializable a => ProcessId -> a -> Process ()
276unsafeSend = Unsafe.send
278-- | Send a message unreliably.
280-- Unlike 'send', this function is insensitive to 'reconnect'. It will
281-- try to send the message regardless of the history of connection failures
284-- Message passing with 'usend' is ordered for a given sender and receiver
285-- if the messages arrive at all.
287usend :: Serializable a => ProcessId -> a -> Process ()
290 let us = processId proc
291 node = processNode proc
292 nodeId = localNodeId node
293 let destNodeId = processNodeId them
294 if nodeId == destNodeId
295 then sendLocal them msg
296 else do -- see note [trace MxSent]
297 let evBus = localEventBus node
298 let msg' = createMessage msg
299 sendCtrlMsg (Just destNodeId) $ UnreliableSend (processLocalId them) msg'
300 liftIO $ traceEvent evBus $ MxSent { whichProcess = us
301 , whereTo = ProcId them
306-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize
307-- the message when the destination process resides on the same local
308-- node. Therefore, a local receiver would need to be prepared to cope with any
309-- errors resulting from evaluation of the message.
310unsafeUSend :: Serializable a => ProcessId -> a -> Process ()
311unsafeUSend = Unsafe.usend
313-- | Wait for a message of a specific type
314expect :: forall a. Serializable a => Process a
315expect = receiveWait [match return]
317--------------------------------------------------------------------------------
319--------------------------------------------------------------------------------
321-- | Create a new typed channel
322newChan :: Serializable a => Process (SendPort a, ReceivePort a)
325 liftIO . modifyMVar (processState proc) $ \st -> do
326 let lcid = st ^. channelCounter
327 let cid = SendPortId { sendPortProcessId = processId proc
328 , sendPortLocalId = lcid
330 let sport = SendPort cid
331 chan <- liftIO newTQueueIO
332 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
333 let rport = ReceivePort $ readTQueue chan
334 let tch = TypedChannel chan'
335 return ( (channelCounter ^: (+ 1))
336 . (typedChannelWithId lcid ^= Just tch)
341 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
342 finalizer st lcid = modifyMVar_ st $
343 return . (typedChannelWithId lcid ^= Nothing)
345-- | Send a message on a typed channel
346sendChan :: Serializable a => SendPort a -> a -> Process ()
347sendChan (SendPort cid) msg = do
349 let node = localNodeId (processNode proc)
350 destNode = processNodeId (sendPortProcessId cid) in do
351 case destNode == node of
352 True -> sendChanLocal cid msg
354 liftIO $ sendBinary (processNode proc)
355 (ProcessIdentifier (processId proc))
356 (SendPortIdentifier cid)
360-- | Send a message on a typed channel. This function makes /no/ attempt to
361-- serialize and (in the case when the @ReceivePort@ resides on the same local
362-- node) therefore ensure that the payload is fully evaluated before it is
364unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
365unsafeSendChan = Unsafe.sendChan
367-- | Wait for a message on a typed channel
368receiveChan :: Serializable a => ReceivePort a -> Process a
369receiveChan = liftIO . atomically . receiveSTM
371-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
372-- non-blocking check for a message.
373receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
374receiveChanTimeout 0 ch = liftIO . atomically $
375 (Just <$> receiveSTM ch) `orElse` return Nothing
376receiveChanTimeout n ch = liftIO . timeout n . atomically $
379-- | Merge a list of typed channels.
381-- The result port is left-biased: if there are messages available on more
382-- than one port, the first available message is returned.
383mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
384mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
386-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
388mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
389mergePortsRR = \ps -> do
390 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
391 return $ ReceivePort (rr psVar)
395 rotate (x:xs) = xs ++ [x]
397 rr :: TVar [STM a] -> STM a
400 a <- foldr1 orElse ps
401 writeTVar psVar (rotate ps)
404--------------------------------------------------------------------------------
405-- Advanced messaging --
406--------------------------------------------------------------------------------
408-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
409newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
412-- | Test the matches in order against each message in the queue
413receiveWait :: [Match b] -> Process b
415 queue <- processQueue <$> ask
416 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
419-- | Like 'receiveWait' but with a timeout.
421-- If the timeout is zero do a non-blocking check for matching messages. A
422-- non-zero timeout is applied only when waiting for incoming messages (that is,
423-- /after/ we have checked the messages that are already in the mailbox).
424receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
425receiveTimeout t ms = do
426 queue <- processQueue <$> ask
427 let blockSpec = if t == 0 then NonBlocking else Timeout t
428 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
430 Nothing -> return Nothing
431 Just proc -> Just <$> proc
433-- | Match on a typed channel
434matchChan :: ReceivePort a -> (a -> Process b) -> Match b
435matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
437-- | Match on an arbitrary STM action.
439-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
440-- actions with checks against our process' mailbox and/or any typed channel
441-- @ReceivePort@s we may hold.
443-- This allows us to process multiple input streams /along with our mailbox/,
444-- in just the same way that 'matchChan' supports checking both the mailbox
445-- /and/ an arbitrary set of typed channels in one atomic transaction.
447-- Note there are no ordering guarnatees with respect to these disparate input
450matchSTM :: STM a -> (a -> Process b) -> Match b
451matchSTM stm fn = Match $ MatchChan (fmap fn stm)
453-- | Match against any message of the right type
454match :: forall a b. Serializable a => (a -> Process b) -> Match b
455match = matchIf (const True)
457-- | Match against any message of the right type that satisfies a predicate
458matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
459matchIf c p = Match $ MatchMsg $ \msg ->
460 case messageFingerprint msg == fingerprint (undefined :: a) of
463 (UnencodedMessage _ m) ->
464 let m' = unsafeCoerce m :: a in
468 (EncodedMessage _ _) ->
469 if (c decoded) then Just (p decoded) else Nothing
472 -- Make sure the value is fully decoded so that we don't hang to
473 -- bytestrings when the process calling 'matchIf' doesn't process
474 -- the values immediately
475 !decoded = decode (messageEncoding msg)
477-- | Match against any message, regardless of the underlying (contained) type
478matchMessage :: (Message -> Process Message) -> Match Message
479matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
481-- | Match against any message (regardless of underlying type) that satisfies a predicate
482matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
483matchMessageIf c p = Match $ MatchMsg $ \msg ->
488-- | Forward a raw 'Message' to the given 'ProcessId'.
489forward :: Message -> ProcessId -> Process ()
492 let node = processNode proc
494 nid = localNodeId node
495 destNode = (processNodeId them) in do
496 case destNode == nid of
497 True -> sendCtrlMsg Nothing (LocalSend them msg)
498 False -> liftIO $ sendPayload (processNode proc)
499 (ProcessIdentifier (processId proc))
500 (ProcessIdentifier them)
502 (messageToPayload msg)
504 liftIO $ traceEvent (localEventBus node)
505 MxSent { whichProcess = us
506 , whereTo = ProcId them
510-- | Forward a raw 'Message' to the given 'ProcessId'.
512-- Unlike 'forward', this function is insensitive to 'reconnect'. It will
513-- try to send the message regardless of the history of connection failures
515uforward :: Message -> ProcessId -> Process ()
516uforward msg them = do
518 let node = processNode proc
520 nid = localNodeId node
521 destNode = (processNodeId them) in do
522 case destNode == nid of
523 True -> sendCtrlMsg Nothing (LocalSend them msg)
524 False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)
526 -- We do not fire the trace event until after the sending is complete;
527 -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
528 liftIO $ traceEvent (localEventBus node)
529 MxSent { whichProcess = us
530 , whereTo = ProcId them
534-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
535-- 'Serializable' - like the datum they contain - but also note, deserialising
536-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
537-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
542-- send self (wrapMessage "blah")
543-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
544-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
545-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
548wrapMessage :: Serializable a => a -> Message
549wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
551-- | This is the /unsafe/ variant of 'wrapMessage'. See
552-- "Control.Distributed.Process.UnsafePrimitives" for details.
553unsafeWrapMessage :: Serializable a => a -> Message
554unsafeWrapMessage = Unsafe.wrapMessage
556-- [note Serializable UnencodedMessage]
557-- if we attempt to serialise an UnencodedMessage, it will be converted to an
558-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
559-- Primitives.hs for the gory details
561-- | Attempt to unwrap a raw 'Message'.
562-- If the type of the decoded message payload matches the expected type, the
563-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
566-- This expression, for example, will evaluate to @Nothing@
567-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
569-- Whereas this expression, will yield @Just "foo"@
570-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
572unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
574 case messageFingerprint msg == fingerprint (undefined :: a) of
575 False -> return Nothing :: m (Maybe a)
577 (UnencodedMessage _ ms) ->
578 let ms' = unsafeCoerce ms :: a
580 (EncodedMessage _ _) ->
581 return (Just (decoded))
583 decoded :: a -- note [decoding]
584 !decoded = decode (messageEncoding msg)
586-- | Attempt to handle a raw 'Message'.
587-- If the type of the message matches the type of the first argument to
588-- the supplied expression, then the message will be decoded and the expression
589-- evaluated against its value. If this runtime type checking fails however,
590-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
591-- evaluation proceeds, the resulting value with be wrapped with @Just@.
593-- Intended for use in `catchesExit` and `matchAny` primitives.
595handleMessage :: forall m a b. (Monad m, Serializable a)
596 => Message -> (a -> m b) -> m (Maybe b)
597handleMessage msg proc = handleMessageIf msg (const True) proc
599-- | Conditionally handle a raw 'Message'.
600-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
601-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
602-- for further information about runtime type checking.
604handleMessageIf :: forall m a b . (Monad m, Serializable a)
609handleMessageIf msg c proc = do
610 case messageFingerprint msg == fingerprint (undefined :: a) of
611 False -> return Nothing :: m (Maybe b)
613 (UnencodedMessage _ ms) ->
614 let ms' = unsafeCoerce ms :: a in
616 True -> do { r <- proc ms'; return (Just r) }
617 False -> return Nothing :: m (Maybe b)
618 (EncodedMessage _ _) ->
620 True -> do { r <- proc (decoded :: a); return (Just r) }
621 False -> return Nothing :: m (Maybe b)
623 decoded :: a -- note [decoding]
624 !decoded = decode (messageEncoding msg)
626-- | As 'handleMessage' but ignores result, which is useful if you don't
627-- care whether or not the handler succeeded.
629handleMessage_ :: forall m a . (Monad m, Serializable a)
630 => Message -> (a -> m ()) -> m ()
631handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
633-- | Conditional version of 'handleMessage_'.
635handleMessageIf_ :: forall m a . (Monad m, Serializable a)
640handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
642-- | Match against an arbitrary message. 'matchAny' removes the first available
643-- message from the process mailbox. To handle arbitrary /raw/ messages once
644-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
646matchAny :: forall b. (Message -> Process b) -> Match b
647matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
649-- | Match against an arbitrary message. Intended for use with 'handleMessage'
650-- and 'unwrapMessage', this function /only/ removes a message from the process
651-- mailbox, /if/ the supplied condition matches. The success (or failure) of
652-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
653-- here, i.e., if the condition evaluates to @True@ then the message will
654-- be removed from the process mailbox and decoded, but that does /not/
655-- guarantee that an expression passed to @handleMessage@ will pass the
656-- runtime type checks and therefore be evaluated.
658matchAnyIf :: forall a b. (Serializable a)
660 -> (Message -> Process b)
662matchAnyIf c p = Match $ MatchMsg $ \msg ->
663 case messageFingerprint msg == fingerprint (undefined :: a) of
664 True | check -> Just (p msg)
669 (EncodedMessage _ _) -> c decoded
670 (UnencodedMessage _ m') -> c (unsafeCoerce m')
672 decoded :: a -- note [decoding]
673 !decoded = decode (messageEncoding msg)
677For an EncodedMessage, we need to ensure the value is fully decoded so that
678we don't hang to bytestrings if the calling process doesn't evaluate
679immediately. For UnencodedMessage we know (because the fingerprint comparison
680succeeds) that unsafeCoerce will not fail.
683-- | Remove any message from the queue
684matchUnknown :: Process b -> Match b
685matchUnknown p = Match $ MatchMsg (const (Just p))
687-- | Receives messages and forwards them to @pid@ if @p msg == True@.
688delegate :: ProcessId -> (Message -> Bool) -> Process ()
691 matchAny (\m -> case (p m) of
692 True -> forward m pid
697-- | A straight relay that forwards all messages to the supplied pid.
698relay :: ProcessId -> Process ()
699relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
701-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
702-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
703-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
704-- fails, no action is taken and the proxy process will continue running.
705proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
709 next <- handleMessage m proc
711 Just True -> forward m pid
712 Just False -> return () -- explicitly ignored
713 Nothing -> return ()) -- un-routable
717--------------------------------------------------------------------------------
718-- Process management --
719--------------------------------------------------------------------------------
721-- | Thrown by 'terminate'
722data ProcessTerminationException = ProcessTerminationException
723 deriving (Show, Typeable)
725instance Exception ProcessTerminationException
727-- | Terminate immediately (throws a ProcessTerminationException)
728terminate :: Process a
729terminate = throwM ProcessTerminationException
732-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
733die :: Serializable a => a -> Process b
735-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
736-- could be decoded by a handler passed to 'catchExit', re-thrown or even
737-- passed to another process via the tracing mechanism.
739 throwM (ProcessExitException pid (createMessage reason))
741-- | Forceful request to kill a process. Where 'exit' provides an exception
742-- that can be caught and handled, 'kill' throws an unexposed exception type
743-- which cannot be handled explicitly (by type).
744kill :: ProcessId -> String -> Process ()
745-- NOTE: We send the message to our local node controller, which will then
746-- forward it to a remote node controller (if applicable). Sending it directly
747-- to a remote node controller means that that the message may overtake a
748-- 'monitor' or 'link' request.
749kill them reason = sendCtrlMsg Nothing (Kill them reason)
751-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
752-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
753-- manner can be handled using the 'catchExit' family of functions.
754exit :: Serializable a => ProcessId -> a -> Process ()
755-- NOTE: We send the message to our local node controller, which will then
756-- forward it to a remote node controller (if applicable). Sending it directly
757-- to a remote node controller means that that the message may overtake a
758-- 'monitor' or 'link' request.
759exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
761-- | Catches 'ProcessExitException'. The handler will not be applied unless its
762-- type matches the encoded data stored in the exception (see the /reason/
763-- argument given to the 'exit' primitive). If the handler cannot be applied,
764-- the exception will be re-thrown.
766-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
767-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
768catchExit :: forall a b . (Show a, Serializable a)
770 -> (ProcessId -> a -> Process b)
772catchExit act exitHandler = Catch.catch act handleExit
774 handleExit ex@(ProcessExitException from msg) =
775 if messageFingerprint msg == fingerprint (undefined :: a)
776 then exitHandler from decoded
780 -- Make sure the value is fully decoded so that we don't hang to
781 -- bytestrings if the caller doesn't use the value immediately
782 !decoded = decode (messageEncoding msg)
784-- | Lift 'Control.Exception.catches' (almost).
786-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
787-- message, a handler must accept inputs of the expected type. In order to
788-- handle a list of potentially different handlers (and therefore input types),
789-- a handler passed to 'catchesExit' must accept 'Message' and return
790-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
792-- See 'maybeHandleMessage' and 'Message' for more details.
793catchesExit :: Process b
794 -> [(ProcessId -> Message -> (Process (Maybe b)))]
796catchesExit act handlers = Catch.catch act ((flip handleExit) handlers)
798 handleExit :: ProcessExitException
799 -> [(ProcessId -> Message -> Process (Maybe b))]
801 handleExit ex [] = throwM ex
802 handleExit ex@(ProcessExitException from msg) (h:hs) = do
805 Nothing -> handleExit ex hs
808-- | Our own process ID
809getSelfPid :: Process ProcessId
810getSelfPid = processId <$> ask
812-- | Get the node ID of our local node
813getSelfNode :: Process NodeId
814getSelfNode = localNodeId . processNode <$> ask
817-- | Get statistics about the specified node
818getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
820 selfNode <- getSelfNode
822 then Right `fmap` getLocalNodeStats -- optimisation
823 else getNodeStatsRemote selfNode
825 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
826 getNodeStatsRemote selfNode = do
827 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
828 bracket (monitorNode nid) unmonitor $ \mRef ->
829 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
830 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
831 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
834-- | Get statistics about our local node
835getLocalNodeStats :: Process NodeStats
836getLocalNodeStats = do
838 sendCtrlMsg Nothing $ GetNodeStats self
839 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
841-- | Get information about the specified process
842getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
844 let them = processNodeId pid in do
846 dest <- mkNode them us
847 sendCtrlMsg dest $ GetInfo pid
849 match (\(p :: ProcessInfo) -> return $ Just p)
850 , match (\(_ :: ProcessInfoNone) -> return Nothing)
852 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
853 mkNode them us = case them == us of
854 True -> return Nothing
855 _ -> return $ Just them
857--------------------------------------------------------------------------------
858-- Monitoring and linking --
859--------------------------------------------------------------------------------
861-- | Link to a remote process (asynchronous)
863-- When process A links to process B (that is, process A calls
864-- @link pidB@) then an asynchronous exception will be thrown to process A
865-- when process B terminates (normally or abnormally), or when process A gets
866-- disconnected from process B. Although it is /technically/ possible to catch
867-- these exceptions, chances are if you find yourself trying to do so you should
868-- probably be using 'monitor' rather than 'link'. In particular, code such as
870-- > link pidB -- Link to process B
871-- > expect -- Wait for a message from process B
872-- > unlink pidB -- Unlink again
874-- doesn't quite do what one might expect: if process B sends a message to
875-- process A, and /subsequently terminates/, then process A might or might not
876-- be terminated too, depending on whether the exception is thrown before or
877-- after the 'unlink' (i.e., this code has a race condition).
879-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
880-- call to 'link' has no effect.
882-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
883-- Linking makes no distinction between normal and abnormal termination of
884-- the remote process.
885link :: ProcessId -> Process ()
886link = sendCtrlMsg Nothing . Link . ProcessIdentifier
888-- | Monitor another process (asynchronous)
890-- When process A monitors process B (that is, process A calls
891-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
892-- when process B terminates (normally or abnormally), or when process A gets
893-- disconnected from process B. You receive this message like any other (using
894-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
895-- 'DiedDisconnect', etc.).
897-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
898-- multiple monitors are set up, multiple notifications will be delivered
899-- and monitors can be disabled individually using 'unmonitor'.
900monitor :: ProcessId -> Process MonitorRef
901monitor = monitor' . ProcessIdentifier
903-- | Establishes temporary monitoring of another process.
905-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
906-- of @code@. Note: although monitoring is no longer active when
907-- @withMonitor@ returns, there might still be unreceived monitor
908-- messages in the queue.
910withMonitor :: ProcessId -> (MonitorRef -> Process a) -> Process a
911withMonitor pid = bracket (monitor pid) unmonitor
913-- | Establishes temporary monitoring of another process.
915-- @withMonitor_ pid code@ sets up monitoring of @pid@ for the duration
916-- of @code@. Note: although monitoring is no longer active when
917-- @withMonitor_@ returns, there might still be unreceived monitor
918-- messages in the queue.
921withMonitor_ :: ProcessId -> Process a -> Process a
922withMonitor_ p = withMonitor p . const
926-- This is synchronous in the sense that once it returns you are guaranteed
927-- that no exception will be raised if the remote process dies. However, it is
928-- asynchronous in the sense that we do not wait for a response from the remote
930unlink :: ProcessId -> Process ()
933 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
937-- | Remove a node link
939-- This has the same synchronous/asynchronous nature as 'unlink'.
940unlinkNode :: NodeId -> Process ()
943 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
947-- | Remove a channel (send port) link
949-- This has the same synchronous/asynchronous nature as 'unlink'.
950unlinkPort :: SendPort a -> Process ()
952 unlinkPortAsync sport
953 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
959-- This has the same synchronous/asynchronous nature as 'unlink'.
961-- ProcessMonitorNotification messages for the given MonitorRef are removed from
963unmonitor :: MonitorRef -> Process ()
966 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
969 -- Discard the notification if any. With the current NC implementation at most
970 -- one notification is in the mailbox for any given ref.
971 void $ receiveTimeout 0
972 [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref' == ref)
976--------------------------------------------------------------------------------
977-- Exception handling --
978--------------------------------------------------------------------------------
980-- | Lift 'Control.Exception.catch'
981catch :: Exception e => Process a -> (e -> Process a) -> Process a
983{-# DEPRECATED catch "Use Control.Monad.Catch.catch instead" #-}
985-- | Lift 'Control.Exception.try'
986try :: Exception e => Process a -> Process (Either e a)
988{-# DEPRECATED try "Use Control.Monad.Catch.mask_ instead" #-}
990-- | Lift 'Control.Exception.mask'
991mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
993{-# DEPRECATED mask "Use Control.Monad.Catch.mask_ instead" #-}
995-- | Lift 'Control.Exception.mask_'
996mask_ :: Process a -> Process a
998{-# DEPRECATED mask_ "Use Control.Monad.Catch.mask_ instead" #-}
1000-- | Lift 'Control.Exception.onException'
1001onException :: Process a -> Process b -> Process a
1002onException = Catch.onException
1003{-# DEPRECATED onException "Use Control.Monad.Catch.onException instead" #-}
1005-- | Lift 'Control.Exception.bracket'
1006bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
1007bracket = Catch.bracket
1008{-# DEPRECATED bracket "Use Control.Monad.Catch.bracket instead" #-}
1010-- | Lift 'Control.Exception.bracket_'
1011bracket_ :: Process a -> Process b -> Process c -> Process c
1012bracket_ = Catch.bracket_
1013{-# DEPRECATED bracket_ "Use Control.Monad.Catch.bracket_ instead" #-}
1015-- | Lift 'Control.Exception.finally'
1016finally :: Process a -> Process b -> Process a
1017finally = Catch.finally
1018{-# DEPRECATED finally "Use Control.Monad.Catch.finally instead" #-}
1020-- | You need this when using 'catches'
1021data Handler a = forall e . Exception e => Handler (e -> Process a)
1023instance Functor Handler where
1024 fmap f (Handler h) = Handler (fmap f . h)
1026-- | Lift 'Control.Exception.catches'
1027catches :: Process a -> [Handler a] -> Process a
1028catches proc handlers = proc `Catch.catch` catchesHandler handlers
1030catchesHandler :: [Handler a] -> SomeException -> Process a
1031catchesHandler handlers e = foldr tryHandler (throwM e) handlers
1032 where tryHandler (Handler handler) res
1033 = case fromException e of
1034 Just e' -> handler e'
1037--------------------------------------------------------------------------------
1039--------------------------------------------------------------------------------
1041-- | Like 'expect' but with a timeout
1042expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
1043expectTimeout n = receiveTimeout n [match return]
1045-- | Asynchronous version of 'spawn'
1047-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
1048spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
1049spawnAsync nid proc = do
1050 spawnRef <- getSpawnRef
1053 then sendCtrlMsg Nothing $ Spawn proc spawnRef
1054 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
1057-- | Monitor a node (asynchronous)
1058monitorNode :: NodeId -> Process MonitorRef
1060 monitor' . NodeIdentifier
1062-- | Monitor a typed channel (asynchronous)
1063monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1064monitorPort (SendPort cid) =
1065 monitor' (SendPortIdentifier cid)
1067-- | Remove a monitor (asynchronous)
1068unmonitorAsync :: MonitorRef -> Process ()
1070 sendCtrlMsg Nothing . Unmonitor
1072-- | Link to a node (asynchronous)
1073linkNode :: NodeId -> Process ()
1074linkNode = link' . NodeIdentifier
1076-- | Link to a channel (asynchronous)
1077linkPort :: SendPort a -> Process ()
1078linkPort (SendPort cid) =
1079 link' (SendPortIdentifier cid)
1081-- | Remove a process link (asynchronous)
1082unlinkAsync :: ProcessId -> Process ()
1084 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1086-- | Remove a node link (asynchronous)
1087unlinkNodeAsync :: NodeId -> Process ()
1089 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1091-- | Remove a channel (send port) link (asynchronous)
1092unlinkPortAsync :: SendPort a -> Process ()
1093unlinkPortAsync (SendPort cid) =
1094 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1096--------------------------------------------------------------------------------
1098--------------------------------------------------------------------------------
1100data SayMessage = SayMessage { sayTime :: UTCTime
1101 , sayProcess :: ProcessId
1102 , sayMessage :: String }
1105-- There is sadly no Show UTCTime instance
1106instance Show SayMessage where
1109 $ showString "SayMessage "
1110 . showString (formatTime defaultTimeLocale "%c" (sayTime msg))
1112 . showsPrec 11 (sayProcess msg) . showChar ' '
1113 . showsPrec 11 (sayMessage msg) . showChar ' '
1115instance Binary SayMessage where
1117 putUTCTime (sayTime s)
1120 get = SayMessage <$> getUTCTime <*> get <*> get
1122-- Sadly there is no Binary UTCTime instance
1123putUTCTime :: UTCTime -> Put
1124putUTCTime (UTCTime (ModifiedJulianDay day) tod) = do
1126 put (toRational tod)
1128getUTCTime :: Get UTCTime
1132 return $! UTCTime (ModifiedJulianDay day)
1137-- @say message@ sends a message of type 'SayMessage' with the current time and
1138-- 'ProcessId' of the current process to the process registered as @logger@. By
1139-- default, this process simply sends the string to @stderr@. Individual Cloud
1140-- Haskell backends might replace this with a different logger process, however.
1141say :: String -> Process ()
1143 now <- liftIO getCurrentTime
1145 nsend "logger" (SayMessage now us string)
1147--------------------------------------------------------------------------------
1149--------------------------------------------------------------------------------
1151-- | Register a process with the local registry (synchronous). The name must not
1152-- already be registered. The process need not be on this node. A bad
1153-- registration will result in a 'ProcessRegistrationException'
1155-- The process to be registered does not have to be local itself.
1156register :: String -> ProcessId -> Process ()
1157register = registerImpl False
1159-- | Like 'register', but will replace an existing registration.
1160-- The name must already be registered.
1161reregister :: String -> ProcessId -> Process ()
1162reregister = registerImpl True
1164registerImpl :: Bool -> String -> ProcessId -> Process ()
1165registerImpl force label pid = do
1166 mynid <- getSelfNode
1167 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1169 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1170 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1173-- | Register a process with a remote registry (asynchronous).
1175-- The process to be registered does not have to live on the same remote node.
1176-- Reply wil come in the form of a 'RegisterReply' message
1178-- See comments in 'whereisRemoteAsync'
1179registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1180registerRemoteAsync nid label pid = do
1182 sendCtrlMsg (if nid == here then Nothing else Just nid)
1183 (Register label nid (Just pid) False)
1185reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1186reregisterRemoteAsync nid label pid =
1187 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1189-- | Remove a process from the local registry (asynchronous).
1190-- This version will wait until a response is gotten from the
1191-- management process. The name must already be registered.
1192unregister :: String -> Process ()
1193unregister label = do
1194 mynid <- getSelfNode
1195 sendCtrlMsg Nothing (Register label mynid Nothing False)
1197 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1198 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1201-- | Deal with the result from an attempted registration or unregistration
1202-- by throwing an exception if necessary
1203handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()
1204handleRegistrationReply label ok owner =
1206 throwM $ ProcessRegistrationException label owner
1208-- | Remove a process from a remote registry (asynchronous).
1210-- Reply wil come in the form of a 'RegisterReply' message
1212-- See comments in 'whereisRemoteAsync'
1213unregisterRemoteAsync :: NodeId -> String -> Process ()
1214unregisterRemoteAsync nid label =
1215 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1217-- | Query the local process registry
1218whereis :: String -> Process (Maybe ProcessId)
1220 sendCtrlMsg Nothing (WhereIs label)
1221 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1222 (\(WhereIsReply _ mPid) -> return mPid)
1225-- | Query a remote process registry (asynchronous)
1227-- Reply will come in the form of a 'WhereIsReply' message.
1229-- There is currently no synchronous version of 'whereisRemoteAsync': if
1230-- you implement one yourself, be sure to take into account that the remote
1231-- node might die or get disconnect before it can respond (i.e. you should
1232-- use 'monitorNode' and take appropriate action when you receive a
1233-- 'NodeMonitorNotification').
1234whereisRemoteAsync :: NodeId -> String -> Process ()
1235whereisRemoteAsync nid label = do
1237 sendCtrlMsg (if nid == here then Nothing else Just nid) (WhereIs label)
1239-- | Named send to a process in the local registry (asynchronous)
1240nsend :: Serializable a => String -> a -> Process ()
1243 let us = processId proc
1244 node = processNode proc
1245 msg' = createMessage msg -- NB: using UnencodedMessage here causes chaos... :/
1247 sendCtrlMsg Nothing (NamedSend label msg')
1249 -- see note [trace MxSent]
1250 liftIO $ traceEvent (localEventBus node)
1251 MxSent { whichProcess = us
1252 , whereTo = ProcName label
1256-- | Named send to a process in the local registry (asynchronous).
1257-- This function makes /no/ attempt to serialize and (in the case when the
1258-- destination process resides on the same local node) therefore ensure that
1259-- the payload is fully evaluated before it is delivered.
1260unsafeNSend :: Serializable a => String -> a -> Process ()
1261unsafeNSend = Unsafe.nsend
1263-- | Named send to a process in a remote registry (asynchronous)
1264nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1265nsendRemote nid label msg = do
1267 let us = processId proc
1268 node = processNode proc
1270 if here == nid then nsend label msg
1271 else do let msg' = createMessage msg
1272 sendCtrlMsg (Just nid) (NamedSend label msg')
1273 -- see note [trace MxSent]
1274 liftIO $ traceEvent (localEventBus node)
1275 MxSent { whichProcess = us
1276 , whereTo = ProcName label
1280-- | Named send to a process in a remote registry (asynchronous)
1281-- This function makes /no/ attempt to serialize and (in the case when the
1282-- destination process resides on the same local node) therefore ensure that
1283-- the payload is fully evaluated before it is delivered.
1284unsafeNSendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1285unsafeNSendRemote = Unsafe.nsendRemote
1287--------------------------------------------------------------------------------
1289--------------------------------------------------------------------------------
1291-- | Resolve a static value
1292unStatic :: Typeable a => Static a -> Process a
1294 rtable <- remoteTable . processNode <$> ask
1295 case Static.unstatic rtable static of
1296 Left err -> fail $ "Could not resolve static value: " ++ err
1299-- | Resolve a closure
1300unClosure :: Typeable a => Closure a -> Process a
1301unClosure closure = do
1302 rtable <- remoteTable . processNode <$> ask
1303 case Static.unclosure rtable closure of
1304 Left err -> fail $ "Could not resolve closure: " ++ err
1307--------------------------------------------------------------------------------
1309--------------------------------------------------------------------------------
1311-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1312-- message passing. However, when network connections get disrupted this
1313-- illusion cannot always be maintained. Once a network connection breaks (even
1314-- temporarily) no further communication on that connection will be possible.
1315-- For example, if process A sends a message to process B, and A is then
1316-- notified (by monitor notification) that it got disconnected from B, A will
1317-- not be able to send any further messages to B, /unless/ A explicitly
1318-- indicates that it is acceptable to attempt to reconnect to B using the
1319-- Cloud Haskell 'reconnect' primitive.
1321-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1322-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1323-- then receives a monitor notification that its connection to B has been lost,
1324-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1325-- and m3 but not m2.
1327-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1328-- to attempt to reconnect on the next send/. In particular, if no further
1329-- communication attempts are made to B then A can use reconnect to clean up
1330-- its connection to B.
1331reconnect :: ProcessId -> Process ()
1334 node <- processNode <$> ask
1335 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1337-- | Reconnect to a sendport. See 'reconnect' for more information.
1338reconnectPort :: SendPort a -> Process ()
1339reconnectPort them = do
1341 node <- processNode <$> ask
1342 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1344--------------------------------------------------------------------------------
1345-- Auxiliary functions --
1346--------------------------------------------------------------------------------
1348sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1349sendLocal to msg = do
1351 let us = processId proc
1352 node = processNode proc
1353 msg' = createUnencodedMessage msg
1355 sendCtrlMsg Nothing $ LocalSend to msg'
1356 liftIO $ traceEvent (localEventBus node)
1357 MxSent { whichProcess = us
1358 , whereTo = ProcId to
1362sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1363sendChanLocal spId msg =
1364 -- we *must* fully serialize/encode the message here, because
1365 -- attempting to use `unsafeCoerce' in the node controller
1366 -- won't work since we know nothing about the required type
1367 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1369getMonitorRefFor :: Identifier -> Process MonitorRef
1370getMonitorRefFor ident = do
1372 liftIO $ modifyMVar (processState proc) $ \st -> do
1373 let counter = st ^. monitorCounter
1374 return ( monitorCounter ^: (+ 1) $ st
1375 , MonitorRef ident counter
1378getSpawnRef :: Process SpawnRef
1381 liftIO $ modifyMVar (processState proc) $ \st -> do
1382 let counter = st ^. spawnCounter
1383 return ( spawnCounter ^: (+ 1) $ st
1387-- | Monitor a process/node/channel
1388monitor' :: Identifier -> Process MonitorRef
1390 monitorRef <- getMonitorRefFor ident
1391 sendCtrlMsg Nothing $ Monitor monitorRef
1394-- | Link to a process/node/channel
1395link' :: Identifier -> Process ()
1396link' = 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