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 liftIO $ traceEvent (localEventBus node)
258 (MxSent them us (createUnencodedMessage msg))
259 if destNode == nodeId
260 then sendLocal them msg
261 else liftIO $ sendMessage (processNode proc)
262 (ProcessIdentifier (processId proc))
263 (ProcessIdentifier them)
267-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
268-- and (in the case when the destination process resides on the same local
269-- node) therefore ensure that the payload is fully evaluated before it is
271unsafeSend :: Serializable a => ProcessId -> a -> Process ()
272unsafeSend = Unsafe.send
274-- | Send a message unreliably.
276-- Unlike 'send', this function is insensitive to 'reconnect'. It will
277-- try to send the message regardless of the history of connection failures
280-- Message passing with 'usend' is ordered for a given sender and receiver
281-- if the messages arrive at all.
283usend :: Serializable a => ProcessId -> a -> Process ()
286 let there = processNodeId them
287 let (us, node) = (processId proc, processNode proc)
288 let msg' = wrapMessage msg
289 liftIO $ traceEvent (localEventBus node) (MxSent them us msg')
290 if localNodeId (processNode proc) == there
291 then sendLocal them msg
292 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
295-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize
296-- the message when the destination process resides on the same local
297-- node. Therefore, a local receiver would need to be prepared to cope with any
298-- errors resulting from evaluation of the message.
299unsafeUSend :: Serializable a => ProcessId -> a -> Process ()
300unsafeUSend = Unsafe.usend
302-- | Wait for a message of a specific type
303expect :: forall a. Serializable a => Process a
304expect = receiveWait [match return]
306--------------------------------------------------------------------------------
308--------------------------------------------------------------------------------
310-- | Create a new typed channel, bound to the calling Process
312-- Note that the channel is bound to the lifecycle of the process that evaluates
313-- this function, such that when it dies/exits, the channel will no longer
314-- function, but will remain accessible. Thus reading from the ReceivePort will
315-- fail silently thereafter, blocking indefinitely (unless a timeout is used).
317newChan :: Serializable a => Process (SendPort a, ReceivePort a)
320 liftIO . modifyMVar (processState proc) $ \st -> do
321 let lcid = st ^. channelCounter
322 let cid = SendPortId { sendPortProcessId = processId proc
323 , sendPortLocalId = lcid
325 let sport = SendPort cid
326 chan <- liftIO newTQueueIO
327 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
328 let rport = ReceivePort $ readTQueue chan
329 let tch = TypedChannel chan'
330 return ( (channelCounter ^: (+ 1))
331 . (typedChannelWithId lcid ^= Just tch)
336 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
337 finalizer st lcid = modifyMVar_ st $
338 return . (typedChannelWithId lcid ^= Nothing)
340-- | Send a message on a typed channel
341sendChan :: Serializable a => SendPort a -> a -> Process ()
342sendChan (SendPort cid) msg = do
344 let node = processNode proc
346 us = localNodeId node
347 them = processNodeId (sendPortProcessId cid) in do
348 liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid $ wrapMessage msg)
350 True -> sendChanLocal cid msg
352 liftIO $ sendBinary node
353 (ProcessIdentifier pid)
354 (SendPortIdentifier cid)
358-- | Send a message on a typed channel. This function makes /no/ attempt to
359-- serialize and (in the case when the @ReceivePort@ resides on the same local
360-- node) therefore ensure that the payload is fully evaluated before it is
362unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
363unsafeSendChan = Unsafe.sendChan
365-- | Wait for a message on a typed channel
366receiveChan :: Serializable a => ReceivePort a -> Process a
367receiveChan = liftIO . atomically . receiveSTM
369-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
370-- non-blocking check for a message.
371receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
372receiveChanTimeout 0 ch = liftIO . atomically $
373 (Just <$> receiveSTM ch) `orElse` return Nothing
374receiveChanTimeout n ch = liftIO . timeout n . atomically $
377-- | Merge a list of typed channels.
379-- The result port is left-biased: if there are messages available on more
380-- than one port, the first available message is returned.
381mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
382mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
384-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
386mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
387mergePortsRR = \ps -> do
388 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
389 return $ ReceivePort (rr psVar)
393 rotate (x:xs) = xs ++ [x]
395 rr :: TVar [STM a] -> STM a
398 a <- foldr1 orElse ps
399 writeTVar psVar (rotate ps)
402--------------------------------------------------------------------------------
403-- Advanced messaging --
404--------------------------------------------------------------------------------
406-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
407newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
410-- | Test the matches in order against each message in the queue
411receiveWait :: [Match b] -> Process b
413 queue <- processQueue <$> ask
414 mProc <- liftIO $ dequeue queue Blocking (map unMatch ms)
417 Nothing -> die $ "System Invariant Violation: CQueue.hs returned `Nothing` "
418 ++ "in the absence of a timeout value."
420-- | Like 'receiveWait' but with a timeout.
422-- If the timeout is zero do a non-blocking check for matching messages. A
423-- non-zero timeout is applied only when waiting for incoming messages (that is,
424-- /after/ we have checked the messages that are already in the mailbox).
425receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
426receiveTimeout t ms = do
427 queue <- processQueue <$> ask
428 let blockSpec = if t == 0 then NonBlocking else Timeout t
429 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
431 Nothing -> return Nothing
432 Just proc -> Just <$> proc
434-- | Match on a typed channel
435matchChan :: ReceivePort a -> (a -> Process b) -> Match b
436matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
438-- | Match on an arbitrary STM action.
440-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
441-- actions with checks against our process' mailbox and/or any typed channel
442-- @ReceivePort@s we may hold.
444-- This allows us to process multiple input streams /along with our mailbox/,
445-- in just the same way that 'matchChan' supports checking both the mailbox
446-- /and/ an arbitrary set of typed channels in one atomic transaction.
448-- Note there are no ordering guarnatees with respect to these disparate input
451matchSTM :: STM a -> (a -> Process b) -> Match b
452matchSTM stm fn = Match $ MatchChan (fmap fn stm)
454-- | Match against any message of the right type
455match :: forall a b. Serializable a => (a -> Process b) -> Match b
456match = matchIf (const True)
458-- | Match against any message of the right type that satisfies a predicate
459matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
460matchIf c p = Match $ MatchMsg $ \msg ->
461 case messageFingerprint msg == fingerprint (undefined :: a) of
464 (UnencodedMessage _ m) ->
465 let m' = unsafeCoerce m :: a in
469 (EncodedMessage _ _) ->
470 if (c decoded) then Just (p decoded) else Nothing
473 -- Make sure the value is fully decoded so that we don't hang to
474 -- bytestrings when the process calling 'matchIf' doesn't process
475 -- the values immediately
476 !decoded = decode (messageEncoding msg)
478-- | Match against any message, regardless of the underlying (contained) type
479matchMessage :: (Message -> Process Message) -> Match Message
480matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
482-- | Match against any message (regardless of underlying type) that satisfies a predicate
483matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
484matchMessageIf c p = Match $ MatchMsg $ \msg ->
489-- | Forward a raw 'Message' to the given 'ProcessId'.
490forward :: Message -> ProcessId -> Process ()
493 let node = processNode proc
495 nid = localNodeId node
496 destNode = (processNodeId them) in do
497 liftIO $ traceEvent (localEventBus node) (MxSent them us msg)
499 then sendCtrlMsg Nothing (LocalSend them msg)
500 else liftIO $ sendPayload (processNode proc)
501 (ProcessIdentifier (processId proc))
502 (ProcessIdentifier them)
504 (messageToPayload msg)
506-- | Forward a raw 'Message' to the given 'ProcessId'.
508-- Unlike 'forward', this function is insensitive to 'reconnect'. It will
509-- try to send the message regardless of the history of connection failures
511uforward :: Message -> ProcessId -> Process ()
512uforward msg them = do
514 let node = processNode proc
516 nid = localNodeId node
517 destNode = (processNodeId them) in do
518 liftIO $ traceEvent (localEventBus node) (MxSent them us msg)
520 then sendCtrlMsg Nothing (LocalSend them msg)
521 else sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them) msg
523-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
524-- 'Serializable' - like the datum they contain - but also note, deserialising
525-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
526-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
531-- send self (wrapMessage "blah")
532-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
533-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
534-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
537wrapMessage :: Serializable a => a -> Message
538wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
540-- | This is the /unsafe/ variant of 'wrapMessage'. See
541-- "Control.Distributed.Process.UnsafePrimitives" for details.
542unsafeWrapMessage :: Serializable a => a -> Message
543unsafeWrapMessage = Unsafe.wrapMessage
545-- [note Serializable UnencodedMessage]
546-- if we attempt to serialise an UnencodedMessage, it will be converted to an
547-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
548-- Primitives.hs for the gory details
550-- | Attempt to unwrap a raw 'Message'.
551-- If the type of the decoded message payload matches the expected type, the
552-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
555-- This expression, for example, will evaluate to @Nothing@
556-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
558-- Whereas this expression, will yield @Just "foo"@
559-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
561unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
563 case messageFingerprint msg == fingerprint (undefined :: a) of
564 False -> return Nothing :: m (Maybe a)
566 (UnencodedMessage _ ms) ->
567 let ms' = unsafeCoerce ms :: a
569 (EncodedMessage _ _) ->
570 return (Just (decoded))
572 decoded :: a -- note [decoding]
573 !decoded = decode (messageEncoding msg)
575-- | Attempt to handle a raw 'Message'.
576-- If the type of the message matches the type of the first argument to
577-- the supplied expression, then the message will be decoded and the expression
578-- evaluated against its value. If this runtime type checking fails however,
579-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
580-- evaluation proceeds, the resulting value with be wrapped with @Just@.
582-- Intended for use in `catchesExit` and `matchAny` primitives.
584handleMessage :: forall m a b. (Monad m, Serializable a)
585 => Message -> (a -> m b) -> m (Maybe b)
586handleMessage msg proc = handleMessageIf msg (const True) proc
588-- | Conditionally handle a raw 'Message'.
589-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
590-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
591-- for further information about runtime type checking.
593handleMessageIf :: forall m a b . (Monad m, Serializable a)
598handleMessageIf msg c proc = do
599 case messageFingerprint msg == fingerprint (undefined :: a) of
600 False -> return Nothing :: m (Maybe b)
602 (UnencodedMessage _ ms) ->
603 let ms' = unsafeCoerce ms :: a in
605 True -> do { r <- proc ms'; return (Just r) }
606 False -> return Nothing :: m (Maybe b)
607 (EncodedMessage _ _) ->
609 True -> do { r <- proc (decoded :: a); return (Just r) }
610 False -> return Nothing :: m (Maybe b)
612 decoded :: a -- note [decoding]
613 !decoded = decode (messageEncoding msg)
615-- | As 'handleMessage' but ignores result, which is useful if you don't
616-- care whether or not the handler succeeded.
618handleMessage_ :: forall m a . (Monad m, Serializable a)
619 => Message -> (a -> m ()) -> m ()
620handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
622-- | Conditional version of 'handleMessage_'.
624handleMessageIf_ :: forall m a . (Monad m, Serializable a)
629handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
631-- | Match against an arbitrary message. 'matchAny' removes the first available
632-- message from the process mailbox. To handle arbitrary /raw/ messages once
633-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
635matchAny :: forall b. (Message -> Process b) -> Match b
636matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
638-- | Match against an arbitrary message. Intended for use with 'handleMessage'
639-- and 'unwrapMessage', this function /only/ removes a message from the process
640-- mailbox, /if/ the supplied condition matches. The success (or failure) of
641-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
642-- here, i.e., if the condition evaluates to @True@ then the message will
643-- be removed from the process mailbox and decoded, but that does /not/
644-- guarantee that an expression passed to @handleMessage@ will pass the
645-- runtime type checks and therefore be evaluated.
647matchAnyIf :: forall a b. (Serializable a)
649 -> (Message -> Process b)
651matchAnyIf c p = Match $ MatchMsg $ \msg ->
652 case messageFingerprint msg == fingerprint (undefined :: a) of
653 True | check -> Just (p msg)
658 (EncodedMessage _ _) -> c decoded
659 (UnencodedMessage _ m') -> c (unsafeCoerce m')
661 decoded :: a -- note [decoding]
662 !decoded = decode (messageEncoding msg)
666For an EncodedMessage, we need to ensure the value is fully decoded so that
667we don't hang to bytestrings if the calling process doesn't evaluate
668immediately. For UnencodedMessage we know (because the fingerprint comparison
669succeeds) that unsafeCoerce will not fail.
672-- | Remove any message from the queue
673matchUnknown :: Process b -> Match b
674matchUnknown p = Match $ MatchMsg (const (Just p))
676-- | Receives messages and forwards them to @pid@ if @p msg == True@.
677delegate :: ProcessId -> (Message -> Bool) -> Process ()
680 matchAny (\m -> case (p m) of
681 True -> forward m pid
686-- | A straight relay that forwards all messages to the supplied pid.
687relay :: ProcessId -> Process ()
688relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
690-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
691-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
692-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
693-- fails, no action is taken and the proxy process will continue running.
694proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
698 next <- handleMessage m proc
700 Just True -> forward m pid
701 Just False -> return () -- explicitly ignored
702 Nothing -> return ()) -- un-routable
706--------------------------------------------------------------------------------
707-- Process management --
708--------------------------------------------------------------------------------
710-- | Thrown by 'terminate'
711data ProcessTerminationException = ProcessTerminationException
712 deriving (Show, Typeable)
714instance Exception ProcessTerminationException
716-- | Terminate immediately (throws a ProcessTerminationException)
717terminate :: Process a
718terminate = throwM ProcessTerminationException
721-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
722die :: Serializable a => a -> Process b
724-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
725-- could be decoded by a handler passed to 'catchExit', re-thrown or even
726-- passed to another process via the tracing mechanism.
728 throwM (ProcessExitException pid (createMessage reason))
730-- | Forceful request to kill a process. Where 'exit' provides an exception
731-- that can be caught and handled, 'kill' throws an unexposed exception type
732-- which cannot be handled explicitly (by type).
733kill :: ProcessId -> String -> Process ()
734-- NOTE: We send the message to our local node controller, which will then
735-- forward it to a remote node controller (if applicable). Sending it directly
736-- to a remote node controller means that that the message may overtake a
737-- 'monitor' or 'link' request.
738kill them reason = sendCtrlMsg Nothing (Kill them reason)
740-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
741-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
742-- manner can be handled using the 'catchExit' family of functions.
743exit :: Serializable a => ProcessId -> a -> Process ()
744-- NOTE: We send the message to our local node controller, which will then
745-- forward it to a remote node controller (if applicable). Sending it directly
746-- to a remote node controller means that that the message may overtake a
747-- 'monitor' or 'link' request.
748exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
750-- | Catches 'ProcessExitException'. The handler will not be applied unless its
751-- type matches the encoded data stored in the exception (see the /reason/
752-- argument given to the 'exit' primitive). If the handler cannot be applied,
753-- the exception will be re-thrown.
755-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
756-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
757catchExit :: forall a b . (Show a, Serializable a)
759 -> (ProcessId -> a -> Process b)
761catchExit act exitHandler = Catch.catch act handleExit
763 handleExit ex@(ProcessExitException from msg) =
764 if messageFingerprint msg == fingerprint (undefined :: a)
765 then exitHandler from decoded
769 -- Make sure the value is fully decoded so that we don't hang to
770 -- bytestrings if the caller doesn't use the value immediately
771 !decoded = decode (messageEncoding msg)
773-- | Lift 'Control.Exception.catches' (almost).
775-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
776-- message, a handler must accept inputs of the expected type. In order to
777-- handle a list of potentially different handlers (and therefore input types),
778-- a handler passed to 'catchesExit' must accept 'Message' and return
779-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
781-- See 'maybeHandleMessage' and 'Message' for more details.
782catchesExit :: Process b
783 -> [(ProcessId -> Message -> (Process (Maybe b)))]
785catchesExit act handlers = Catch.catch act ((flip handleExit) handlers)
787 handleExit :: ProcessExitException
788 -> [(ProcessId -> Message -> Process (Maybe b))]
790 handleExit ex [] = throwM ex
791 handleExit ex@(ProcessExitException from msg) (h:hs) = do
794 Nothing -> handleExit ex hs
797-- | Our own process ID
798getSelfPid :: Process ProcessId
799getSelfPid = processId <$> ask
801-- | Get the node ID of our local node
802getSelfNode :: Process NodeId
803getSelfNode = localNodeId . processNode <$> ask
806-- | Get statistics about the specified node
807getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
809 selfNode <- getSelfNode
811 then Right `fmap` getLocalNodeStats -- optimisation
812 else getNodeStatsRemote selfNode
814 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
815 getNodeStatsRemote selfNode = do
816 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
817 bracket (monitorNode nid) unmonitor $ \mRef ->
818 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
819 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
820 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
823-- | Get statistics about our local node
824getLocalNodeStats :: Process NodeStats
825getLocalNodeStats = do
827 sendCtrlMsg Nothing $ GetNodeStats self
828 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
830-- | Get information about the specified process
831getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
833 let them = processNodeId pid in do
835 dest <- mkNode them us
836 sendCtrlMsg dest $ GetInfo pid
838 match (\(p :: ProcessInfo) -> return $ Just p)
839 , match (\(_ :: ProcessInfoNone) -> return Nothing)
841 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
842 mkNode them us = case them == us of
843 True -> return Nothing
844 _ -> return $ Just them
846--------------------------------------------------------------------------------
847-- Monitoring and linking --
848--------------------------------------------------------------------------------
850-- | Link to a remote process (asynchronous)
852-- When process A links to process B (that is, process A calls
853-- @link pidB@) then an asynchronous exception will be thrown to process A
854-- when process B terminates (normally or abnormally), or when process A gets
855-- disconnected from process B. Although it is /technically/ possible to catch
856-- these exceptions, chances are if you find yourself trying to do so you should
857-- probably be using 'monitor' rather than 'link'. In particular, code such as
859-- > link pidB -- Link to process B
860-- > expect -- Wait for a message from process B
861-- > unlink pidB -- Unlink again
863-- doesn't quite do what one might expect: if process B sends a message to
864-- process A, and /subsequently terminates/, then process A might or might not
865-- be terminated too, depending on whether the exception is thrown before or
866-- after the 'unlink' (i.e., this code has a race condition).
868-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
869-- call to 'link' has no effect.
871-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
872-- Linking makes no distinction between normal and abnormal termination of
873-- the remote process.
874link :: ProcessId -> Process ()
875link = sendCtrlMsg Nothing . Link . ProcessIdentifier
877-- | Monitor another process (asynchronous)
879-- When process A monitors process B (that is, process A calls
880-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
881-- when process B terminates (normally or abnormally), or when process A gets
882-- disconnected from process B. You receive this message like any other (using
883-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
884-- 'DiedDisconnect', etc.).
886-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
887-- multiple monitors are set up, multiple notifications will be delivered
888-- and monitors can be disabled individually using 'unmonitor'.
889monitor :: ProcessId -> Process MonitorRef
890monitor = monitor' . ProcessIdentifier
892-- | Establishes temporary monitoring of another process.
894-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
895-- of @code@. Note: although monitoring is no longer active when
896-- @withMonitor@ returns, there might still be unreceived monitor
897-- messages in the queue.
899withMonitor :: ProcessId -> (MonitorRef -> Process a) -> Process a
900withMonitor pid = bracket (monitor pid) unmonitor
902-- | Establishes temporary monitoring of another process.
904-- @withMonitor_ pid code@ sets up monitoring of @pid@ for the duration
905-- of @code@. Note: although monitoring is no longer active when
906-- @withMonitor_@ returns, there might still be unreceived monitor
907-- messages in the queue.
910withMonitor_ :: ProcessId -> Process a -> Process a
911withMonitor_ p = withMonitor p . const
915-- This is synchronous in the sense that once it returns you are guaranteed
916-- that no exception will be raised if the remote process dies. However, it is
917-- asynchronous in the sense that we do not wait for a response from the remote
919unlink :: ProcessId -> Process ()
922 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
926-- | Remove a node link
928-- This has the same synchronous/asynchronous nature as 'unlink'.
929unlinkNode :: NodeId -> Process ()
932 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
936-- | Remove a channel (send port) link
938-- This has the same synchronous/asynchronous nature as 'unlink'.
939unlinkPort :: SendPort a -> Process ()
941 unlinkPortAsync sport
942 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
948-- This has the same synchronous/asynchronous nature as 'unlink'.
950-- ProcessMonitorNotification messages for the given MonitorRef are removed from
952unmonitor :: MonitorRef -> Process ()
955 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
958 -- Discard the notification if any. With the current NC implementation at most
959 -- one notification is in the mailbox for any given ref.
960 void $ receiveTimeout 0
961 [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref' == ref)
965--------------------------------------------------------------------------------
966-- Exception handling --
967--------------------------------------------------------------------------------
969-- | Lift 'Control.Exception.catch'
970catch :: Exception e => Process a -> (e -> Process a) -> Process a
972{-# DEPRECATED catch "Use Control.Monad.Catch.catch instead" #-}
974-- | Lift 'Control.Exception.try'
975try :: Exception e => Process a -> Process (Either e a)
977{-# DEPRECATED try "Use Control.Monad.Catch.try instead" #-}
979-- | Lift 'Control.Exception.mask'
980mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
982{-# DEPRECATED mask "Use Control.Monad.Catch.mask_ instead" #-}
984-- | Lift 'Control.Exception.mask_'
985mask_ :: Process a -> Process a
987{-# DEPRECATED mask_ "Use Control.Monad.Catch.mask_ instead" #-}
989-- | Lift 'Control.Exception.onException'
990onException :: Process a -> Process b -> Process a
991onException = Catch.onException
992{-# DEPRECATED onException "Use Control.Monad.Catch.onException instead" #-}
994-- | Lift 'Control.Exception.bracket'
995bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
996bracket = Catch.bracket
997{-# DEPRECATED bracket "Use Control.Monad.Catch.bracket instead" #-}
999-- | Lift 'Control.Exception.bracket_'
1000bracket_ :: Process a -> Process b -> Process c -> Process c
1001bracket_ = Catch.bracket_
1002{-# DEPRECATED bracket_ "Use Control.Monad.Catch.bracket_ instead" #-}
1004-- | Lift 'Control.Exception.finally'
1005finally :: Process a -> Process b -> Process a
1006finally = Catch.finally
1007{-# DEPRECATED finally "Use Control.Monad.Catch.finally instead" #-}
1009-- | You need this when using 'catches'
1010data Handler a = forall e . Exception e => Handler (e -> Process a)
1012instance Functor Handler where
1013 fmap f (Handler h) = Handler (fmap f . h)
1015-- | Lift 'Control.Exception.catches'
1016catches :: Process a -> [Handler a] -> Process a
1017catches proc handlers = proc `Catch.catch` catchesHandler handlers
1019catchesHandler :: [Handler a] -> SomeException -> Process a
1020catchesHandler handlers e = foldr tryHandler (throwM e) handlers
1021 where tryHandler (Handler handler) res
1022 = case fromException e of
1023 Just e' -> handler e'
1026--------------------------------------------------------------------------------
1028--------------------------------------------------------------------------------
1030-- | Like 'expect' but with a timeout
1031expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
1032expectTimeout n = receiveTimeout n [match return]
1034-- | Asynchronous version of 'spawn'
1036-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
1037spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
1038spawnAsync nid proc = do
1039 spawnRef <- getSpawnRef
1042 then sendCtrlMsg Nothing $ Spawn proc spawnRef
1043 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
1046-- | Monitor a node (asynchronous)
1047monitorNode :: NodeId -> Process MonitorRef
1049 monitor' . NodeIdentifier
1051-- | Monitor a typed channel (asynchronous)
1052monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1053monitorPort (SendPort cid) =
1054 monitor' (SendPortIdentifier cid)
1056-- | Remove a monitor (asynchronous)
1057unmonitorAsync :: MonitorRef -> Process ()
1059 sendCtrlMsg Nothing . Unmonitor
1061-- | Link to a node (asynchronous)
1062linkNode :: NodeId -> Process ()
1063linkNode = link' . NodeIdentifier
1065-- | Link to a channel (asynchronous)
1066linkPort :: SendPort a -> Process ()
1067linkPort (SendPort cid) =
1068 link' (SendPortIdentifier cid)
1070-- | Remove a process link (asynchronous)
1071unlinkAsync :: ProcessId -> Process ()
1073 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1075-- | Remove a node link (asynchronous)
1076unlinkNodeAsync :: NodeId -> Process ()
1078 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1080-- | Remove a channel (send port) link (asynchronous)
1081unlinkPortAsync :: SendPort a -> Process ()
1082unlinkPortAsync (SendPort cid) =
1083 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1085--------------------------------------------------------------------------------
1087--------------------------------------------------------------------------------
1089data SayMessage = SayMessage { sayTime :: UTCTime
1090 , sayProcess :: ProcessId
1091 , sayMessage :: String }
1094-- There is sadly no Show UTCTime instance
1095instance Show SayMessage where
1098 $ showString "SayMessage "
1099 . showString (formatTime defaultTimeLocale "%c" (sayTime msg))
1101 . showsPrec 11 (sayProcess msg) . showChar ' '
1102 . showsPrec 11 (sayMessage msg) . showChar ' '
1104instance Binary SayMessage where
1106 putUTCTime (sayTime s)
1109 get = SayMessage <$> getUTCTime <*> get <*> get
1111-- Sadly there is no Binary UTCTime instance
1112putUTCTime :: UTCTime -> Put
1113putUTCTime (UTCTime (ModifiedJulianDay day) tod) = do
1115 put (toRational tod)
1117getUTCTime :: Get UTCTime
1121 return $! UTCTime (ModifiedJulianDay day)
1126-- @say message@ sends a message of type 'SayMessage' with the current time and
1127-- 'ProcessId' of the current process to the process registered as @logger@. By
1128-- default, this process simply sends the string to @stderr@. Individual Cloud
1129-- Haskell backends might replace this with a different logger process, however.
1130say :: String -> Process ()
1132 now <- liftIO getCurrentTime
1134 nsend "logger" (SayMessage now us string)
1136--------------------------------------------------------------------------------
1138--------------------------------------------------------------------------------
1140-- | Register a process with the local registry (synchronous). The name must not
1141-- already be registered. The process need not be on this node. A bad
1142-- registration will result in a 'ProcessRegistrationException'
1144-- The process to be registered does not have to be local itself.
1145register :: String -> ProcessId -> Process ()
1146register = registerImpl False
1148-- | Like 'register', but will replace an existing registration.
1149-- The name must already be registered.
1150reregister :: String -> ProcessId -> Process ()
1151reregister = registerImpl True
1153registerImpl :: Bool -> String -> ProcessId -> Process ()
1154registerImpl force label pid = do
1155 mynid <- getSelfNode
1156 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1158 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1159 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1162-- | Register a process with a remote registry (asynchronous).
1164-- The process to be registered does not have to live on the same remote node.
1165-- Reply wil come in the form of a 'RegisterReply' message
1167-- See comments in 'whereisRemoteAsync'
1168registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1169registerRemoteAsync nid label pid = do
1171 sendCtrlMsg (if nid == here then Nothing else Just nid)
1172 (Register label nid (Just pid) False)
1174reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1175reregisterRemoteAsync nid label pid =
1176 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1178-- | Remove a process from the local registry (asynchronous).
1179-- This version will wait until a response is gotten from the
1180-- management process. The name must already be registered.
1181unregister :: String -> Process ()
1182unregister label = do
1183 mynid <- getSelfNode
1184 sendCtrlMsg Nothing (Register label mynid Nothing False)
1186 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1187 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1190-- | Deal with the result from an attempted registration or unregistration
1191-- by throwing an exception if necessary
1192handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()
1193handleRegistrationReply label ok owner =
1195 throwM $ ProcessRegistrationException label owner
1197-- | Remove a process from a remote registry (asynchronous).
1199-- Reply wil come in the form of a 'RegisterReply' message
1201-- See comments in 'whereisRemoteAsync'
1202unregisterRemoteAsync :: NodeId -> String -> Process ()
1203unregisterRemoteAsync nid label =
1204 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1206-- | Query the local process registry
1207whereis :: String -> Process (Maybe ProcessId)
1209 sendCtrlMsg Nothing (WhereIs label)
1210 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1211 (\(WhereIsReply _ mPid) -> return mPid)
1214-- | Query a remote process registry (asynchronous)
1216-- Reply will come in the form of a 'WhereIsReply' message.
1218-- There is currently no synchronous version of 'whereisRemoteAsync': if
1219-- you implement one yourself, be sure to take into account that the remote
1220-- node might die or get disconnect before it can respond (i.e. you should
1221-- use 'monitorNode' and take appropriate action when you receive a
1222-- 'NodeMonitorNotification').
1223whereisRemoteAsync :: NodeId -> String -> Process ()
1224whereisRemoteAsync nid label = do
1226 sendCtrlMsg (if nid == here then Nothing else Just nid) (WhereIs label)
1228-- | Named send to a process in the local registry (asynchronous)
1229nsend :: Serializable a => String -> a -> Process ()
1232 let msg' = createUnencodedMessage msg
1233 liftIO $ traceEvent (localEventBus (processNode proc))
1234 (MxSentToName label (processId proc) msg')
1235 sendCtrlMsg Nothing (NamedSend label msg')
1237-- | Named send to a process in the local registry (asynchronous).
1238-- This function makes /no/ attempt to serialize and (in the case when the
1239-- destination process resides on the same local node) therefore ensure that
1240-- the payload is fully evaluated before it is delivered.
1241unsafeNSend :: Serializable a => String -> a -> Process ()
1242unsafeNSend = Unsafe.nsend
1244-- | Named send to a process in a remote registry (asynchronous)
1245nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1246nsendRemote nid label msg = do
1248 let us = processId proc
1249 let node = processNode proc
1250 if localNodeId node == nid
1251 then nsend label msg
1252 else let lbl = label ++ "@" ++ show nid in do
1253 liftIO $ traceEvent (localEventBus node)
1254 (MxSentToName lbl us (wrapMessage msg))
1255 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1257-- | Named send to a process in a remote registry (asynchronous)
1258-- This function makes /no/ attempt to serialize and (in the case when the
1259-- destination process resides on the same local node) therefore ensure that
1260-- the payload is fully evaluated before it is delivered.
1261unsafeNSendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1262unsafeNSendRemote = Unsafe.nsendRemote
1264--------------------------------------------------------------------------------
1266--------------------------------------------------------------------------------
1268-- | Resolve a static value
1269unStatic :: Typeable a => Static a -> Process a
1271 rtable <- remoteTable . processNode <$> ask
1272 case Static.unstatic rtable static of
1273 Left err -> fail $ "Could not resolve static value: " ++ err
1276-- | Resolve a closure
1277unClosure :: Typeable a => Closure a -> Process a
1278unClosure closure = do
1279 rtable <- remoteTable . processNode <$> ask
1280 case Static.unclosure rtable closure of
1281 Left err -> fail $ "Could not resolve closure: " ++ err
1284--------------------------------------------------------------------------------
1286--------------------------------------------------------------------------------
1288-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1289-- message passing. However, when network connections get disrupted this
1290-- illusion cannot always be maintained. Once a network connection breaks (even
1291-- temporarily) no further communication on that connection will be possible.
1292-- For example, if process A sends a message to process B, and A is then
1293-- notified (by monitor notification) that it got disconnected from B, A will
1294-- not be able to send any further messages to B, /unless/ A explicitly
1295-- indicates that it is acceptable to attempt to reconnect to B using the
1296-- Cloud Haskell 'reconnect' primitive.
1298-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1299-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1300-- then receives a monitor notification that its connection to B has been lost,
1301-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1302-- and m3 but not m2.
1304-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1305-- to attempt to reconnect on the next send/. In particular, if no further
1306-- communication attempts are made to B then A can use reconnect to clean up
1307-- its connection to B.
1308reconnect :: ProcessId -> Process ()
1311 node <- processNode <$> ask
1312 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1314-- | Reconnect to a sendport. See 'reconnect' for more information.
1315reconnectPort :: SendPort a -> Process ()
1316reconnectPort them = do
1318 node <- processNode <$> ask
1319 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1321--------------------------------------------------------------------------------
1322-- Auxiliary functions --
1323--------------------------------------------------------------------------------
1325sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1327 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1329sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1330sendChanLocal spId msg =
1331 -- we *must* fully serialize/encode the message here, because
1332 -- attempting to use `unsafeCoerce' in the node controller
1333 -- won't work since we know nothing about the required type
1334 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1336getMonitorRefFor :: Identifier -> Process MonitorRef
1337getMonitorRefFor ident = do
1339 liftIO $ modifyMVar (processState proc) $ \st -> do
1340 let counter = st ^. monitorCounter
1341 return ( monitorCounter ^: (+ 1) $ st
1342 , MonitorRef ident counter
1345getSpawnRef :: Process SpawnRef
1348 liftIO $ modifyMVar (processState proc) $ \st -> do
1349 let counter = st ^. spawnCounter
1350 return ( spawnCounter ^: (+ 1) $ st
1354-- | Monitor a process/node/channel
1355monitor' :: Identifier -> Process MonitorRef
1357 monitorRef <- getMonitorRefFor ident
1358 sendCtrlMsg Nothing $ Monitor monitorRef
1361-- | Link to a process/node/channel
1362link' :: Identifier -> Process ()
1363link' = 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