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
89 , reregisterRemoteAsync
90 , unregisterRemoteAsync
96 -- * Exception handling
120 -- * Internal Exports
124#if ! MIN_VERSION_base(4,6,0)
125import Prelude hiding (catch)
128import Data.Binary (decode)
129import Data.Time.Clock (getCurrentTime)
130import Data.Time.Format (formatTime)
131#if MIN_VERSION_time(1,5,0)
132import Data.Time.Format (defaultTimeLocale)
134import System.Locale (defaultTimeLocale)
136import System.Timeout (timeout)
137import Control.Monad (when)
138import Control.Monad.Reader (ask)
139import Control.Monad.IO.Class (liftIO)
140import Control.Monad.Catch
146import qualified Control.Monad.Catch as Catch
147import Control.Applicative
148import Control.Distributed.Process.Internal.StrictMVar
153import Control.Concurrent.STM
162import Control.Distributed.Process.Internal.CQueue
167import Control.Distributed.Process.Serializable (Serializable, fingerprint)
168import Data.Accessor ((^.), (^:), (^=))
169import Control.Distributed.Static
173import Data.Rank1Typeable (Typeable)
174import qualified Control.Distributed.Static as Static (unstatic, unclosure)
175import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
176import Control.Distributed.Process.Internal.Types
186 , NodeMonitorNotification(..)
196 , ProcessExitException(..)
199 , DidUnlinkProcess(..)
204 , ProcessRegistrationException(..)
206 , ProcessInfoNone(..)
210 , createUnencodedMessage
211 , ImplicitReconnect( NoImplicitReconnect)
216import Control.Distributed.Process.Internal.Messaging
223import Control.Distributed.Process.Management.Internal.Types
226import Control.Distributed.Process.Management.Internal.Trace.Types
229import Control.Distributed.Process.Internal.WeakTQueue
238--------------------------------------------------------------------------------
240--------------------------------------------------------------------------------
243send :: Serializable a => ProcessId -> a -> Process ()
244-- This requires a lookup on every send. If we want to avoid that we need to
245-- modify serializable to allow for stateful (IO) deserialization.
248 let us = processId proc
249 node = processNode proc
250 nodeId = localNodeId node
251 destNode = (processNodeId them) in do
252 case destNode == nodeId of
253 True -> sendLocal them msg
254 False -> liftIO $ sendMessage (processNode proc)
255 (ProcessIdentifier (processId proc))
256 (ProcessIdentifier them)
259 -- We do not fire the trace event until after the sending is complete;
260 -- In the remote case, 'sendMessage' can block in the networking stack.
261 liftIO $ traceEvent (localEventBus node)
262 (MxSent them us (createUnencodedMessage msg))
264-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
265-- and (in the case when the destination process resides on the same local
266-- node) therefore ensure that the payload is fully evaluated before it is
268unsafeSend :: Serializable a => ProcessId -> a -> Process ()
269unsafeSend = Unsafe.send
271-- | Send a message unreliably.
273-- Unlike 'send', this function is insensitive to 'reconnect'. It will
274-- try to send the message regardless of the history of connection failures
277-- Message passing with 'usend' is ordered for a given sender and receiver
278-- if the messages arrive at all.
280usend :: Serializable a => ProcessId -> a -> Process ()
283 let there = processNodeId them
285 then sendLocal them msg
286 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
289-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize
290-- the message when the destination process resides on the same local
291-- node. Therefore, a local receiver would need to be prepared to cope with any
292-- errors resulting from evaluation of the message.
293unsafeUSend :: Serializable a => ProcessId -> a -> Process ()
294unsafeUSend = Unsafe.usend
296-- | Wait for a message of a specific type
297expect :: forall a. Serializable a => Process a
298expect = receiveWait [match return]
300--------------------------------------------------------------------------------
302--------------------------------------------------------------------------------
304-- | Create a new typed channel
305newChan :: Serializable a => Process (SendPort a, ReceivePort a)
308 liftIO . modifyMVar (processState proc) $ \st -> do
309 let lcid = st ^. channelCounter
310 let cid = SendPortId { sendPortProcessId = processId proc
311 , sendPortLocalId = lcid
313 let sport = SendPort cid
314 chan <- liftIO newTQueueIO
315 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
316 let rport = ReceivePort $ readTQueue chan
317 let tch = TypedChannel chan'
318 return ( (channelCounter ^: (+ 1))
319 . (typedChannelWithId lcid ^= Just tch)
324 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
325 finalizer st lcid = modifyMVar_ st $
326 return . (typedChannelWithId lcid ^= Nothing)
328-- | Send a message on a typed channel
329sendChan :: Serializable a => SendPort a -> a -> Process ()
330sendChan (SendPort cid) msg = do
332 let node = localNodeId (processNode proc)
333 destNode = processNodeId (sendPortProcessId cid) in do
334 case destNode == node of
335 True -> sendChanLocal cid msg
337 liftIO $ sendBinary (processNode proc)
338 (ProcessIdentifier (processId proc))
339 (SendPortIdentifier cid)
343-- | Send a message on a typed channel. This function makes /no/ attempt to
344-- serialize and (in the case when the @ReceivePort@ resides on the same local
345-- node) therefore ensure that the payload is fully evaluated before it is
347unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
348unsafeSendChan = Unsafe.sendChan
350-- | Wait for a message on a typed channel
351receiveChan :: Serializable a => ReceivePort a -> Process a
352receiveChan = liftIO . atomically . receiveSTM
354-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
355-- non-blocking check for a message.
356receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
357receiveChanTimeout 0 ch = liftIO . atomically $
358 (Just <$> receiveSTM ch) `orElse` return Nothing
359receiveChanTimeout n ch = liftIO . timeout n . atomically $
362-- | Merge a list of typed channels.
364-- The result port is left-biased: if there are messages available on more
365-- than one port, the first available message is returned.
366mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
367mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
369-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
371mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
372mergePortsRR = \ps -> do
373 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
374 return $ ReceivePort (rr psVar)
378 rotate (x:xs) = xs ++ [x]
380 rr :: TVar [STM a] -> STM a
383 a <- foldr1 orElse ps
384 writeTVar psVar (rotate ps)
387--------------------------------------------------------------------------------
388-- Advanced messaging --
389--------------------------------------------------------------------------------
391-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
392newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
395-- | Test the matches in order against each message in the queue
396receiveWait :: [Match b] -> Process b
398 queue <- processQueue <$> ask
399 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
402-- | Like 'receiveWait' but with a timeout.
404-- If the timeout is zero do a non-blocking check for matching messages. A
405-- non-zero timeout is applied only when waiting for incoming messages (that is,
406-- /after/ we have checked the messages that are already in the mailbox).
407receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
408receiveTimeout t ms = do
409 queue <- processQueue <$> ask
410 let blockSpec = if t == 0 then NonBlocking else Timeout t
411 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
413 Nothing -> return Nothing
414 Just proc -> Just <$> proc
416-- | Match on a typed channel
417matchChan :: ReceivePort a -> (a -> Process b) -> Match b
418matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
420-- | Match on an arbitrary STM action.
422-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
423-- actions with checks against our process' mailbox and/or any typed channel
424-- @ReceivePort@s we may hold.
426-- This allows us to process multiple input streams /along with our mailbox/,
427-- in just the same way that 'matchChan' supports checking both the mailbox
428-- /and/ an arbitrary set of typed channels in one atomic transaction.
430-- Note there are no ordering guarnatees with respect to these disparate input
433matchSTM :: STM a -> (a -> Process b) -> Match b
434matchSTM stm fn = Match $ MatchChan (fmap fn stm)
436-- | Match against any message of the right type
437match :: forall a b. Serializable a => (a -> Process b) -> Match b
438match = matchIf (const True)
440-- | Match against any message of the right type that satisfies a predicate
441matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
442matchIf c p = Match $ MatchMsg $ \msg ->
443 case messageFingerprint msg == fingerprint (undefined :: a) of
446 (UnencodedMessage _ m) ->
447 let m' = unsafeCoerce m :: a in
451 (EncodedMessage _ _) ->
452 if (c decoded) then Just (p decoded) else Nothing
455 -- Make sure the value is fully decoded so that we don't hang to
456 -- bytestrings when the process calling 'matchIf' doesn't process
457 -- the values immediately
458 !decoded = decode (messageEncoding msg)
460-- | Match against any message, regardless of the underlying (contained) type
461matchMessage :: (Message -> Process Message) -> Match Message
462matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
464-- | Match against any message (regardless of underlying type) that satisfies a predicate
465matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
466matchMessageIf c p = Match $ MatchMsg $ \msg ->
471-- | Forward a raw 'Message' to the given 'ProcessId'.
472forward :: Message -> ProcessId -> Process ()
475 let node = processNode proc
477 nid = localNodeId node
478 destNode = (processNodeId them) in do
479 case destNode == nid of
480 True -> sendCtrlMsg Nothing (LocalSend them msg)
481 False -> liftIO $ sendPayload (processNode proc)
482 (ProcessIdentifier (processId proc))
483 (ProcessIdentifier them)
485 (messageToPayload msg)
486 -- We do not fire the trace event until after the sending is complete;
487 -- In the remote case, 'sendMessage' can block in the networking stack.
488 liftIO $ traceEvent (localEventBus node)
491-- | Forward a raw 'Message' to the given 'ProcessId'.
493-- Unlike 'forward', this function is insensitive to 'reconnect'. It will
494-- try to send the message regardless of the history of connection failures
496uforward :: Message -> ProcessId -> Process ()
497uforward msg them = do
499 let node = processNode proc
501 nid = localNodeId node
502 destNode = (processNodeId them) in do
503 case destNode == nid of
504 True -> sendCtrlMsg Nothing (LocalSend them msg)
505 False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)
507 -- We do not fire the trace event until after the sending is complete;
508 -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
509 liftIO $ traceEvent (localEventBus node)
512-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
513-- 'Serializable' - like the datum they contain - but also note, deserialising
514-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
515-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
520-- send self (wrapMessage "blah")
521-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
522-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
523-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
526wrapMessage :: Serializable a => a -> Message
527wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
529-- | This is the /unsafe/ variant of 'wrapMessage'. See
530-- "Control.Distributed.Process.UnsafePrimitives" for details.
531unsafeWrapMessage :: Serializable a => a -> Message
532unsafeWrapMessage = Unsafe.wrapMessage
534-- [note Serializable UnencodedMessage]
535-- if we attempt to serialise an UnencodedMessage, it will be converted to an
536-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
537-- Primitives.hs for the gory details
539-- | Attempt to unwrap a raw 'Message'.
540-- If the type of the decoded message payload matches the expected type, the
541-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
544-- This expression, for example, will evaluate to @Nothing@
545-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
547-- Whereas this expression, will yield @Just "foo"@
548-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
550unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
552 case messageFingerprint msg == fingerprint (undefined :: a) of
553 False -> return Nothing :: m (Maybe a)
555 (UnencodedMessage _ ms) ->
556 let ms' = unsafeCoerce ms :: a
558 (EncodedMessage _ _) ->
559 return (Just (decoded))
561 decoded :: a -- note [decoding]
562 !decoded = decode (messageEncoding msg)
564-- | Attempt to handle a raw 'Message'.
565-- If the type of the message matches the type of the first argument to
566-- the supplied expression, then the message will be decoded and the expression
567-- evaluated against its value. If this runtime type checking fails however,
568-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
569-- evaluation proceeds, the resulting value with be wrapped with @Just@.
571-- Intended for use in `catchesExit` and `matchAny` primitives.
573handleMessage :: forall m a b. (Monad m, Serializable a)
574 => Message -> (a -> m b) -> m (Maybe b)
575handleMessage msg proc = handleMessageIf msg (const True) proc
577-- | Conditionally handle a raw 'Message'.
578-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
579-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
580-- for further information about runtime type checking.
582handleMessageIf :: forall m a b . (Monad m, Serializable a)
587handleMessageIf msg c proc = do
588 case messageFingerprint msg == fingerprint (undefined :: a) of
589 False -> return Nothing :: m (Maybe b)
591 (UnencodedMessage _ ms) ->
592 let ms' = unsafeCoerce ms :: a in
594 True -> do { r <- proc ms'; return (Just r) }
595 False -> return Nothing :: m (Maybe b)
596 (EncodedMessage _ _) ->
598 True -> do { r <- proc (decoded :: a); return (Just r) }
599 False -> return Nothing :: m (Maybe b)
601 decoded :: a -- note [decoding]
602 !decoded = decode (messageEncoding msg)
604-- | As 'handleMessage' but ignores result, which is useful if you don't
605-- care whether or not the handler succeeded.
607handleMessage_ :: forall m a . (Monad m, Serializable a)
608 => Message -> (a -> m ()) -> m ()
609handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
611-- | Conditional version of 'handleMessage_'.
613handleMessageIf_ :: forall m a . (Monad m, Serializable a)
618handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
620-- | Match against an arbitrary message. 'matchAny' removes the first available
621-- message from the process mailbox. To handle arbitrary /raw/ messages once
622-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
624matchAny :: forall b. (Message -> Process b) -> Match b
625matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
627-- | Match against an arbitrary message. Intended for use with 'handleMessage'
628-- and 'unwrapMessage', this function /only/ removes a message from the process
629-- mailbox, /if/ the supplied condition matches. The success (or failure) of
630-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
631-- here, i.e., if the condition evaluates to @True@ then the message will
632-- be removed from the process mailbox and decoded, but that does /not/
633-- guarantee that an expression passed to @handleMessage@ will pass the
634-- runtime type checks and therefore be evaluated.
636matchAnyIf :: forall a b. (Serializable a)
638 -> (Message -> Process b)
640matchAnyIf c p = Match $ MatchMsg $ \msg ->
641 case messageFingerprint msg == fingerprint (undefined :: a) of
642 True | check -> Just (p msg)
647 (EncodedMessage _ _) -> c decoded
648 (UnencodedMessage _ m') -> c (unsafeCoerce m')
650 decoded :: a -- note [decoding]
651 !decoded = decode (messageEncoding msg)
655For an EncodedMessage, we need to ensure the value is fully decoded so that
656we don't hang to bytestrings if the calling process doesn't evaluate
657immediately. For UnencodedMessage we know (because the fingerprint comparison
658succeeds) that unsafeCoerce will not fail.
661-- | Remove any message from the queue
662matchUnknown :: Process b -> Match b
663matchUnknown p = Match $ MatchMsg (const (Just p))
665-- | Receives messages and forwards them to @pid@ if @p msg == True@.
666delegate :: ProcessId -> (Message -> Bool) -> Process ()
669 matchAny (\m -> case (p m) of
670 True -> forward m pid
675-- | A straight relay that forwards all messages to the supplied pid.
676relay :: ProcessId -> Process ()
677relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
679-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
680-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
681-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
682-- fails, no action is taken and the proxy process will continue running.
683proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
687 next <- handleMessage m proc
689 Just True -> forward m pid
690 Just False -> return () -- explicitly ignored
691 Nothing -> return ()) -- un-routable
695--------------------------------------------------------------------------------
696-- Process management --
697--------------------------------------------------------------------------------
699-- | Thrown by 'terminate'
700data ProcessTerminationException = ProcessTerminationException
701 deriving (Show, Typeable)
703instance Exception ProcessTerminationException
705-- | Terminate immediately (throws a ProcessTerminationException)
706terminate :: Process a
707terminate = throwM ProcessTerminationException
710-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
711die :: Serializable a => a -> Process b
713-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
714-- could be decoded by a handler passed to 'catchExit', re-thrown or even
715-- passed to another process via the tracing mechanism.
717 throwM (ProcessExitException pid (createMessage reason))
719-- | Forceful request to kill a process. Where 'exit' provides an exception
720-- that can be caught and handled, 'kill' throws an unexposed exception type
721-- which cannot be handled explicitly (by type).
722kill :: ProcessId -> String -> Process ()
723-- NOTE: We send the message to our local node controller, which will then
724-- forward it to a remote node controller (if applicable). Sending it directly
725-- to a remote node controller means that that the message may overtake a
726-- 'monitor' or 'link' request.
727kill them reason = sendCtrlMsg Nothing (Kill them reason)
729-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
730-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
731-- manner can be handled using the 'catchExit' family of functions.
732exit :: Serializable a => ProcessId -> a -> Process ()
733-- NOTE: We send the message to our local node controller, which will then
734-- forward it to a remote node controller (if applicable). Sending it directly
735-- to a remote node controller means that that the message may overtake a
736-- 'monitor' or 'link' request.
737exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
739-- | Catches 'ProcessExitException'. The handler will not be applied unless its
740-- type matches the encoded data stored in the exception (see the /reason/
741-- argument given to the 'exit' primitive). If the handler cannot be applied,
742-- the exception will be re-thrown.
744-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
745-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
746catchExit :: forall a b . (Show a, Serializable a)
748 -> (ProcessId -> a -> Process b)
750catchExit act exitHandler = Catch.catch act handleExit
752 handleExit ex@(ProcessExitException from msg) =
753 if messageFingerprint msg == fingerprint (undefined :: a)
754 then exitHandler from decoded
758 -- Make sure the value is fully decoded so that we don't hang to
759 -- bytestrings if the caller doesn't use the value immediately
760 !decoded = decode (messageEncoding msg)
762-- | Lift 'Control.Exception.catches' (almost).
764-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
765-- message, a handler must accept inputs of the expected type. In order to
766-- handle a list of potentially different handlers (and therefore input types),
767-- a handler passed to 'catchesExit' must accept 'Message' and return
768-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
770-- See 'maybeHandleMessage' and 'Message' for more details.
771catchesExit :: Process b
772 -> [(ProcessId -> Message -> (Process (Maybe b)))]
774catchesExit act handlers = Catch.catch act ((flip handleExit) handlers)
776 handleExit :: ProcessExitException
777 -> [(ProcessId -> Message -> Process (Maybe b))]
779 handleExit ex [] = throwM ex
780 handleExit ex@(ProcessExitException from msg) (h:hs) = do
783 Nothing -> handleExit ex hs
786-- | Our own process ID
787getSelfPid :: Process ProcessId
788getSelfPid = processId <$> ask
790-- | Get the node ID of our local node
791getSelfNode :: Process NodeId
792getSelfNode = localNodeId . processNode <$> ask
795-- | Get statistics about the specified node
796getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
798 selfNode <- getSelfNode
800 then Right `fmap` getLocalNodeStats -- optimisation
801 else getNodeStatsRemote selfNode
803 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
804 getNodeStatsRemote selfNode = do
805 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
806 bracket (monitorNode nid) unmonitor $ \mRef ->
807 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
808 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
809 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
812-- | Get statistics about our local node
813getLocalNodeStats :: Process NodeStats
814getLocalNodeStats = do
816 sendCtrlMsg Nothing $ GetNodeStats self
817 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
819-- | Get information about the specified process
820getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
822 let them = processNodeId pid in do
824 dest <- mkNode them us
825 sendCtrlMsg dest $ GetInfo pid
827 match (\(p :: ProcessInfo) -> return $ Just p)
828 , match (\(_ :: ProcessInfoNone) -> return Nothing)
830 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
831 mkNode them us = case them == us of
832 True -> return Nothing
833 _ -> return $ Just them
835--------------------------------------------------------------------------------
836-- Monitoring and linking --
837--------------------------------------------------------------------------------
839-- | Link to a remote process (asynchronous)
841-- When process A links to process B (that is, process A calls
842-- @link pidB@) then an asynchronous exception will be thrown to process A
843-- when process B terminates (normally or abnormally), or when process A gets
844-- disconnected from process B. Although it is /technically/ possible to catch
845-- these exceptions, chances are if you find yourself trying to do so you should
846-- probably be using 'monitor' rather than 'link'. In particular, code such as
848-- > link pidB -- Link to process B
849-- > expect -- Wait for a message from process B
850-- > unlink pidB -- Unlink again
852-- doesn't quite do what one might expect: if process B sends a message to
853-- process A, and /subsequently terminates/, then process A might or might not
854-- be terminated too, depending on whether the exception is thrown before or
855-- after the 'unlink' (i.e., this code has a race condition).
857-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
858-- call to 'link' has no effect.
860-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
861-- Linking makes no distinction between normal and abnormal termination of
862-- the remote process.
863link :: ProcessId -> Process ()
864link = sendCtrlMsg Nothing . Link . ProcessIdentifier
866-- | Monitor another process (asynchronous)
868-- When process A monitors process B (that is, process A calls
869-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
870-- when process B terminates (normally or abnormally), or when process A gets
871-- disconnected from process B. You receive this message like any other (using
872-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
873-- 'DiedDisconnect', etc.).
875-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
876-- multiple monitors are set up, multiple notifications will be delivered
877-- and monitors can be disabled individually using 'unmonitor'.
878monitor :: ProcessId -> Process MonitorRef
879monitor = monitor' . ProcessIdentifier
881-- | Establishes temporary monitoring of another process.
883-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
884-- of @code@. Note: although monitoring is no longer active when
885-- @withMonitor@ returns, there might still be unreceived monitor
886-- messages in the queue.
888withMonitor :: ProcessId -> Process a -> Process a
889withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
890 -- unmonitor blocks waiting for the response, so there's a possibility
891 -- that an exception might interrupt withMonitor before the unmonitor
892 -- has completed. I think that's better than making the unmonitor
897-- This is synchronous in the sense that once it returns you are guaranteed
898-- that no exception will be raised if the remote process dies. However, it is
899-- asynchronous in the sense that we do not wait for a response from the remote
901unlink :: ProcessId -> Process ()
904 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
908-- | Remove a node link
910-- This has the same synchronous/asynchronous nature as 'unlink'.
911unlinkNode :: NodeId -> Process ()
914 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
918-- | Remove a channel (send port) link
920-- This has the same synchronous/asynchronous nature as 'unlink'.
921unlinkPort :: SendPort a -> Process ()
923 unlinkPortAsync sport
924 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
930-- This has the same synchronous/asynchronous nature as 'unlink'.
931unmonitor :: MonitorRef -> Process ()
934 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
938--------------------------------------------------------------------------------
939-- Exception handling --
940--------------------------------------------------------------------------------
942-- | Lift 'Control.Exception.catch'
943catch :: Exception e => Process a -> (e -> Process a) -> Process a
945{-# DEPRECATED catch "Use Control.Monad.Catch.catch instead" #-}
947-- | Lift 'Control.Exception.try'
948try :: Exception e => Process a -> Process (Either e a)
950{-# DEPRECATED try "Use Control.Monad.Catch.mask_ instead" #-}
952-- | Lift 'Control.Exception.mask'
953mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
955{-# DEPRECATED mask "Use Control.Monad.Catch.mask_ instead" #-}
957-- | Lift 'Control.Exception.mask_'
958mask_ :: Process a -> Process a
960{-# DEPRECATED mask_ "Use Control.Monad.Catch.mask_ instead" #-}
962-- | Lift 'Control.Exception.onException'
963onException :: Process a -> Process b -> Process a
964onException = Catch.onException
965{-# DEPRECATED onException "Use Control.Monad.Catch.onException instead" #-}
967-- | Lift 'Control.Exception.bracket'
968bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
969bracket = Catch.bracket
970{-# DEPRECATED bracket "Use Control.Monad.Catch.bracket instead" #-}
972-- | Lift 'Control.Exception.bracket_'
973bracket_ :: Process a -> Process b -> Process c -> Process c
974bracket_ = Catch.bracket_
975{-# DEPRECATED bracket_ "Use Control.Monad.Catch.bracket_ instead" #-}
977-- | Lift 'Control.Exception.finally'
978finally :: Process a -> Process b -> Process a
979finally = Catch.finally
980{-# DEPRECATED finally "Use Control.Monad.Catch.finally instead" #-}
982-- | You need this when using 'catches'
983data Handler a = forall e . Exception e => Handler (e -> Process a)
985instance Functor Handler where
986 fmap f (Handler h) = Handler (fmap f . h)
988-- | Lift 'Control.Exception.catches'
989catches :: Process a -> [Handler a] -> Process a
990catches proc handlers = proc `Catch.catch` catchesHandler handlers
992catchesHandler :: [Handler a] -> SomeException -> Process a
993catchesHandler handlers e = foldr tryHandler (throwM e) handlers
994 where tryHandler (Handler handler) res
995 = case fromException e of
996 Just e' -> handler e'
999--------------------------------------------------------------------------------
1001--------------------------------------------------------------------------------
1003-- | Like 'expect' but with a timeout
1004expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
1005expectTimeout n = receiveTimeout n [match return]
1007-- | Asynchronous version of 'spawn'
1009-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
1010spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
1011spawnAsync nid proc = do
1012 spawnRef <- getSpawnRef
1015 then sendCtrlMsg Nothing $ Spawn proc spawnRef
1016 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
1019-- | Monitor a node (asynchronous)
1020monitorNode :: NodeId -> Process MonitorRef
1022 monitor' . NodeIdentifier
1024-- | Monitor a typed channel (asynchronous)
1025monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1026monitorPort (SendPort cid) =
1027 monitor' (SendPortIdentifier cid)
1029-- | Remove a monitor (asynchronous)
1030unmonitorAsync :: MonitorRef -> Process ()
1032 sendCtrlMsg Nothing . Unmonitor
1034-- | Link to a node (asynchronous)
1035linkNode :: NodeId -> Process ()
1036linkNode = link' . NodeIdentifier
1038-- | Link to a channel (asynchronous)
1039linkPort :: SendPort a -> Process ()
1040linkPort (SendPort cid) =
1041 link' (SendPortIdentifier cid)
1043-- | Remove a process link (asynchronous)
1044unlinkAsync :: ProcessId -> Process ()
1046 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1048-- | Remove a node link (asynchronous)
1049unlinkNodeAsync :: NodeId -> Process ()
1051 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1053-- | Remove a channel (send port) link (asynchronous)
1054unlinkPortAsync :: SendPort a -> Process ()
1055unlinkPortAsync (SendPort cid) =
1056 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1058--------------------------------------------------------------------------------
1060--------------------------------------------------------------------------------
1064-- @say message@ sends a message (time, pid of the current process, message)
1065-- to the process registered as 'logger'. By default, this process simply
1066-- sends the string to 'stderr'. Individual Cloud Haskell backends might
1067-- replace this with a different logger process, however.
1068say :: String -> Process ()
1070 now <- liftIO getCurrentTime
1072 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
1074--------------------------------------------------------------------------------
1076--------------------------------------------------------------------------------
1078-- | Register a process with the local registry (synchronous). The name must not
1079-- already be registered. The process need not be on this node. A bad
1080-- registration will result in a 'ProcessRegistrationException'
1082-- The process to be registered does not have to be local itself.
1083register :: String -> ProcessId -> Process ()
1084register = registerImpl False
1086-- | Like 'register', but will replace an existing registration.
1087-- The name must already be registered.
1088reregister :: String -> ProcessId -> Process ()
1089reregister = registerImpl True
1091registerImpl :: Bool -> String -> ProcessId -> Process ()
1092registerImpl force label pid = do
1093 mynid <- getSelfNode
1094 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1096 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1097 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1100-- | Register a process with a remote registry (asynchronous).
1102-- The process to be registered does not have to live on the same remote node.
1103-- Reply wil come in the form of a 'RegisterReply' message
1105-- See comments in 'whereisRemoteAsync'
1106registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1107registerRemoteAsync nid label pid = do
1109 sendCtrlMsg (if nid == here then Nothing else Just nid)
1110 (Register label nid (Just pid) False)
1112reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1113reregisterRemoteAsync nid label pid =
1114 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1116-- | Remove a process from the local registry (asynchronous).
1117-- This version will wait until a response is gotten from the
1118-- management process. The name must already be registered.
1119unregister :: String -> Process ()
1120unregister label = do
1121 mynid <- getSelfNode
1122 sendCtrlMsg Nothing (Register label mynid Nothing False)
1124 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1125 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1128-- | Deal with the result from an attempted registration or unregistration
1129-- by throwing an exception if necessary
1130handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()
1131handleRegistrationReply label ok owner =
1133 throwM $ ProcessRegistrationException label owner
1135-- | Remove a process from a remote registry (asynchronous).
1137-- Reply wil come in the form of a 'RegisterReply' message
1139-- See comments in 'whereisRemoteAsync'
1140unregisterRemoteAsync :: NodeId -> String -> Process ()
1141unregisterRemoteAsync nid label =
1142 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1144-- | Query the local process registry
1145whereis :: String -> Process (Maybe ProcessId)
1147 sendCtrlMsg Nothing (WhereIs label)
1148 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1149 (\(WhereIsReply _ mPid) -> return mPid)
1152-- | Query a remote process registry (asynchronous)
1154-- Reply will come in the form of a 'WhereIsReply' message.
1156-- There is currently no synchronous version of 'whereisRemoteAsync': if
1157-- you implement one yourself, be sure to take into account that the remote
1158-- node might die or get disconnect before it can respond (i.e. you should
1159-- use 'monitorNode' and take appropriate action when you receive a
1160-- 'NodeMonitorNotification').
1161whereisRemoteAsync :: NodeId -> String -> Process ()
1162whereisRemoteAsync nid label = do
1164 sendCtrlMsg (if nid == here then Nothing else Just nid) (WhereIs label)
1166-- | Named send to a process in the local registry (asynchronous)
1167nsend :: Serializable a => String -> a -> Process ()
1169 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1171-- | Named send to a process in the local registry (asynchronous).
1172-- This function makes /no/ attempt to serialize and (in the case when the
1173-- destination process resides on the same local node) therefore ensure that
1174-- the payload is fully evaluated before it is delivered.
1175unsafeNSend :: Serializable a => String -> a -> Process ()
1176unsafeNSend = Unsafe.nsend
1178-- | Named send to a process in a remote registry (asynchronous)
1179nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1180nsendRemote nid label msg = do
1182 if here == nid then nsend label msg
1183 else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1185-- | Named send to a process in a remote registry (asynchronous)
1186-- This function makes /no/ attempt to serialize and (in the case when the
1187-- destination process resides on the same local node) therefore ensure that
1188-- the payload is fully evaluated before it is delivered.
1189unsafeNSendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1190unsafeNSendRemote = Unsafe.nsendRemote
1192--------------------------------------------------------------------------------
1194--------------------------------------------------------------------------------
1196-- | Resolve a static value
1197unStatic :: Typeable a => Static a -> Process a
1199 rtable <- remoteTable . processNode <$> ask
1200 case Static.unstatic rtable static of
1201 Left err -> fail $ "Could not resolve static value: " ++ err
1204-- | Resolve a closure
1205unClosure :: Typeable a => Closure a -> Process a
1206unClosure closure = do
1207 rtable <- remoteTable . processNode <$> ask
1208 case Static.unclosure rtable closure of
1209 Left err -> fail $ "Could not resolve closure: " ++ err
1212--------------------------------------------------------------------------------
1214--------------------------------------------------------------------------------
1216-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1217-- message passing. However, when network connections get disrupted this
1218-- illusion cannot always be maintained. Once a network connection breaks (even
1219-- temporarily) no further communication on that connection will be possible.
1220-- For example, if process A sends a message to process B, and A is then
1221-- notified (by monitor notification) that it got disconnected from B, A will
1222-- not be able to send any further messages to B, /unless/ A explicitly
1223-- indicates that it is acceptable to attempt to reconnect to B using the
1224-- Cloud Haskell 'reconnect' primitive.
1226-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1227-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1228-- then receives a monitor notification that its connection to B has been lost,
1229-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1230-- and m3 but not m2.
1232-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1233-- to attempt to reconnect on the next send/. In particular, if no further
1234-- communication attempts are made to B then A can use reconnect to clean up
1235-- its connection to B.
1236reconnect :: ProcessId -> Process ()
1239 node <- processNode <$> ask
1240 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1242-- | Reconnect to a sendport. See 'reconnect' for more information.
1243reconnectPort :: SendPort a -> Process ()
1244reconnectPort them = do
1246 node <- processNode <$> ask
1247 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1249--------------------------------------------------------------------------------
1250-- Auxiliary functions --
1251--------------------------------------------------------------------------------
1253sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1255 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1257sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1258sendChanLocal spId msg =
1259 -- we *must* fully serialize/encode the message here, because
1260 -- attempting to use `unsafeCoerce' in the node controller
1261 -- won't work since we know nothing about the required type
1262 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1264getMonitorRefFor :: Identifier -> Process MonitorRef
1265getMonitorRefFor ident = do
1267 liftIO $ modifyMVar (processState proc) $ \st -> do
1268 let counter = st ^. monitorCounter
1269 return ( monitorCounter ^: (+ 1) $ st
1270 , MonitorRef ident counter
1273getSpawnRef :: Process SpawnRef
1276 liftIO $ modifyMVar (processState proc) $ \st -> do
1277 let counter = st ^. spawnCounter
1278 return ( spawnCounter ^: (+ 1) $ st
1282-- | Monitor a process/node/channel
1283monitor' :: Identifier -> Process MonitorRef
1285 monitorRef <- getMonitorRefFor ident
1286 sendCtrlMsg Nothing $ Monitor monitorRef
1289-- | Link to a process/node/channel
1290link' :: Identifier -> Process ()
1291link' = 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