2{-# LANGUAGE RankNTypes #-}
3{-# LANGUAGE DeriveDataTypeable #-}
4{-# LANGUAGE DeriveFunctor #-}
5{-# LANGUAGE ScopedTypeVariables #-}
6{-# LANGUAGE ExistentialQuantification #-}
7{-# LANGUAGE BangPatterns #-}
9-- | Cloud Haskell primitives
11-- We define these in a separate module so that we don't have to rely on
12-- the closure combinators
13module Control.Distributed.Process.Internal.Primitives
14 ( -- * Basic messaging
24 -- * Unsafe messaging variants
30 -- * Advanced messaging
56 -- * Process management
58 , ProcessTerminationException(..)
64 -- keep the exception constructor hidden, so that handling exit
65 -- reasons /must/ take place via the 'catchExit' family of primitives
66 , ProcessExitException()
74 -- * Monitoring and linking
92 , reregisterRemoteAsync
93 , unregisterRemoteAsync
99 -- * Exception handling
123 -- * Internal Exports
127#if ! MIN_VERSION_base(4,6,0)
128import Prelude hiding (catch)
131import Data.Binary (Binary(..), Put, Get, decode)
132import Data.Time.Clock (getCurrentTime, UTCTime(..))
133import Data.Time.Calendar (Day(..))
134import Data.Time.Format (formatTime)
135#if MIN_VERSION_time(1,5,0)
136import Data.Time.Format (defaultTimeLocale)
138import System.Locale (defaultTimeLocale)
140import System.Timeout (timeout)
141import Control.Monad (when, void)
142import Control.Monad.Reader (ask)
143import Control.Monad.IO.Class (liftIO)
144import Control.Monad.Catch
150import qualified Control.Monad.Catch as Catch
151import Control.Applicative
152import Control.Distributed.Process.Internal.StrictMVar
157import Control.Concurrent.STM
166import Control.Distributed.Process.Internal.CQueue
171import Control.Distributed.Process.Serializable (Serializable, fingerprint)
172import Data.Accessor ((^.), (^:), (^=))
173import Control.Distributed.Static
177import Data.Rank1Typeable (Typeable)
178import qualified Control.Distributed.Static as Static (unstatic, unclosure)
179import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
180import Control.Distributed.Process.Internal.Types
190 , NodeMonitorNotification(..)
191 , ProcessMonitorNotification(..)
201 , ProcessExitException(..)
204 , DidUnlinkProcess(..)
209 , ProcessRegistrationException(..)
211 , ProcessInfoNone(..)
215 , createUnencodedMessage
216 , ImplicitReconnect( NoImplicitReconnect)
221import Control.Distributed.Process.Internal.Messaging
228import Control.Distributed.Process.Management.Internal.Types
231import Control.Distributed.Process.Management.Internal.Trace.Types
234import Control.Distributed.Process.Internal.WeakTQueue
243--------------------------------------------------------------------------------
245--------------------------------------------------------------------------------
248send :: Serializable a => ProcessId -> a -> Process ()
249-- This requires a lookup on every send. If we want to avoid that we need to
250-- modify serializable to allow for stateful (IO) deserialization.
253 let us = processId proc
254 node = processNode proc
255 nodeId = localNodeId node
256 destNode = (processNodeId them) in do
257 case destNode == nodeId of
258 True -> sendLocal them msg
259 False -> liftIO $ sendMessage (processNode proc)
260 (ProcessIdentifier (processId proc))
261 (ProcessIdentifier them)
264 -- We do not fire the trace event until after the sending is complete;
265 -- In the remote case, 'sendMessage' can block in the networking stack.
266 liftIO $ traceEvent (localEventBus node)
267 (MxSent them us (createUnencodedMessage msg))
269-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
270-- and (in the case when the destination process resides on the same local
271-- node) therefore ensure that the payload is fully evaluated before it is
273unsafeSend :: Serializable a => ProcessId -> a -> Process ()
274unsafeSend = Unsafe.send
276-- | Send a message unreliably.
278-- Unlike 'send', this function is insensitive to 'reconnect'. It will
279-- try to send the message regardless of the history of connection failures
282-- Message passing with 'usend' is ordered for a given sender and receiver
283-- if the messages arrive at all.
285usend :: Serializable a => ProcessId -> a -> Process ()
288 let there = processNodeId them
290 then sendLocal them msg
291 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
294-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize
295-- the message when the destination process resides on the same local
296-- node. Therefore, a local receiver would need to be prepared to cope with any
297-- errors resulting from evaluation of the message.
298unsafeUSend :: Serializable a => ProcessId -> a -> Process ()
299unsafeUSend = Unsafe.usend
301-- | Wait for a message of a specific type
302expect :: forall a. Serializable a => Process a
303expect = receiveWait [match return]
305--------------------------------------------------------------------------------
307--------------------------------------------------------------------------------
309-- | Create a new typed channel
310newChan :: Serializable a => Process (SendPort a, ReceivePort a)
313 liftIO . modifyMVar (processState proc) $ \st -> do
314 let lcid = st ^. channelCounter
315 let cid = SendPortId { sendPortProcessId = processId proc
316 , sendPortLocalId = lcid
318 let sport = SendPort cid
319 chan <- liftIO newTQueueIO
320 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
321 let rport = ReceivePort $ readTQueue chan
322 let tch = TypedChannel chan'
323 return ( (channelCounter ^: (+ 1))
324 . (typedChannelWithId lcid ^= Just tch)
329 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
330 finalizer st lcid = modifyMVar_ st $
331 return . (typedChannelWithId lcid ^= Nothing)
333-- | Send a message on a typed channel
334sendChan :: Serializable a => SendPort a -> a -> Process ()
335sendChan (SendPort cid) msg = do
337 let node = localNodeId (processNode proc)
338 destNode = processNodeId (sendPortProcessId cid) in do
339 case destNode == node of
340 True -> sendChanLocal cid msg
342 liftIO $ sendBinary (processNode proc)
343 (ProcessIdentifier (processId proc))
344 (SendPortIdentifier cid)
348-- | Send a message on a typed channel. This function makes /no/ attempt to
349-- serialize and (in the case when the @ReceivePort@ resides on the same local
350-- node) therefore ensure that the payload is fully evaluated before it is
352unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
353unsafeSendChan = Unsafe.sendChan
355-- | Wait for a message on a typed channel
356receiveChan :: Serializable a => ReceivePort a -> Process a
357receiveChan = liftIO . atomically . receiveSTM
359-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
360-- non-blocking check for a message.
361receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
362receiveChanTimeout 0 ch = liftIO . atomically $
363 (Just <$> receiveSTM ch) `orElse` return Nothing
364receiveChanTimeout n ch = liftIO . timeout n . atomically $
367-- | Merge a list of typed channels.
369-- The result port is left-biased: if there are messages available on more
370-- than one port, the first available message is returned.
371mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
372mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
374-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
376mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
377mergePortsRR = \ps -> do
378 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
379 return $ ReceivePort (rr psVar)
383 rotate (x:xs) = xs ++ [x]
385 rr :: TVar [STM a] -> STM a
388 a <- foldr1 orElse ps
389 writeTVar psVar (rotate ps)
392--------------------------------------------------------------------------------
393-- Advanced messaging --
394--------------------------------------------------------------------------------
396-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
397newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
400-- | Test the matches in order against each message in the queue
401receiveWait :: [Match b] -> Process b
403 queue <- processQueue <$> ask
404 mProc <- liftIO $ dequeue queue Blocking (map unMatch ms)
407 Nothing -> die $ "System Invariant Violation: CQueue.hs returned `Nothing` "
408 ++ "in the absence of a timeout value."
410-- | Like 'receiveWait' but with a timeout.
412-- If the timeout is zero do a non-blocking check for matching messages. A
413-- non-zero timeout is applied only when waiting for incoming messages (that is,
414-- /after/ we have checked the messages that are already in the mailbox).
415receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
416receiveTimeout t ms = do
417 queue <- processQueue <$> ask
418 let blockSpec = if t == 0 then NonBlocking else Timeout t
419 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
421 Nothing -> return Nothing
422 Just proc -> Just <$> proc
424-- | Match on a typed channel
425matchChan :: ReceivePort a -> (a -> Process b) -> Match b
426matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
428-- | Match on an arbitrary STM action.
430-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
431-- actions with checks against our process' mailbox and/or any typed channel
432-- @ReceivePort@s we may hold.
434-- This allows us to process multiple input streams /along with our mailbox/,
435-- in just the same way that 'matchChan' supports checking both the mailbox
436-- /and/ an arbitrary set of typed channels in one atomic transaction.
438-- Note there are no ordering guarnatees with respect to these disparate input
441matchSTM :: STM a -> (a -> Process b) -> Match b
442matchSTM stm fn = Match $ MatchChan (fmap fn stm)
444-- | Match against any message of the right type
445match :: forall a b. Serializable a => (a -> Process b) -> Match b
446match = matchIf (const True)
448-- | Match against any message of the right type that satisfies a predicate
449matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
450matchIf c p = Match $ MatchMsg $ \msg ->
451 case messageFingerprint msg == fingerprint (undefined :: a) of
454 (UnencodedMessage _ m) ->
455 let m' = unsafeCoerce m :: a in
459 (EncodedMessage _ _) ->
460 if (c decoded) then Just (p decoded) else Nothing
463 -- Make sure the value is fully decoded so that we don't hang to
464 -- bytestrings when the process calling 'matchIf' doesn't process
465 -- the values immediately
466 !decoded = decode (messageEncoding msg)
468-- | Match against any message, regardless of the underlying (contained) type
469matchMessage :: (Message -> Process Message) -> Match Message
470matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
472-- | Match against any message (regardless of underlying type) that satisfies a predicate
473matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
474matchMessageIf c p = Match $ MatchMsg $ \msg ->
479-- | Forward a raw 'Message' to the given 'ProcessId'.
480forward :: Message -> ProcessId -> Process ()
483 let node = processNode proc
485 nid = localNodeId node
486 destNode = (processNodeId them) in do
487 case destNode == nid of
488 True -> sendCtrlMsg Nothing (LocalSend them msg)
489 False -> liftIO $ sendPayload (processNode proc)
490 (ProcessIdentifier (processId proc))
491 (ProcessIdentifier them)
493 (messageToPayload msg)
494 -- We do not fire the trace event until after the sending is complete;
495 -- In the remote case, 'sendMessage' can block in the networking stack.
496 liftIO $ traceEvent (localEventBus node)
499-- | Forward a raw 'Message' to the given 'ProcessId'.
501-- Unlike 'forward', this function is insensitive to 'reconnect'. It will
502-- try to send the message regardless of the history of connection failures
504uforward :: Message -> ProcessId -> Process ()
505uforward msg them = do
507 let node = processNode proc
509 nid = localNodeId node
510 destNode = (processNodeId them) in do
511 case destNode == nid of
512 True -> sendCtrlMsg Nothing (LocalSend them msg)
513 False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)
515 -- We do not fire the trace event until after the sending is complete;
516 -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
517 liftIO $ traceEvent (localEventBus node)
520-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
521-- 'Serializable' - like the datum they contain - but also note, deserialising
522-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
523-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
528-- send self (wrapMessage "blah")
529-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
530-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
531-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
534wrapMessage :: Serializable a => a -> Message
535wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
537-- | This is the /unsafe/ variant of 'wrapMessage'. See
538-- "Control.Distributed.Process.UnsafePrimitives" for details.
539unsafeWrapMessage :: Serializable a => a -> Message
540unsafeWrapMessage = Unsafe.wrapMessage
542-- [note Serializable UnencodedMessage]
543-- if we attempt to serialise an UnencodedMessage, it will be converted to an
544-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
545-- Primitives.hs for the gory details
547-- | Attempt to unwrap a raw 'Message'.
548-- If the type of the decoded message payload matches the expected type, the
549-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
552-- This expression, for example, will evaluate to @Nothing@
553-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
555-- Whereas this expression, will yield @Just "foo"@
556-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
558unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
560 case messageFingerprint msg == fingerprint (undefined :: a) of
561 False -> return Nothing :: m (Maybe a)
563 (UnencodedMessage _ ms) ->
564 let ms' = unsafeCoerce ms :: a
566 (EncodedMessage _ _) ->
567 return (Just (decoded))
569 decoded :: a -- note [decoding]
570 !decoded = decode (messageEncoding msg)
572-- | Attempt to handle a raw 'Message'.
573-- If the type of the message matches the type of the first argument to
574-- the supplied expression, then the message will be decoded and the expression
575-- evaluated against its value. If this runtime type checking fails however,
576-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
577-- evaluation proceeds, the resulting value with be wrapped with @Just@.
579-- Intended for use in `catchesExit` and `matchAny` primitives.
581handleMessage :: forall m a b. (Monad m, Serializable a)
582 => Message -> (a -> m b) -> m (Maybe b)
583handleMessage msg proc = handleMessageIf msg (const True) proc
585-- | Conditionally handle a raw 'Message'.
586-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
587-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
588-- for further information about runtime type checking.
590handleMessageIf :: forall m a b . (Monad m, Serializable a)
595handleMessageIf msg c proc = do
596 case messageFingerprint msg == fingerprint (undefined :: a) of
597 False -> return Nothing :: m (Maybe b)
599 (UnencodedMessage _ ms) ->
600 let ms' = unsafeCoerce ms :: a in
602 True -> do { r <- proc ms'; return (Just r) }
603 False -> return Nothing :: m (Maybe b)
604 (EncodedMessage _ _) ->
606 True -> do { r <- proc (decoded :: a); return (Just r) }
607 False -> return Nothing :: m (Maybe b)
609 decoded :: a -- note [decoding]
610 !decoded = decode (messageEncoding msg)
612-- | As 'handleMessage' but ignores result, which is useful if you don't
613-- care whether or not the handler succeeded.
615handleMessage_ :: forall m a . (Monad m, Serializable a)
616 => Message -> (a -> m ()) -> m ()
617handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
619-- | Conditional version of 'handleMessage_'.
621handleMessageIf_ :: forall m a . (Monad m, Serializable a)
626handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
628-- | Match against an arbitrary message. 'matchAny' removes the first available
629-- message from the process mailbox. To handle arbitrary /raw/ messages once
630-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
632matchAny :: forall b. (Message -> Process b) -> Match b
633matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
635-- | Match against an arbitrary message. Intended for use with 'handleMessage'
636-- and 'unwrapMessage', this function /only/ removes a message from the process
637-- mailbox, /if/ the supplied condition matches. The success (or failure) of
638-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
639-- here, i.e., if the condition evaluates to @True@ then the message will
640-- be removed from the process mailbox and decoded, but that does /not/
641-- guarantee that an expression passed to @handleMessage@ will pass the
642-- runtime type checks and therefore be evaluated.
644matchAnyIf :: forall a b. (Serializable a)
646 -> (Message -> Process b)
648matchAnyIf c p = Match $ MatchMsg $ \msg ->
649 case messageFingerprint msg == fingerprint (undefined :: a) of
650 True | check -> Just (p msg)
655 (EncodedMessage _ _) -> c decoded
656 (UnencodedMessage _ m') -> c (unsafeCoerce m')
658 decoded :: a -- note [decoding]
659 !decoded = decode (messageEncoding msg)
663For an EncodedMessage, we need to ensure the value is fully decoded so that
664we don't hang to bytestrings if the calling process doesn't evaluate
665immediately. For UnencodedMessage we know (because the fingerprint comparison
666succeeds) that unsafeCoerce will not fail.
669-- | Remove any message from the queue
670matchUnknown :: Process b -> Match b
671matchUnknown p = Match $ MatchMsg (const (Just p))
673-- | Receives messages and forwards them to @pid@ if @p msg == True@.
674delegate :: ProcessId -> (Message -> Bool) -> Process ()
677 matchAny (\m -> case (p m) of
678 True -> forward m pid
683-- | A straight relay that forwards all messages to the supplied pid.
684relay :: ProcessId -> Process ()
685relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
687-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
688-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
689-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
690-- fails, no action is taken and the proxy process will continue running.
691proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
695 next <- handleMessage m proc
697 Just True -> forward m pid
698 Just False -> return () -- explicitly ignored
699 Nothing -> return ()) -- un-routable
703--------------------------------------------------------------------------------
704-- Process management --
705--------------------------------------------------------------------------------
707-- | Thrown by 'terminate'
708data ProcessTerminationException = ProcessTerminationException
709 deriving (Show, Typeable)
711instance Exception ProcessTerminationException
713-- | Terminate immediately (throws a ProcessTerminationException)
714terminate :: Process a
715terminate = throwM ProcessTerminationException
718-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
719die :: Serializable a => a -> Process b
721-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
722-- could be decoded by a handler passed to 'catchExit', re-thrown or even
723-- passed to another process via the tracing mechanism.
725 throwM (ProcessExitException pid (createMessage reason))
727-- | Forceful request to kill a process. Where 'exit' provides an exception
728-- that can be caught and handled, 'kill' throws an unexposed exception type
729-- which cannot be handled explicitly (by type).
730kill :: ProcessId -> String -> Process ()
731-- NOTE: We send the message to our local node controller, which will then
732-- forward it to a remote node controller (if applicable). Sending it directly
733-- to a remote node controller means that that the message may overtake a
734-- 'monitor' or 'link' request.
735kill them reason = sendCtrlMsg Nothing (Kill them reason)
737-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
738-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
739-- manner can be handled using the 'catchExit' family of functions.
740exit :: Serializable a => ProcessId -> a -> Process ()
741-- NOTE: We send the message to our local node controller, which will then
742-- forward it to a remote node controller (if applicable). Sending it directly
743-- to a remote node controller means that that the message may overtake a
744-- 'monitor' or 'link' request.
745exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
747-- | Catches 'ProcessExitException'. The handler will not be applied unless its
748-- type matches the encoded data stored in the exception (see the /reason/
749-- argument given to the 'exit' primitive). If the handler cannot be applied,
750-- the exception will be re-thrown.
752-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
753-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
754catchExit :: forall a b . (Show a, Serializable a)
756 -> (ProcessId -> a -> Process b)
758catchExit act exitHandler = Catch.catch act handleExit
760 handleExit ex@(ProcessExitException from msg) =
761 if messageFingerprint msg == fingerprint (undefined :: a)
762 then exitHandler from decoded
766 -- Make sure the value is fully decoded so that we don't hang to
767 -- bytestrings if the caller doesn't use the value immediately
768 !decoded = decode (messageEncoding msg)
770-- | Lift 'Control.Exception.catches' (almost).
772-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
773-- message, a handler must accept inputs of the expected type. In order to
774-- handle a list of potentially different handlers (and therefore input types),
775-- a handler passed to 'catchesExit' must accept 'Message' and return
776-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
778-- See 'maybeHandleMessage' and 'Message' for more details.
779catchesExit :: Process b
780 -> [(ProcessId -> Message -> (Process (Maybe b)))]
782catchesExit act handlers = Catch.catch act ((flip handleExit) handlers)
784 handleExit :: ProcessExitException
785 -> [(ProcessId -> Message -> Process (Maybe b))]
787 handleExit ex [] = throwM ex
788 handleExit ex@(ProcessExitException from msg) (h:hs) = do
791 Nothing -> handleExit ex hs
794-- | Our own process ID
795getSelfPid :: Process ProcessId
796getSelfPid = processId <$> ask
798-- | Get the node ID of our local node
799getSelfNode :: Process NodeId
800getSelfNode = localNodeId . processNode <$> ask
803-- | Get statistics about the specified node
804getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
806 selfNode <- getSelfNode
808 then Right `fmap` getLocalNodeStats -- optimisation
809 else getNodeStatsRemote selfNode
811 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
812 getNodeStatsRemote selfNode = do
813 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
814 bracket (monitorNode nid) unmonitor $ \mRef ->
815 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
816 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
817 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
820-- | Get statistics about our local node
821getLocalNodeStats :: Process NodeStats
822getLocalNodeStats = do
824 sendCtrlMsg Nothing $ GetNodeStats self
825 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
827-- | Get information about the specified process
828getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
830 let them = processNodeId pid in do
832 dest <- mkNode them us
833 sendCtrlMsg dest $ GetInfo pid
835 match (\(p :: ProcessInfo) -> return $ Just p)
836 , match (\(_ :: ProcessInfoNone) -> return Nothing)
838 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
839 mkNode them us = case them == us of
840 True -> return Nothing
841 _ -> return $ Just them
843--------------------------------------------------------------------------------
844-- Monitoring and linking --
845--------------------------------------------------------------------------------
847-- | Link to a remote process (asynchronous)
849-- When process A links to process B (that is, process A calls
850-- @link pidB@) then an asynchronous exception will be thrown to process A
851-- when process B terminates (normally or abnormally), or when process A gets
852-- disconnected from process B. Although it is /technically/ possible to catch
853-- these exceptions, chances are if you find yourself trying to do so you should
854-- probably be using 'monitor' rather than 'link'. In particular, code such as
856-- > link pidB -- Link to process B
857-- > expect -- Wait for a message from process B
858-- > unlink pidB -- Unlink again
860-- doesn't quite do what one might expect: if process B sends a message to
861-- process A, and /subsequently terminates/, then process A might or might not
862-- be terminated too, depending on whether the exception is thrown before or
863-- after the 'unlink' (i.e., this code has a race condition).
865-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
866-- call to 'link' has no effect.
868-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
869-- Linking makes no distinction between normal and abnormal termination of
870-- the remote process.
871link :: ProcessId -> Process ()
872link = sendCtrlMsg Nothing . Link . ProcessIdentifier
874-- | Monitor another process (asynchronous)
876-- When process A monitors process B (that is, process A calls
877-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
878-- when process B terminates (normally or abnormally), or when process A gets
879-- disconnected from process B. You receive this message like any other (using
880-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
881-- 'DiedDisconnect', etc.).
883-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
884-- multiple monitors are set up, multiple notifications will be delivered
885-- and monitors can be disabled individually using 'unmonitor'.
886monitor :: ProcessId -> Process MonitorRef
887monitor = monitor' . ProcessIdentifier
889-- | Establishes temporary monitoring of another process.
891-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
892-- of @code@. Note: although monitoring is no longer active when
893-- @withMonitor@ returns, there might still be unreceived monitor
894-- messages in the queue.
896withMonitor :: ProcessId -> (MonitorRef -> Process a) -> Process a
897withMonitor pid = bracket (monitor pid) unmonitor
899-- | Establishes temporary monitoring of another process.
901-- @withMonitor_ pid code@ sets up monitoring of @pid@ for the duration
902-- of @code@. Note: although monitoring is no longer active when
903-- @withMonitor_@ returns, there might still be unreceived monitor
904-- messages in the queue.
907withMonitor_ :: ProcessId -> Process a -> Process a
908withMonitor_ p = withMonitor p . const
912-- This is synchronous in the sense that once it returns you are guaranteed
913-- that no exception will be raised if the remote process dies. However, it is
914-- asynchronous in the sense that we do not wait for a response from the remote
916unlink :: ProcessId -> Process ()
919 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
923-- | Remove a node link
925-- This has the same synchronous/asynchronous nature as 'unlink'.
926unlinkNode :: NodeId -> Process ()
929 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
933-- | Remove a channel (send port) link
935-- This has the same synchronous/asynchronous nature as 'unlink'.
936unlinkPort :: SendPort a -> Process ()
938 unlinkPortAsync sport
939 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
945-- This has the same synchronous/asynchronous nature as 'unlink'.
947-- ProcessMonitorNotification messages for the given MonitorRef are removed from
949unmonitor :: MonitorRef -> Process ()
952 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
955 -- Discard the notification if any. With the current NC implementation at most
956 -- one notification is in the mailbox for any given ref.
957 void $ receiveTimeout 0
958 [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref' == ref)
962--------------------------------------------------------------------------------
963-- Exception handling --
964--------------------------------------------------------------------------------
966-- | Lift 'Control.Exception.catch'
967catch :: Exception e => Process a -> (e -> Process a) -> Process a
969{-# DEPRECATED catch "Use Control.Monad.Catch.catch instead" #-}
971-- | Lift 'Control.Exception.try'
972try :: Exception e => Process a -> Process (Either e a)
974{-# DEPRECATED try "Use Control.Monad.Catch.try instead" #-}
976-- | Lift 'Control.Exception.mask'
977mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
979{-# DEPRECATED mask "Use Control.Monad.Catch.mask_ instead" #-}
981-- | Lift 'Control.Exception.mask_'
982mask_ :: Process a -> Process a
984{-# DEPRECATED mask_ "Use Control.Monad.Catch.mask_ instead" #-}
986-- | Lift 'Control.Exception.onException'
987onException :: Process a -> Process b -> Process a
988onException = Catch.onException
989{-# DEPRECATED onException "Use Control.Monad.Catch.onException instead" #-}
991-- | Lift 'Control.Exception.bracket'
992bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
993bracket = Catch.bracket
994{-# DEPRECATED bracket "Use Control.Monad.Catch.bracket instead" #-}
996-- | Lift 'Control.Exception.bracket_'
997bracket_ :: Process a -> Process b -> Process c -> Process c
998bracket_ = Catch.bracket_
999{-# DEPRECATED bracket_ "Use Control.Monad.Catch.bracket_ instead" #-}
1001-- | Lift 'Control.Exception.finally'
1002finally :: Process a -> Process b -> Process a
1003finally = Catch.finally
1004{-# DEPRECATED finally "Use Control.Monad.Catch.finally instead" #-}
1006-- | You need this when using 'catches'
1007data Handler a = forall e . Exception e => Handler (e -> Process a)
1009instance Functor Handler where
1010 fmap f (Handler h) = Handler (fmap f . h)
1012-- | Lift 'Control.Exception.catches'
1013catches :: Process a -> [Handler a] -> Process a
1014catches proc handlers = proc `Catch.catch` catchesHandler handlers
1016catchesHandler :: [Handler a] -> SomeException -> Process a
1017catchesHandler handlers e = foldr tryHandler (throwM e) handlers
1018 where tryHandler (Handler handler) res
1019 = case fromException e of
1020 Just e' -> handler e'
1023--------------------------------------------------------------------------------
1025--------------------------------------------------------------------------------
1027-- | Like 'expect' but with a timeout
1028expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
1029expectTimeout n = receiveTimeout n [match return]
1031-- | Asynchronous version of 'spawn'
1033-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
1034spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
1035spawnAsync nid proc = do
1036 spawnRef <- getSpawnRef
1039 then sendCtrlMsg Nothing $ Spawn proc spawnRef
1040 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
1043-- | Monitor a node (asynchronous)
1044monitorNode :: NodeId -> Process MonitorRef
1046 monitor' . NodeIdentifier
1048-- | Monitor a typed channel (asynchronous)
1049monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1050monitorPort (SendPort cid) =
1051 monitor' (SendPortIdentifier cid)
1053-- | Remove a monitor (asynchronous)
1054unmonitorAsync :: MonitorRef -> Process ()
1056 sendCtrlMsg Nothing . Unmonitor
1058-- | Link to a node (asynchronous)
1059linkNode :: NodeId -> Process ()
1060linkNode = link' . NodeIdentifier
1062-- | Link to a channel (asynchronous)
1063linkPort :: SendPort a -> Process ()
1064linkPort (SendPort cid) =
1065 link' (SendPortIdentifier cid)
1067-- | Remove a process link (asynchronous)
1068unlinkAsync :: ProcessId -> Process ()
1070 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1072-- | Remove a node link (asynchronous)
1073unlinkNodeAsync :: NodeId -> Process ()
1075 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1077-- | Remove a channel (send port) link (asynchronous)
1078unlinkPortAsync :: SendPort a -> Process ()
1079unlinkPortAsync (SendPort cid) =
1080 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1082--------------------------------------------------------------------------------
1084--------------------------------------------------------------------------------
1086data SayMessage = SayMessage { sayTime :: UTCTime
1087 , sayProcess :: ProcessId
1088 , sayMessage :: String }
1091-- There is sadly no Show UTCTime instance
1092instance Show SayMessage where
1095 $ showString "SayMessage "
1096 . showString (formatTime defaultTimeLocale "%c" (sayTime msg))
1098 . showsPrec 11 (sayProcess msg) . showChar ' '
1099 . showsPrec 11 (sayMessage msg) . showChar ' '
1101instance Binary SayMessage where
1103 putUTCTime (sayTime s)
1106 get = SayMessage <$> getUTCTime <*> get <*> get
1108-- Sadly there is no Binary UTCTime instance
1109putUTCTime :: UTCTime -> Put
1110putUTCTime (UTCTime (ModifiedJulianDay day) tod) = do
1112 put (toRational tod)
1114getUTCTime :: Get UTCTime
1118 return $! UTCTime (ModifiedJulianDay day)
1123-- @say message@ sends a message of type 'SayMessage' with the current time and
1124-- 'ProcessId' of the current process to the process registered as @logger@. By
1125-- default, this process simply sends the string to @stderr@. Individual Cloud
1126-- Haskell backends might replace this with a different logger process, however.
1127say :: String -> Process ()
1129 now <- liftIO getCurrentTime
1131 nsend "logger" (SayMessage now us string)
1133--------------------------------------------------------------------------------
1135--------------------------------------------------------------------------------
1137-- | Register a process with the local registry (synchronous). The name must not
1138-- already be registered. The process need not be on this node. A bad
1139-- registration will result in a 'ProcessRegistrationException'
1141-- The process to be registered does not have to be local itself.
1142register :: String -> ProcessId -> Process ()
1143register = registerImpl False
1145-- | Like 'register', but will replace an existing registration.
1146-- The name must already be registered.
1147reregister :: String -> ProcessId -> Process ()
1148reregister = registerImpl True
1150registerImpl :: Bool -> String -> ProcessId -> Process ()
1151registerImpl force label pid = do
1152 mynid <- getSelfNode
1153 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1155 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1156 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1159-- | Register a process with a remote registry (asynchronous).
1161-- The process to be registered does not have to live on the same remote node.
1162-- Reply wil come in the form of a 'RegisterReply' message
1164-- See comments in 'whereisRemoteAsync'
1165registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1166registerRemoteAsync nid label pid = do
1168 sendCtrlMsg (if nid == here then Nothing else Just nid)
1169 (Register label nid (Just pid) False)
1171reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1172reregisterRemoteAsync nid label pid =
1173 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1175-- | Remove a process from the local registry (asynchronous).
1176-- This version will wait until a response is gotten from the
1177-- management process. The name must already be registered.
1178unregister :: String -> Process ()
1179unregister label = do
1180 mynid <- getSelfNode
1181 sendCtrlMsg Nothing (Register label mynid Nothing False)
1183 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1184 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1187-- | Deal with the result from an attempted registration or unregistration
1188-- by throwing an exception if necessary
1189handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()
1190handleRegistrationReply label ok owner =
1192 throwM $ ProcessRegistrationException label owner
1194-- | Remove a process from a remote registry (asynchronous).
1196-- Reply wil come in the form of a 'RegisterReply' message
1198-- See comments in 'whereisRemoteAsync'
1199unregisterRemoteAsync :: NodeId -> String -> Process ()
1200unregisterRemoteAsync nid label =
1201 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1203-- | Query the local process registry
1204whereis :: String -> Process (Maybe ProcessId)
1206 sendCtrlMsg Nothing (WhereIs label)
1207 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1208 (\(WhereIsReply _ mPid) -> return mPid)
1211-- | Query a remote process registry (asynchronous)
1213-- Reply will come in the form of a 'WhereIsReply' message.
1215-- There is currently no synchronous version of 'whereisRemoteAsync': if
1216-- you implement one yourself, be sure to take into account that the remote
1217-- node might die or get disconnect before it can respond (i.e. you should
1218-- use 'monitorNode' and take appropriate action when you receive a
1219-- 'NodeMonitorNotification').
1220whereisRemoteAsync :: NodeId -> String -> Process ()
1221whereisRemoteAsync nid label = do
1223 sendCtrlMsg (if nid == here then Nothing else Just nid) (WhereIs label)
1225-- | Named send to a process in the local registry (asynchronous)
1226nsend :: Serializable a => String -> a -> Process ()
1228 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1230-- | Named send to a process in the local registry (asynchronous).
1231-- This function makes /no/ attempt to serialize and (in the case when the
1232-- destination process resides on the same local node) therefore ensure that
1233-- the payload is fully evaluated before it is delivered.
1234unsafeNSend :: Serializable a => String -> a -> Process ()
1235unsafeNSend = Unsafe.nsend
1237-- | Named send to a process in a remote registry (asynchronous)
1238nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1239nsendRemote nid label msg = do
1241 if here == nid then nsend label msg
1242 else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1244-- | Named send to a process in a remote registry (asynchronous)
1245-- This function makes /no/ attempt to serialize and (in the case when the
1246-- destination process resides on the same local node) therefore ensure that
1247-- the payload is fully evaluated before it is delivered.
1248unsafeNSendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1249unsafeNSendRemote = Unsafe.nsendRemote
1251--------------------------------------------------------------------------------
1253--------------------------------------------------------------------------------
1255-- | Resolve a static value
1256unStatic :: Typeable a => Static a -> Process a
1258 rtable <- remoteTable . processNode <$> ask
1259 case Static.unstatic rtable static of
1260 Left err -> fail $ "Could not resolve static value: " ++ err
1263-- | Resolve a closure
1264unClosure :: Typeable a => Closure a -> Process a
1265unClosure closure = do
1266 rtable <- remoteTable . processNode <$> ask
1267 case Static.unclosure rtable closure of
1268 Left err -> fail $ "Could not resolve closure: " ++ err
1271--------------------------------------------------------------------------------
1273--------------------------------------------------------------------------------
1275-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1276-- message passing. However, when network connections get disrupted this
1277-- illusion cannot always be maintained. Once a network connection breaks (even
1278-- temporarily) no further communication on that connection will be possible.
1279-- For example, if process A sends a message to process B, and A is then
1280-- notified (by monitor notification) that it got disconnected from B, A will
1281-- not be able to send any further messages to B, /unless/ A explicitly
1282-- indicates that it is acceptable to attempt to reconnect to B using the
1283-- Cloud Haskell 'reconnect' primitive.
1285-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1286-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1287-- then receives a monitor notification that its connection to B has been lost,
1288-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1289-- and m3 but not m2.
1291-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1292-- to attempt to reconnect on the next send/. In particular, if no further
1293-- communication attempts are made to B then A can use reconnect to clean up
1294-- its connection to B.
1295reconnect :: ProcessId -> Process ()
1298 node <- processNode <$> ask
1299 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1301-- | Reconnect to a sendport. See 'reconnect' for more information.
1302reconnectPort :: SendPort a -> Process ()
1303reconnectPort them = do
1305 node <- processNode <$> ask
1306 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1308--------------------------------------------------------------------------------
1309-- Auxiliary functions --
1310--------------------------------------------------------------------------------
1312sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1314 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1316sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1317sendChanLocal spId msg =
1318 -- we *must* fully serialize/encode the message here, because
1319 -- attempting to use `unsafeCoerce' in the node controller
1320 -- won't work since we know nothing about the required type
1321 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1323getMonitorRefFor :: Identifier -> Process MonitorRef
1324getMonitorRefFor ident = do
1326 liftIO $ modifyMVar (processState proc) $ \st -> do
1327 let counter = st ^. monitorCounter
1328 return ( monitorCounter ^: (+ 1) $ st
1329 , MonitorRef ident counter
1332getSpawnRef :: Process SpawnRef
1335 liftIO $ modifyMVar (processState proc) $ \st -> do
1336 let counter = st ^. spawnCounter
1337 return ( spawnCounter ^: (+ 1) $ st
1341-- | Monitor a process/node/channel
1342monitor' :: Identifier -> Process MonitorRef
1344 monitorRef <- getMonitorRefFor ident
1345 sendCtrlMsg Nothing $ Monitor monitorRef
1348-- | Link to a process/node/channel
1349link' :: Identifier -> Process ()
1350link' = 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