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
29 -- * Advanced messaging
55 -- * Process management
57 , ProcessTerminationException(..)
63 -- keep the exception constructor hidden, so that handling exit
64 -- reasons /must/ take place via the 'catchExit' family of primitives
65 , ProcessExitException()
73 -- * Monitoring and linking
88 , reregisterRemoteAsync
89 , unregisterRemoteAsync
95 -- * Exception handling
119 -- * Internal Exports
123#if ! MIN_VERSION_base(4,6,0)
124import Prelude hiding (catch)
127import Data.Binary (decode)
128import Data.Time.Clock (getCurrentTime)
129import Data.Time.Format (formatTime)
130#if MIN_VERSION_time(1,5,0)
131import Data.Time.Format (defaultTimeLocale)
133import System.Locale (defaultTimeLocale)
135import System.Timeout (timeout)
136import Control.Monad (when)
137import Control.Monad.Reader (ask)
138import Control.Monad.IO.Class (liftIO)
139import Control.Applicative ((<$>))
140import Control.Exception (Exception(..), throw, throwIO, SomeException)
141import qualified Control.Exception as Ex (catch, mask, mask_, try)
142import Control.Distributed.Process.Internal.StrictMVar
147import Control.Concurrent.STM
156import Control.Distributed.Process.Internal.CQueue
161import Control.Distributed.Process.Serializable (Serializable, fingerprint)
162import Data.Accessor ((^.), (^:), (^=))
163import Control.Distributed.Static
167import Data.Rank1Typeable (Typeable)
168import qualified Control.Distributed.Static as Static (unstatic, unclosure)
169import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
170import Control.Distributed.Process.Internal.Types
180 , NodeMonitorNotification(..)
190 , ProcessExitException(..)
193 , DidUnlinkProcess(..)
198 , ProcessRegistrationException(..)
200 , ProcessInfoNone(..)
204 , createUnencodedMessage
206 , ImplicitReconnect( NoImplicitReconnect)
211import Control.Distributed.Process.Internal.Messaging
218import Control.Distributed.Process.Management.Internal.Types
221import Control.Distributed.Process.Management.Internal.Trace.Types
224import Control.Distributed.Process.Internal.WeakTQueue
232--------------------------------------------------------------------------------
234--------------------------------------------------------------------------------
237send :: Serializable a => ProcessId -> a -> Process ()
238-- This requires a lookup on every send. If we want to avoid that we need to
239-- modify serializable to allow for stateful (IO) deserialization.
242 let us = processId proc
243 node = processNode proc
244 nodeId = localNodeId node
245 destNode = (processNodeId them) in do
246 case destNode == nodeId of
247 True -> sendLocal them msg
248 False -> liftIO $ sendMessage (processNode proc)
249 (ProcessIdentifier (processId proc))
250 (ProcessIdentifier them)
253 -- We do not fire the trace event until after the sending is complete;
254 -- In the remote case, 'sendMessage' can block in the networking stack.
255 liftIO $ traceEvent (localEventBus node)
256 (MxSent them us (createUnencodedMessage msg))
258-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
259-- and (in the case when the destination process resides on the same local
260-- node) therefore ensure that the payload is fully evaluated before it is
262unsafeSend :: Serializable a => ProcessId -> a -> Process ()
263unsafeSend = Unsafe.send
265-- | Send a message unreliably.
267-- Unlike 'send', this function is insensitive to 'reconnect'. It will
268-- try to send the message regardless of the history of connection failures
271-- Message passing with 'usend' is ordered for a given sender and receiver
272-- if the messages arrive at all.
274usend :: Serializable a => ProcessId -> a -> Process ()
277 let there = processNodeId them
279 then sendLocal them msg
280 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
283-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize
284-- the message when the destination process resides on the same local
285-- node. Therefore, a local receiver would need to be prepared to cope with any
286-- errors resulting from evaluation of the message.
287unsafeUSend :: Serializable a => ProcessId -> a -> Process ()
288unsafeUSend = Unsafe.usend
290-- | Wait for a message of a specific type
291expect :: forall a. Serializable a => Process a
292expect = receiveWait [match return]
294--------------------------------------------------------------------------------
296--------------------------------------------------------------------------------
298-- | Create a new typed channel
299newChan :: Serializable a => Process (SendPort a, ReceivePort a)
302 liftIO . modifyMVar (processState proc) $ \st -> do
303 let lcid = st ^. channelCounter
304 let cid = SendPortId { sendPortProcessId = processId proc
305 , sendPortLocalId = lcid
307 let sport = SendPort cid
308 chan <- liftIO newTQueueIO
309 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
310 let rport = ReceivePort $ readTQueue chan
311 let tch = TypedChannel chan'
312 return ( (channelCounter ^: (+ 1))
313 . (typedChannelWithId lcid ^= Just tch)
318 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
319 finalizer st lcid = modifyMVar_ st $
320 return . (typedChannelWithId lcid ^= Nothing)
322-- | Send a message on a typed channel
323sendChan :: Serializable a => SendPort a -> a -> Process ()
324sendChan (SendPort cid) msg = do
326 let node = localNodeId (processNode proc)
327 destNode = processNodeId (sendPortProcessId cid) in do
328 case destNode == node of
329 True -> sendChanLocal cid msg
331 liftIO $ sendBinary (processNode proc)
332 (ProcessIdentifier (processId proc))
333 (SendPortIdentifier cid)
337-- | Send a message on a typed channel. This function makes /no/ attempt to
338-- serialize and (in the case when the @ReceivePort@ resides on the same local
339-- node) therefore ensure that the payload is fully evaluated before it is
341unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
342unsafeSendChan = Unsafe.sendChan
344-- | Wait for a message on a typed channel
345receiveChan :: Serializable a => ReceivePort a -> Process a
346receiveChan = liftIO . atomically . receiveSTM
348-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
349-- non-blocking check for a message.
350receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
351receiveChanTimeout 0 ch = liftIO . atomically $
352 (Just <$> receiveSTM ch) `orElse` return Nothing
353receiveChanTimeout n ch = liftIO . timeout n . atomically $
356-- | Merge a list of typed channels.
358-- The result port is left-biased: if there are messages available on more
359-- than one port, the first available message is returned.
360mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
361mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
363-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
365mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
366mergePortsRR = \ps -> do
367 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
368 return $ ReceivePort (rr psVar)
372 rotate (x:xs) = xs ++ [x]
374 rr :: TVar [STM a] -> STM a
377 a <- foldr1 orElse ps
378 writeTVar psVar (rotate ps)
381--------------------------------------------------------------------------------
382-- Advanced messaging --
383--------------------------------------------------------------------------------
385-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
386newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
389-- | Test the matches in order against each message in the queue
390receiveWait :: [Match b] -> Process b
392 queue <- processQueue <$> ask
393 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
396-- | Like 'receiveWait' but with a timeout.
398-- If the timeout is zero do a non-blocking check for matching messages. A
399-- non-zero timeout is applied only when waiting for incoming messages (that is,
400-- /after/ we have checked the messages that are already in the mailbox).
401receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
402receiveTimeout t ms = do
403 queue <- processQueue <$> ask
404 let blockSpec = if t == 0 then NonBlocking else Timeout t
405 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
407 Nothing -> return Nothing
408 Just proc -> Just <$> proc
410-- | Match on a typed channel
411matchChan :: ReceivePort a -> (a -> Process b) -> Match b
412matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
414-- | Match on an arbitrary STM action.
416-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
417-- actions with checks against our process' mailbox and/or any typed channel
418-- @ReceivePort@s we may hold.
420-- This allows us to process multiple input streams /along with our mailbox/,
421-- in just the same way that 'matchChan' supports checking both the mailbox
422-- /and/ an arbitrary set of typed channels in one atomic transaction.
424-- Note there are no ordering guarnatees with respect to these disparate input
427matchSTM :: STM a -> (a -> Process b) -> Match b
428matchSTM stm fn = Match $ MatchChan (fmap fn stm)
430-- | Match against any message of the right type
431match :: forall a b. Serializable a => (a -> Process b) -> Match b
432match = matchIf (const True)
434-- | Match against any message of the right type that satisfies a predicate
435matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
436matchIf c p = Match $ MatchMsg $ \msg ->
437 case messageFingerprint msg == fingerprint (undefined :: a) of
440 (UnencodedMessage _ m) ->
441 let m' = unsafeCoerce m :: a in
445 (EncodedMessage _ _) ->
446 if (c decoded) then Just (p decoded) else Nothing
449 -- Make sure the value is fully decoded so that we don't hang to
450 -- bytestrings when the process calling 'matchIf' doesn't process
451 -- the values immediately
452 !decoded = decode (messageEncoding msg)
454-- | Match against any message, regardless of the underlying (contained) type
455matchMessage :: (Message -> Process Message) -> Match Message
456matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
458-- | Match against any message (regardless of underlying type) that satisfies a predicate
459matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
460matchMessageIf c p = Match $ MatchMsg $ \msg ->
465-- | Forward a raw 'Message' to the given 'ProcessId'.
466forward :: Message -> ProcessId -> Process ()
469 let node = processNode proc
471 nid = localNodeId node
472 destNode = (processNodeId them) in do
473 case destNode == nid of
474 True -> sendCtrlMsg Nothing (LocalSend them msg)
475 False -> liftIO $ sendPayload (processNode proc)
476 (ProcessIdentifier (processId proc))
477 (ProcessIdentifier them)
479 (messageToPayload msg)
480 -- We do not fire the trace event until after the sending is complete;
481 -- In the remote case, 'sendMessage' can block in the networking stack.
482 liftIO $ traceEvent (localEventBus node)
485-- | Forward a raw 'Message' to the given 'ProcessId'.
487-- Unlike 'forward', this function is insensitive to 'reconnect'. It will
488-- try to send the message regardless of the history of connection failures
490uforward :: Message -> ProcessId -> Process ()
491uforward msg them = do
493 let node = processNode proc
495 nid = localNodeId node
496 destNode = (processNodeId them) in do
497 case destNode == nid of
498 True -> sendCtrlMsg Nothing (LocalSend them msg)
499 False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)
501 -- We do not fire the trace event until after the sending is complete;
502 -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
503 liftIO $ traceEvent (localEventBus node)
506-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
507-- 'Serializable' - like the datum they contain - but also note, deserialising
508-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
509-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
514-- send self (wrapMessage "blah")
515-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
516-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
517-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
520wrapMessage :: Serializable a => a -> Message
521wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
523-- | This is the /unsafe/ variant of 'wrapMessage'. See
524-- "Control.Distributed.Process.UnsafePrimitives" for details.
525unsafeWrapMessage :: Serializable a => a -> Message
526unsafeWrapMessage = Unsafe.wrapMessage
528-- [note Serializable UnencodedMessage]
529-- if we attempt to serialise an UnencodedMessage, it will be converted to an
530-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
531-- Primitives.hs for the gory details
533-- | Attempt to unwrap a raw 'Message'.
534-- If the type of the decoded message payload matches the expected type, the
535-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
538-- This expression, for example, will evaluate to @Nothing@
539-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
541-- Whereas this expression, will yield @Just "foo"@
542-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
544unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
546 case messageFingerprint msg == fingerprint (undefined :: a) of
547 False -> return Nothing :: m (Maybe a)
549 (UnencodedMessage _ ms) ->
550 let ms' = unsafeCoerce ms :: a
552 (EncodedMessage _ _) ->
553 return (Just (decoded))
555 decoded :: a -- note [decoding]
556 !decoded = decode (messageEncoding msg)
558-- | Attempt to handle a raw 'Message'.
559-- If the type of the message matches the type of the first argument to
560-- the supplied expression, then the message will be decoded and the expression
561-- evaluated against its value. If this runtime type checking fails however,
562-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
563-- evaluation proceeds, the resulting value with be wrapped with @Just@.
565-- Intended for use in `catchesExit` and `matchAny` primitives.
567handleMessage :: forall m a b. (Monad m, Serializable a)
568 => Message -> (a -> m b) -> m (Maybe b)
569handleMessage msg proc = handleMessageIf msg (const True) proc
571-- | Conditionally handle a raw 'Message'.
572-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
573-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
574-- for further information about runtime type checking.
576handleMessageIf :: forall m a b . (Monad m, Serializable a)
581handleMessageIf msg c proc = do
582 case messageFingerprint msg == fingerprint (undefined :: a) of
583 False -> return Nothing :: m (Maybe b)
585 (UnencodedMessage _ ms) ->
586 let ms' = unsafeCoerce ms :: a in
588 True -> do { r <- proc ms'; return (Just r) }
589 False -> return Nothing :: m (Maybe b)
590 (EncodedMessage _ _) ->
592 True -> do { r <- proc (decoded :: a); return (Just r) }
593 False -> return Nothing :: m (Maybe b)
595 decoded :: a -- note [decoding]
596 !decoded = decode (messageEncoding msg)
598-- | As 'handleMessage' but ignores result, which is useful if you don't
599-- care whether or not the handler succeeded.
601handleMessage_ :: forall m a . (Monad m, Serializable a)
602 => Message -> (a -> m ()) -> m ()
603handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
605-- | Conditional version of 'handleMessage_'.
607handleMessageIf_ :: forall m a . (Monad m, Serializable a)
612handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
614-- | Match against an arbitrary message. 'matchAny' removes the first available
615-- message from the process mailbox. To handle arbitrary /raw/ messages once
616-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
618matchAny :: forall b. (Message -> Process b) -> Match b
619matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
621-- | Match against an arbitrary message. Intended for use with 'handleMessage'
622-- and 'unwrapMessage', this function /only/ removes a message from the process
623-- mailbox, /if/ the supplied condition matches. The success (or failure) of
624-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
625-- here, i.e., if the condition evaluates to @True@ then the message will
626-- be removed from the process mailbox and decoded, but that does /not/
627-- guarantee that an expression passed to @handleMessage@ will pass the
628-- runtime type checks and therefore be evaluated.
630matchAnyIf :: forall a b. (Serializable a)
632 -> (Message -> Process b)
634matchAnyIf c p = Match $ MatchMsg $ \msg ->
635 case messageFingerprint msg == fingerprint (undefined :: a) of
636 True | check -> Just (p msg)
641 (EncodedMessage _ _) -> c decoded
642 (UnencodedMessage _ m') -> c (unsafeCoerce m')
644 decoded :: a -- note [decoding]
645 !decoded = decode (messageEncoding msg)
649For an EncodedMessage, we need to ensure the value is fully decoded so that
650we don't hang to bytestrings if the calling process doesn't evaluate
651immediately. For UnencodedMessage we know (because the fingerprint comparison
652succeeds) that unsafeCoerce will not fail.
655-- | Remove any message from the queue
656matchUnknown :: Process b -> Match b
657matchUnknown p = Match $ MatchMsg (const (Just p))
659-- | Receives messages and forwards them to @pid@ if @p msg == True@.
660delegate :: ProcessId -> (Message -> Bool) -> Process ()
663 matchAny (\m -> case (p m) of
664 True -> forward m pid
669-- | A straight relay that forwards all messages to the supplied pid.
670relay :: ProcessId -> Process ()
671relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
673-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
674-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
675-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
676-- fails, no action is taken and the proxy process will continue running.
677proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
681 next <- handleMessage m proc
683 Just True -> forward m pid
684 Just False -> return () -- explicitly ignored
685 Nothing -> return ()) -- un-routable
689--------------------------------------------------------------------------------
690-- Process management --
691--------------------------------------------------------------------------------
693-- | Thrown by 'terminate'
694data ProcessTerminationException = ProcessTerminationException
695 deriving (Show, Typeable)
697instance Exception ProcessTerminationException
699-- | Terminate immediately (throws a ProcessTerminationException)
700terminate :: Process a
701terminate = liftIO $ throwIO ProcessTerminationException
704-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
705die :: Serializable a => a -> Process b
707-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
708-- could be decoded by a handler passed to 'catchExit', re-thrown or even
709-- passed to another process via the tracing mechanism.
711 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
713-- | Forceful request to kill a process. Where 'exit' provides an exception
714-- that can be caught and handled, 'kill' throws an unexposed exception type
715-- which cannot be handled explicitly (by type).
716kill :: ProcessId -> String -> Process ()
717-- NOTE: We send the message to our local node controller, which will then
718-- forward it to a remote node controller (if applicable). Sending it directly
719-- to a remote node controller means that that the message may overtake a
720-- 'monitor' or 'link' request.
721kill them reason = sendCtrlMsg Nothing (Kill them reason)
723-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
724-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
725-- manner can be handled using the 'catchExit' family of functions.
726exit :: Serializable a => ProcessId -> a -> Process ()
727-- NOTE: We send the message to our local node controller, which will then
728-- forward it to a remote node controller (if applicable). Sending it directly
729-- to a remote node controller means that that the message may overtake a
730-- 'monitor' or 'link' request.
731exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
733-- | Catches 'ProcessExitException'. The handler will not be applied unless its
734-- type matches the encoded data stored in the exception (see the /reason/
735-- argument given to the 'exit' primitive). If the handler cannot be applied,
736-- the exception will be re-thrown.
738-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
739-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
740catchExit :: forall a b . (Show a, Serializable a)
742 -> (ProcessId -> a -> Process b)
744catchExit act exitHandler = catch act handleExit
746 handleExit ex@(ProcessExitException from msg) =
747 if messageFingerprint msg == fingerprint (undefined :: a)
748 then exitHandler from decoded
749 else liftIO $ throwIO ex
752 -- Make sure the value is fully decoded so that we don't hang to
753 -- bytestrings if the caller doesn't use the value immediately
754 !decoded = decode (messageEncoding msg)
756-- | Lift 'Control.Exception.catches' (almost).
758-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
759-- message, a handler must accept inputs of the expected type. In order to
760-- handle a list of potentially different handlers (and therefore input types),
761-- a handler passed to 'catchesExit' must accept 'Message' and return
762-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
764-- See 'maybeHandleMessage' and 'Message' for more details.
765catchesExit :: Process b
766 -> [(ProcessId -> Message -> (Process (Maybe b)))]
768catchesExit act handlers = catch act ((flip handleExit) handlers)
770 handleExit :: ProcessExitException
771 -> [(ProcessId -> Message -> Process (Maybe b))]
773 handleExit ex [] = liftIO $ throwIO ex
774 handleExit ex@(ProcessExitException from msg) (h:hs) = do
777 Nothing -> handleExit ex hs
780-- | Our own process ID
781getSelfPid :: Process ProcessId
782getSelfPid = processId <$> ask
784-- | Get the node ID of our local node
785getSelfNode :: Process NodeId
786getSelfNode = localNodeId . processNode <$> ask
789-- | Get statistics about the specified node
790getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
792 selfNode <- getSelfNode
794 then Right `fmap` getLocalNodeStats -- optimisation
795 else getNodeStatsRemote selfNode
797 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
798 getNodeStatsRemote selfNode = do
799 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
800 bracket (monitorNode nid) unmonitor $ \mRef ->
801 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
802 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
803 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
806-- | Get statistics about our local node
807getLocalNodeStats :: Process NodeStats
808getLocalNodeStats = do
810 sendCtrlMsg Nothing $ GetNodeStats self
811 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
813-- | Get information about the specified process
814getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
816 let them = processNodeId pid in do
818 dest <- mkNode them us
819 sendCtrlMsg dest $ GetInfo pid
821 match (\(p :: ProcessInfo) -> return $ Just p)
822 , match (\(_ :: ProcessInfoNone) -> return Nothing)
824 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
825 mkNode them us = case them == us of
826 True -> return Nothing
827 _ -> return $ Just them
829--------------------------------------------------------------------------------
830-- Monitoring and linking --
831--------------------------------------------------------------------------------
833-- | Link to a remote process (asynchronous)
835-- When process A links to process B (that is, process A calls
836-- @link pidB@) then an asynchronous exception will be thrown to process A
837-- when process B terminates (normally or abnormally), or when process A gets
838-- disconnected from process B. Although it is /technically/ possible to catch
839-- these exceptions, chances are if you find yourself trying to do so you should
840-- probably be using 'monitor' rather than 'link'. In particular, code such as
842-- > link pidB -- Link to process B
843-- > expect -- Wait for a message from process B
844-- > unlink pidB -- Unlink again
846-- doesn't quite do what one might expect: if process B sends a message to
847-- process A, and /subsequently terminates/, then process A might or might not
848-- be terminated too, depending on whether the exception is thrown before or
849-- after the 'unlink' (i.e., this code has a race condition).
851-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
852-- call to 'link' has no effect.
854-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
855-- Linking makes no distinction between normal and abnormal termination of
856-- the remote process.
857link :: ProcessId -> Process ()
858link = sendCtrlMsg Nothing . Link . ProcessIdentifier
860-- | Monitor another process (asynchronous)
862-- When process A monitors process B (that is, process A calls
863-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
864-- when process B terminates (normally or abnormally), or when process A gets
865-- disconnected from process B. You receive this message like any other (using
866-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
867-- 'DiedDisconnect', etc.).
869-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
870-- multiple monitors are set up, multiple notifications will be delivered
871-- and monitors can be disabled individually using 'unmonitor'.
872monitor :: ProcessId -> Process MonitorRef
873monitor = monitor' . ProcessIdentifier
875-- | Establishes temporary monitoring of another process.
877-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
878-- of @code@. Note: although monitoring is no longer active when
879-- @withMonitor@ returns, there might still be unreceived monitor
880-- messages in the queue.
882withMonitor :: ProcessId -> Process a -> Process a
883withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
884 -- unmonitor blocks waiting for the response, so there's a possibility
885 -- that an exception might interrupt withMonitor before the unmonitor
886 -- has completed. I think that's better than making the unmonitor
891-- This is synchronous in the sense that once it returns you are guaranteed
892-- that no exception will be raised if the remote process dies. However, it is
893-- asynchronous in the sense that we do not wait for a response from the remote
895unlink :: ProcessId -> Process ()
898 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
902-- | Remove a node link
904-- This has the same synchronous/asynchronous nature as 'unlink'.
905unlinkNode :: NodeId -> Process ()
908 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
912-- | Remove a channel (send port) link
914-- This has the same synchronous/asynchronous nature as 'unlink'.
915unlinkPort :: SendPort a -> Process ()
917 unlinkPortAsync sport
918 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
924-- This has the same synchronous/asynchronous nature as 'unlink'.
925unmonitor :: MonitorRef -> Process ()
928 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
932--------------------------------------------------------------------------------
933-- Exception handling --
934--------------------------------------------------------------------------------
936-- | Lift 'Control.Exception.catch'
937catch :: Exception e => Process a -> (e -> Process a) -> Process a
940 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
942-- | Lift 'Control.Exception.try'
943try :: Exception e => Process a -> Process (Either e a)
946 liftIO $ Ex.try (runLocalProcess lproc p)
948-- | Lift 'Control.Exception.mask'
949mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
952 liftIO $ Ex.mask $ \restore ->
953 runLocalProcess lproc (p (liftRestore restore))
955 liftRestore :: (forall a. IO a -> IO a)
956 -> (forall a. Process a -> Process a)
957 liftRestore restoreIO = \p2 -> do
959 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
961-- | Lift 'Control.Exception.mask_'
962mask_ :: Process a -> Process a
965 liftIO $ Ex.mask_ $ runLocalProcess lproc p
967-- | Lift 'Control.Exception.onException'
968onException :: Process a -> Process b -> Process a
969onException p what = p `catch` \e -> do _ <- what
970 liftIO $ throwIO (e :: SomeException)
972-- | Lift 'Control.Exception.bracket'
973bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
974bracket before after thing =
975 mask $ \restore -> do
977 r <- restore (thing a) `onException` after a
981-- | Lift 'Control.Exception.bracket_'
982bracket_ :: Process a -> Process b -> Process c -> Process c
983bracket_ before after thing = bracket before (const after) (const thing)
985-- | Lift 'Control.Exception.finally'
986finally :: Process a -> Process b -> Process a
987finally a sequel = bracket_ (return ()) sequel a
989-- | You need this when using 'catches'
990data Handler a = forall e . Exception e => Handler (e -> Process a)
992instance Functor Handler where
993 fmap f (Handler h) = Handler (fmap f . h)
995-- | Lift 'Control.Exception.catches'
996catches :: Process a -> [Handler a] -> Process a
997catches proc handlers = proc `catch` catchesHandler handlers
999catchesHandler :: [Handler a] -> SomeException -> Process a
1000catchesHandler handlers e = foldr tryHandler (throw e) handlers
1001 where tryHandler (Handler handler) res
1002 = case fromException e of
1003 Just e' -> handler e'
1006--------------------------------------------------------------------------------
1008--------------------------------------------------------------------------------
1010-- | Like 'expect' but with a timeout
1011expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
1012expectTimeout n = receiveTimeout n [match return]
1014-- | Asynchronous version of 'spawn'
1016-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
1017spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
1018spawnAsync nid proc = do
1019 spawnRef <- getSpawnRef
1022 then sendCtrlMsg Nothing $ Spawn proc spawnRef
1023 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
1026-- | Monitor a node (asynchronous)
1027monitorNode :: NodeId -> Process MonitorRef
1029 monitor' . NodeIdentifier
1031-- | Monitor a typed channel (asynchronous)
1032monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1033monitorPort (SendPort cid) =
1034 monitor' (SendPortIdentifier cid)
1036-- | Remove a monitor (asynchronous)
1037unmonitorAsync :: MonitorRef -> Process ()
1039 sendCtrlMsg Nothing . Unmonitor
1041-- | Link to a node (asynchronous)
1042linkNode :: NodeId -> Process ()
1043linkNode = link' . NodeIdentifier
1045-- | Link to a channel (asynchronous)
1046linkPort :: SendPort a -> Process ()
1047linkPort (SendPort cid) =
1048 link' (SendPortIdentifier cid)
1050-- | Remove a process link (asynchronous)
1051unlinkAsync :: ProcessId -> Process ()
1053 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1055-- | Remove a node link (asynchronous)
1056unlinkNodeAsync :: NodeId -> Process ()
1058 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1060-- | Remove a channel (send port) link (asynchronous)
1061unlinkPortAsync :: SendPort a -> Process ()
1062unlinkPortAsync (SendPort cid) =
1063 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1065--------------------------------------------------------------------------------
1067--------------------------------------------------------------------------------
1071-- @say message@ sends a message (time, pid of the current process, message)
1072-- to the process registered as 'logger'. By default, this process simply
1073-- sends the string to 'stderr'. Individual Cloud Haskell backends might
1074-- replace this with a different logger process, however.
1075say :: String -> Process ()
1077 now <- liftIO getCurrentTime
1079 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
1081--------------------------------------------------------------------------------
1083--------------------------------------------------------------------------------
1085-- | Register a process with the local registry (synchronous). The name must not
1086-- already be registered. The process need not be on this node. A bad
1087-- registration will result in a 'ProcessRegistrationException'
1089-- The process to be registered does not have to be local itself.
1090register :: String -> ProcessId -> Process ()
1091register = registerImpl False
1093-- | Like 'register', but will replace an existing registration.
1094-- The name must already be registered.
1095reregister :: String -> ProcessId -> Process ()
1096reregister = registerImpl True
1098registerImpl :: Bool -> String -> ProcessId -> Process ()
1099registerImpl force label pid = do
1100 mynid <- getSelfNode
1101 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1103 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1104 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1107-- | Register a process with a remote registry (asynchronous).
1109-- The process to be registered does not have to live on the same remote node.
1110-- Reply wil come in the form of a 'RegisterReply' message
1112-- See comments in 'whereisRemoteAsync'
1113registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1114registerRemoteAsync nid label pid =
1115 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
1117reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1118reregisterRemoteAsync nid label pid =
1119 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1121-- | Remove a process from the local registry (asynchronous).
1122-- This version will wait until a response is gotten from the
1123-- management process. The name must already be registered.
1124unregister :: String -> Process ()
1125unregister label = do
1126 mynid <- getSelfNode
1127 sendCtrlMsg Nothing (Register label mynid Nothing False)
1129 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1130 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1133-- | Deal with the result from an attempted registration or unregistration
1134-- by throwing an exception if necessary
1135handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()
1136handleRegistrationReply label ok owner =
1138 liftIO $ throwIO $ ProcessRegistrationException label owner
1140-- | Remove a process from a remote registry (asynchronous).
1142-- Reply wil come in the form of a 'RegisterReply' message
1144-- See comments in 'whereisRemoteAsync'
1145unregisterRemoteAsync :: NodeId -> String -> Process ()
1146unregisterRemoteAsync nid label =
1147 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1149-- | Query the local process registry
1150whereis :: String -> Process (Maybe ProcessId)
1152 sendCtrlMsg Nothing (WhereIs label)
1153 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1154 (\(WhereIsReply _ mPid) -> return mPid)
1157-- | Query a remote process registry (asynchronous)
1159-- Reply will come in the form of a 'WhereIsReply' message.
1161-- There is currently no synchronous version of 'whereisRemoteAsync': if
1162-- you implement one yourself, be sure to take into account that the remote
1163-- node might die or get disconnect before it can respond (i.e. you should
1164-- use 'monitorNode' and take appropriate action when you receive a
1165-- 'NodeMonitorNotification').
1166whereisRemoteAsync :: NodeId -> String -> Process ()
1167whereisRemoteAsync nid label =
1168 sendCtrlMsg (Just nid) (WhereIs label)
1170-- | Named send to a process in the local registry (asynchronous)
1171nsend :: Serializable a => String -> a -> Process ()
1173 sendCtrlMsg Nothing (NamedSend label (createMessage msg))
1175-- | Named send to a process in the local registry (asynchronous).
1176-- This function makes /no/ attempt to serialize and (in the case when the
1177-- destination process resides on the same local node) therefore ensure that
1178-- the payload is fully evaluated before it is delivered.
1179unsafeNSend :: Serializable a => String -> a -> Process ()
1180unsafeNSend = Unsafe.nsend
1182-- | Named send to a process in a remote registry (asynchronous)
1183nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1184nsendRemote nid label msg =
1185 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1187--------------------------------------------------------------------------------
1189--------------------------------------------------------------------------------
1191-- | Resolve a static value
1192unStatic :: Typeable a => Static a -> Process a
1194 rtable <- remoteTable . processNode <$> ask
1195 case Static.unstatic rtable static of
1196 Left err -> fail $ "Could not resolve static value: " ++ err
1199-- | Resolve a closure
1200unClosure :: Typeable a => Closure a -> Process a
1201unClosure closure = do
1202 rtable <- remoteTable . processNode <$> ask
1203 case Static.unclosure rtable closure of
1204 Left err -> fail $ "Could not resolve closure: " ++ err
1207--------------------------------------------------------------------------------
1209--------------------------------------------------------------------------------
1211-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1212-- message passing. However, when network connections get disrupted this
1213-- illusion cannot always be maintained. Once a network connection breaks (even
1214-- temporarily) no further communication on that connection will be possible.
1215-- For example, if process A sends a message to process B, and A is then
1216-- notified (by monitor notification) that it got disconnected from B, A will
1217-- not be able to send any further messages to B, /unless/ A explicitly
1218-- indicates that it is acceptable to attempt to reconnect to B using the
1219-- Cloud Haskell 'reconnect' primitive.
1221-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1222-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1223-- then receives a monitor notification that its connection to B has been lost,
1224-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1225-- and m3 but not m2.
1227-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1228-- to attempt to reconnect on the next send/. In particular, if no further
1229-- communication attempts are made to B then A can use reconnect to clean up
1230-- its connection to B.
1231reconnect :: ProcessId -> Process ()
1234 node <- processNode <$> ask
1235 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1237-- | Reconnect to a sendport. See 'reconnect' for more information.
1238reconnectPort :: SendPort a -> Process ()
1239reconnectPort them = do
1241 node <- processNode <$> ask
1242 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1244--------------------------------------------------------------------------------
1245-- Auxiliary functions --
1246--------------------------------------------------------------------------------
1248sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1250 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1252sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1253sendChanLocal spId msg =
1254 -- we *must* fully serialize/encode the message here, because
1255 -- attempting to use `unsafeCoerce' in the node controller
1256 -- won't work since we know nothing about the required type
1257 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1259getMonitorRefFor :: Identifier -> Process MonitorRef
1260getMonitorRefFor ident = do
1262 liftIO $ modifyMVar (processState proc) $ \st -> do
1263 let counter = st ^. monitorCounter
1264 return ( monitorCounter ^: (+ 1) $ st
1265 , MonitorRef ident counter
1268getSpawnRef :: Process SpawnRef
1271 liftIO $ modifyMVar (processState proc) $ \st -> do
1272 let counter = st ^. spawnCounter
1273 return ( spawnCounter ^: (+ 1) $ st
1277-- | Monitor a process/node/channel
1278monitor' :: Identifier -> Process MonitorRef
1280 monitorRef <- getMonitorRefFor ident
1281 sendCtrlMsg Nothing $ Monitor monitorRef
1284-- | Link to a process/node/channel
1285link' :: Identifier -> Process ()
1286link' = 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