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
28 -- * Advanced messaging
54 -- * Process management
56 , ProcessTerminationException(..)
62 -- keep the exception constructor hidden, so that handling exit
63 -- reasons /must/ take place via the 'catchExit' family of primitives
64 , ProcessExitException()
72 -- * Monitoring and linking
87 , reregisterRemoteAsync
88 , unregisterRemoteAsync
94 -- * Exception handling
118 -- * Internal Exports
122#if ! MIN_VERSION_base(4,6,0)
123import Prelude hiding (catch)
126import Data.Binary (decode)
127import Data.Time.Clock (getCurrentTime)
128import Data.Time.Format (formatTime)
129#if MIN_VERSION_time(1,5,0)
130import Data.Time.Format (defaultTimeLocale)
132import System.Locale (defaultTimeLocale)
134import System.Timeout (timeout)
135import Control.Monad (when)
136import Control.Monad.Reader (ask)
137import Control.Monad.IO.Class (liftIO)
138import Control.Applicative ((<$>))
139import Control.Exception (Exception(..), throw, throwIO, SomeException)
140import qualified Control.Exception as Ex (catch, mask, mask_, try)
141import Control.Distributed.Process.Internal.StrictMVar
146import Control.Concurrent.STM
155import Control.Distributed.Process.Internal.CQueue
160import Control.Distributed.Process.Serializable (Serializable, fingerprint)
161import Data.Accessor ((^.), (^:), (^=))
162import Control.Distributed.Static
166import Data.Rank1Typeable (Typeable)
167import qualified Control.Distributed.Static as Static (unstatic, unclosure)
168import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
169import Control.Distributed.Process.Internal.Types
179 , NodeMonitorNotification(..)
189 , ProcessExitException(..)
192 , DidUnlinkProcess(..)
197 , ProcessRegistrationException(..)
199 , ProcessInfoNone(..)
203 , createUnencodedMessage
205 , ImplicitReconnect( NoImplicitReconnect)
210import Control.Distributed.Process.Internal.Messaging
217import Control.Distributed.Process.Management.Internal.Types
220import Control.Distributed.Process.Management.Internal.Trace.Types
223import Control.Distributed.Process.Internal.WeakTQueue
231--------------------------------------------------------------------------------
233--------------------------------------------------------------------------------
236send :: Serializable a => ProcessId -> a -> Process ()
237-- This requires a lookup on every send. If we want to avoid that we need to
238-- modify serializable to allow for stateful (IO) deserialization.
241 let us = processId proc
242 node = processNode proc
243 nodeId = localNodeId node
244 destNode = (processNodeId them) in do
245 case destNode == nodeId of
246 True -> sendLocal them msg
247 False -> liftIO $ sendMessage (processNode proc)
248 (ProcessIdentifier (processId proc))
249 (ProcessIdentifier them)
252 -- We do not fire the trace event until after the sending is complete;
253 -- In the remote case, 'sendMessage' can block in the networking stack.
254 liftIO $ traceEvent (localEventBus node)
255 (MxSent them us (createUnencodedMessage msg))
257-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
258-- and (in the case when the destination process resides on the same local
259-- node) therefore ensure that the payload is fully evaluated before it is
261unsafeSend :: Serializable a => ProcessId -> a -> Process ()
262unsafeSend = Unsafe.send
264-- | Send a message unreliably.
266-- Unlike 'send', this function is insensitive to 'reconnect'. It will
267-- try to send the message regardless of the history of connection failures
270-- Message passing with 'usend' is ordered for a given sender and receiver
271-- if the messages arrive at all.
273usend :: Serializable a => ProcessId -> a -> Process ()
276 let there = processNodeId them
278 then sendLocal them msg
279 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
282-- | Wait for a message of a specific type
283expect :: forall a. Serializable a => Process a
284expect = receiveWait [match return]
286--------------------------------------------------------------------------------
288--------------------------------------------------------------------------------
290-- | Create a new typed channel
291newChan :: Serializable a => Process (SendPort a, ReceivePort a)
294 liftIO . modifyMVar (processState proc) $ \st -> do
295 let lcid = st ^. channelCounter
296 let cid = SendPortId { sendPortProcessId = processId proc
297 , sendPortLocalId = lcid
299 let sport = SendPort cid
300 chan <- liftIO newTQueueIO
301 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
302 let rport = ReceivePort $ readTQueue chan
303 let tch = TypedChannel chan'
304 return ( (channelCounter ^: (+ 1))
305 . (typedChannelWithId lcid ^= Just tch)
310 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
311 finalizer st lcid = modifyMVar_ st $
312 return . (typedChannelWithId lcid ^= Nothing)
314-- | Send a message on a typed channel
315sendChan :: Serializable a => SendPort a -> a -> Process ()
316sendChan (SendPort cid) msg = do
318 let node = localNodeId (processNode proc)
319 destNode = processNodeId (sendPortProcessId cid) in do
320 case destNode == node of
321 True -> sendChanLocal cid msg
323 liftIO $ sendBinary (processNode proc)
324 (ProcessIdentifier (processId proc))
325 (SendPortIdentifier cid)
329-- | Send a message on a typed channel. This function makes /no/ attempt to
330-- serialize and (in the case when the @ReceivePort@ resides on the same local
331-- node) therefore ensure that the payload is fully evaluated before it is
333unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
334unsafeSendChan = Unsafe.sendChan
336-- | Wait for a message on a typed channel
337receiveChan :: Serializable a => ReceivePort a -> Process a
338receiveChan = liftIO . atomically . receiveSTM
340-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
341-- non-blocking check for a message.
342receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
343receiveChanTimeout 0 ch = liftIO . atomically $
344 (Just <$> receiveSTM ch) `orElse` return Nothing
345receiveChanTimeout n ch = liftIO . timeout n . atomically $
348-- | Merge a list of typed channels.
350-- The result port is left-biased: if there are messages available on more
351-- than one port, the first available message is returned.
352mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
353mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
355-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
357mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
358mergePortsRR = \ps -> do
359 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
360 return $ ReceivePort (rr psVar)
364 rotate (x:xs) = xs ++ [x]
366 rr :: TVar [STM a] -> STM a
369 a <- foldr1 orElse ps
370 writeTVar psVar (rotate ps)
373--------------------------------------------------------------------------------
374-- Advanced messaging --
375--------------------------------------------------------------------------------
377-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
378newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
381-- | Test the matches in order against each message in the queue
382receiveWait :: [Match b] -> Process b
384 queue <- processQueue <$> ask
385 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
388-- | Like 'receiveWait' but with a timeout.
390-- If the timeout is zero do a non-blocking check for matching messages. A
391-- non-zero timeout is applied only when waiting for incoming messages (that is,
392-- /after/ we have checked the messages that are already in the mailbox).
393receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
394receiveTimeout t ms = do
395 queue <- processQueue <$> ask
396 let blockSpec = if t == 0 then NonBlocking else Timeout t
397 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
399 Nothing -> return Nothing
400 Just proc -> Just <$> proc
402-- | Match on a typed channel
403matchChan :: ReceivePort a -> (a -> Process b) -> Match b
404matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
406-- | Match on an arbitrary STM action.
408-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
409-- actions with checks against our process' mailbox and/or any typed channel
410-- @ReceivePort@s we may hold.
412-- This allows us to process multiple input streams /along with our mailbox/,
413-- in just the same way that 'matchChan' supports checking both the mailbox
414-- /and/ an arbitrary set of typed channels in one atomic transaction.
416-- Note there are no ordering guarnatees with respect to these disparate input
419matchSTM :: STM a -> (a -> Process b) -> Match b
420matchSTM stm fn = Match $ MatchChan (fmap fn stm)
422-- | Match against any message of the right type
423match :: forall a b. Serializable a => (a -> Process b) -> Match b
424match = matchIf (const True)
426-- | Match against any message of the right type that satisfies a predicate
427matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
428matchIf c p = Match $ MatchMsg $ \msg ->
429 case messageFingerprint msg == fingerprint (undefined :: a) of
432 (UnencodedMessage _ m) ->
433 let m' = unsafeCoerce m :: a in
437 (EncodedMessage _ _) ->
438 if (c decoded) then Just (p decoded) else Nothing
441 -- Make sure the value is fully decoded so that we don't hang to
442 -- bytestrings when the process calling 'matchIf' doesn't process
443 -- the values immediately
444 !decoded = decode (messageEncoding msg)
446-- | Match against any message, regardless of the underlying (contained) type
447matchMessage :: (Message -> Process Message) -> Match Message
448matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
450-- | Match against any message (regardless of underlying type) that satisfies a predicate
451matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
452matchMessageIf c p = Match $ MatchMsg $ \msg ->
457-- | Forward a raw 'Message' to the given 'ProcessId'.
458forward :: Message -> ProcessId -> Process ()
461 let node = processNode proc
463 nid = localNodeId node
464 destNode = (processNodeId them) in do
465 case destNode == nid of
466 True -> sendCtrlMsg Nothing (LocalSend them msg)
467 False -> liftIO $ sendPayload (processNode proc)
468 (ProcessIdentifier (processId proc))
469 (ProcessIdentifier them)
471 (messageToPayload msg)
472 -- We do not fire the trace event until after the sending is complete;
473 -- In the remote case, 'sendMessage' can block in the networking stack.
474 liftIO $ traceEvent (localEventBus node)
477-- | Forward a raw 'Message' to the given 'ProcessId'.
479-- Unlike 'forward', this function is insensitive to 'reconnect'. It will
480-- try to send the message regardless of the history of connection failures
482uforward :: Message -> ProcessId -> Process ()
483uforward msg them = do
485 let node = processNode proc
487 nid = localNodeId node
488 destNode = (processNodeId them) in do
489 case destNode == nid of
490 True -> sendCtrlMsg Nothing (LocalSend them msg)
491 False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)
493 -- We do not fire the trace event until after the sending is complete;
494 -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
495 liftIO $ traceEvent (localEventBus node)
498-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
499-- 'Serializable' - like the datum they contain - but also note, deserialising
500-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
501-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
506-- send self (wrapMessage "blah")
507-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
508-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
509-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
512wrapMessage :: Serializable a => a -> Message
513wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
515-- | This is the /unsafe/ variant of 'wrapMessage'. See
516-- "Control.Distributed.Process.UnsafePrimitives" for details.
517unsafeWrapMessage :: Serializable a => a -> Message
518unsafeWrapMessage = Unsafe.wrapMessage
520-- [note Serializable UnencodedMessage]
521-- if we attempt to serialise an UnencodedMessage, it will be converted to an
522-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
523-- Primitives.hs for the gory details
525-- | Attempt to unwrap a raw 'Message'.
526-- If the type of the decoded message payload matches the expected type, the
527-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
530-- This expression, for example, will evaluate to @Nothing@
531-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
533-- Whereas this expression, will yield @Just "foo"@
534-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
536unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
538 case messageFingerprint msg == fingerprint (undefined :: a) of
539 False -> return Nothing :: m (Maybe a)
541 (UnencodedMessage _ ms) ->
542 let ms' = unsafeCoerce ms :: a
544 (EncodedMessage _ _) ->
545 return (Just (decoded))
547 decoded :: a -- note [decoding]
548 !decoded = decode (messageEncoding msg)
550-- | Attempt to handle a raw 'Message'.
551-- If the type of the message matches the type of the first argument to
552-- the supplied expression, then the message will be decoded and the expression
553-- evaluated against its value. If this runtime type checking fails however,
554-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
555-- evaluation proceeds, the resulting value with be wrapped with @Just@.
557-- Intended for use in `catchesExit` and `matchAny` primitives.
559handleMessage :: forall m a b. (Monad m, Serializable a)
560 => Message -> (a -> m b) -> m (Maybe b)
561handleMessage msg proc = handleMessageIf msg (const True) proc
563-- | Conditionally handle a raw 'Message'.
564-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
565-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
566-- for further information about runtime type checking.
568handleMessageIf :: forall m a b . (Monad m, Serializable a)
573handleMessageIf msg c proc = do
574 case messageFingerprint msg == fingerprint (undefined :: a) of
575 False -> return Nothing :: m (Maybe b)
577 (UnencodedMessage _ ms) ->
578 let ms' = unsafeCoerce ms :: a in
580 True -> do { r <- proc ms'; return (Just r) }
581 False -> return Nothing :: m (Maybe b)
582 (EncodedMessage _ _) ->
584 True -> do { r <- proc (decoded :: a); return (Just r) }
585 False -> return Nothing :: m (Maybe b)
587 decoded :: a -- note [decoding]
588 !decoded = decode (messageEncoding msg)
590-- | As 'handleMessage' but ignores result, which is useful if you don't
591-- care whether or not the handler succeeded.
593handleMessage_ :: forall m a . (Monad m, Serializable a)
594 => Message -> (a -> m ()) -> m ()
595handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
597-- | Conditional version of 'handleMessage_'.
599handleMessageIf_ :: forall m a . (Monad m, Serializable a)
604handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
606-- | Match against an arbitrary message. 'matchAny' removes the first available
607-- message from the process mailbox. To handle arbitrary /raw/ messages once
608-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
610matchAny :: forall b. (Message -> Process b) -> Match b
611matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
613-- | Match against an arbitrary message. Intended for use with 'handleMessage'
614-- and 'unwrapMessage', this function /only/ removes a message from the process
615-- mailbox, /if/ the supplied condition matches. The success (or failure) of
616-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
617-- here, i.e., if the condition evaluates to @True@ then the message will
618-- be removed from the process mailbox and decoded, but that does /not/
619-- guarantee that an expression passed to @handleMessage@ will pass the
620-- runtime type checks and therefore be evaluated.
622matchAnyIf :: forall a b. (Serializable a)
624 -> (Message -> Process b)
626matchAnyIf c p = Match $ MatchMsg $ \msg ->
627 case messageFingerprint msg == fingerprint (undefined :: a) of
628 True | check -> Just (p msg)
633 (EncodedMessage _ _) -> c decoded
634 (UnencodedMessage _ m') -> c (unsafeCoerce m')
636 decoded :: a -- note [decoding]
637 !decoded = decode (messageEncoding msg)
641For an EncodedMessage, we need to ensure the value is fully decoded so that
642we don't hang to bytestrings if the calling process doesn't evaluate
643immediately. For UnencodedMessage we know (because the fingerprint comparison
644succeeds) that unsafeCoerce will not fail.
647-- | Remove any message from the queue
648matchUnknown :: Process b -> Match b
649matchUnknown p = Match $ MatchMsg (const (Just p))
651-- | Receives messages and forwards them to @pid@ if @p msg == True@.
652delegate :: ProcessId -> (Message -> Bool) -> Process ()
655 matchAny (\m -> case (p m) of
656 True -> forward m pid
661-- | A straight relay that forwards all messages to the supplied pid.
662relay :: ProcessId -> Process ()
663relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
665-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
666-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
667-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
668-- fails, no action is taken and the proxy process will continue running.
669proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
673 next <- handleMessage m proc
675 Just True -> forward m pid
676 Just False -> return () -- explicitly ignored
677 Nothing -> return ()) -- un-routable
681--------------------------------------------------------------------------------
682-- Process management --
683--------------------------------------------------------------------------------
685-- | Thrown by 'terminate'
686data ProcessTerminationException = ProcessTerminationException
687 deriving (Show, Typeable)
689instance Exception ProcessTerminationException
691-- | Terminate immediately (throws a ProcessTerminationException)
692terminate :: Process a
693terminate = liftIO $ throwIO ProcessTerminationException
696-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
697die :: Serializable a => a -> Process b
699-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
700-- could be decoded by a handler passed to 'catchExit', re-thrown or even
701-- passed to another process via the tracing mechanism.
703 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
705-- | Forceful request to kill a process. Where 'exit' provides an exception
706-- that can be caught and handled, 'kill' throws an unexposed exception type
707-- which cannot be handled explicitly (by type).
708kill :: ProcessId -> String -> Process ()
709-- NOTE: We send the message to our local node controller, which will then
710-- forward it to a remote node controller (if applicable). Sending it directly
711-- to a remote node controller means that that the message may overtake a
712-- 'monitor' or 'link' request.
713kill them reason = sendCtrlMsg Nothing (Kill them reason)
715-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
716-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
717-- manner can be handled using the 'catchExit' family of functions.
718exit :: Serializable a => ProcessId -> a -> Process ()
719-- NOTE: We send the message to our local node controller, which will then
720-- forward it to a remote node controller (if applicable). Sending it directly
721-- to a remote node controller means that that the message may overtake a
722-- 'monitor' or 'link' request.
723exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
725-- | Catches 'ProcessExitException'. The handler will not be applied unless its
726-- type matches the encoded data stored in the exception (see the /reason/
727-- argument given to the 'exit' primitive). If the handler cannot be applied,
728-- the exception will be re-thrown.
730-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
731-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
732catchExit :: forall a b . (Show a, Serializable a)
734 -> (ProcessId -> a -> Process b)
736catchExit act exitHandler = catch act handleExit
738 handleExit ex@(ProcessExitException from msg) =
739 if messageFingerprint msg == fingerprint (undefined :: a)
740 then exitHandler from decoded
741 else liftIO $ throwIO ex
744 -- Make sure the value is fully decoded so that we don't hang to
745 -- bytestrings if the caller doesn't use the value immediately
746 !decoded = decode (messageEncoding msg)
748-- | Lift 'Control.Exception.catches' (almost).
750-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
751-- message, a handler must accept inputs of the expected type. In order to
752-- handle a list of potentially different handlers (and therefore input types),
753-- a handler passed to 'catchesExit' must accept 'Message' and return
754-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
756-- See 'maybeHandleMessage' and 'Message' for more details.
757catchesExit :: Process b
758 -> [(ProcessId -> Message -> (Process (Maybe b)))]
760catchesExit act handlers = catch act ((flip handleExit) handlers)
762 handleExit :: ProcessExitException
763 -> [(ProcessId -> Message -> Process (Maybe b))]
765 handleExit ex [] = liftIO $ throwIO ex
766 handleExit ex@(ProcessExitException from msg) (h:hs) = do
769 Nothing -> handleExit ex hs
772-- | Our own process ID
773getSelfPid :: Process ProcessId
774getSelfPid = processId <$> ask
776-- | Get the node ID of our local node
777getSelfNode :: Process NodeId
778getSelfNode = localNodeId . processNode <$> ask
781-- | Get statistics about the specified node
782getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
784 selfNode <- getSelfNode
786 then Right `fmap` getLocalNodeStats -- optimisation
787 else getNodeStatsRemote selfNode
789 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
790 getNodeStatsRemote selfNode = do
791 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
792 bracket (monitorNode nid) unmonitor $ \mRef ->
793 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
794 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
795 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
798-- | Get statistics about our local node
799getLocalNodeStats :: Process NodeStats
800getLocalNodeStats = do
802 sendCtrlMsg Nothing $ GetNodeStats self
803 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
805-- | Get information about the specified process
806getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
808 let them = processNodeId pid in do
810 dest <- mkNode them us
811 sendCtrlMsg dest $ GetInfo pid
813 match (\(p :: ProcessInfo) -> return $ Just p)
814 , match (\(_ :: ProcessInfoNone) -> return Nothing)
816 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
817 mkNode them us = case them == us of
818 True -> return Nothing
819 _ -> return $ Just them
821--------------------------------------------------------------------------------
822-- Monitoring and linking --
823--------------------------------------------------------------------------------
825-- | Link to a remote process (asynchronous)
827-- When process A links to process B (that is, process A calls
828-- @link pidB@) then an asynchronous exception will be thrown to process A
829-- when process B terminates (normally or abnormally), or when process A gets
830-- disconnected from process B. Although it is /technically/ possible to catch
831-- these exceptions, chances are if you find yourself trying to do so you should
832-- probably be using 'monitor' rather than 'link'. In particular, code such as
834-- > link pidB -- Link to process B
835-- > expect -- Wait for a message from process B
836-- > unlink pidB -- Unlink again
838-- doesn't quite do what one might expect: if process B sends a message to
839-- process A, and /subsequently terminates/, then process A might or might not
840-- be terminated too, depending on whether the exception is thrown before or
841-- after the 'unlink' (i.e., this code has a race condition).
843-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
844-- call to 'link' has no effect.
846-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
847-- Linking makes no distinction between normal and abnormal termination of
848-- the remote process.
849link :: ProcessId -> Process ()
850link = sendCtrlMsg Nothing . Link . ProcessIdentifier
852-- | Monitor another process (asynchronous)
854-- When process A monitors process B (that is, process A calls
855-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
856-- when process B terminates (normally or abnormally), or when process A gets
857-- disconnected from process B. You receive this message like any other (using
858-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
859-- 'DiedDisconnect', etc.).
861-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
862-- multiple monitors are set up, multiple notifications will be delivered
863-- and monitors can be disabled individually using 'unmonitor'.
864monitor :: ProcessId -> Process MonitorRef
865monitor = monitor' . ProcessIdentifier
867-- | Establishes temporary monitoring of another process.
869-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
870-- of @code@. Note: although monitoring is no longer active when
871-- @withMonitor@ returns, there might still be unreceived monitor
872-- messages in the queue.
874withMonitor :: ProcessId -> Process a -> Process a
875withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
876 -- unmonitor blocks waiting for the response, so there's a possibility
877 -- that an exception might interrupt withMonitor before the unmonitor
878 -- has completed. I think that's better than making the unmonitor
883-- This is synchronous in the sense that once it returns you are guaranteed
884-- that no exception will be raised if the remote process dies. However, it is
885-- asynchronous in the sense that we do not wait for a response from the remote
887unlink :: ProcessId -> Process ()
890 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
894-- | Remove a node link
896-- This has the same synchronous/asynchronous nature as 'unlink'.
897unlinkNode :: NodeId -> Process ()
900 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
904-- | Remove a channel (send port) link
906-- This has the same synchronous/asynchronous nature as 'unlink'.
907unlinkPort :: SendPort a -> Process ()
909 unlinkPortAsync sport
910 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
916-- This has the same synchronous/asynchronous nature as 'unlink'.
917unmonitor :: MonitorRef -> Process ()
920 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
924--------------------------------------------------------------------------------
925-- Exception handling --
926--------------------------------------------------------------------------------
928-- | Lift 'Control.Exception.catch'
929catch :: Exception e => Process a -> (e -> Process a) -> Process a
932 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
934-- | Lift 'Control.Exception.try'
935try :: Exception e => Process a -> Process (Either e a)
938 liftIO $ Ex.try (runLocalProcess lproc p)
940-- | Lift 'Control.Exception.mask'
941mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
944 liftIO $ Ex.mask $ \restore ->
945 runLocalProcess lproc (p (liftRestore restore))
947 liftRestore :: (forall a. IO a -> IO a)
948 -> (forall a. Process a -> Process a)
949 liftRestore restoreIO = \p2 -> do
951 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
953-- | Lift 'Control.Exception.mask_'
954mask_ :: Process a -> Process a
957 liftIO $ Ex.mask_ $ runLocalProcess lproc p
959-- | Lift 'Control.Exception.onException'
960onException :: Process a -> Process b -> Process a
961onException p what = p `catch` \e -> do _ <- what
962 liftIO $ throwIO (e :: SomeException)
964-- | Lift 'Control.Exception.bracket'
965bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
966bracket before after thing =
967 mask $ \restore -> do
969 r <- restore (thing a) `onException` after a
973-- | Lift 'Control.Exception.bracket_'
974bracket_ :: Process a -> Process b -> Process c -> Process c
975bracket_ before after thing = bracket before (const after) (const thing)
977-- | Lift 'Control.Exception.finally'
978finally :: Process a -> Process b -> Process a
979finally a sequel = bracket_ (return ()) sequel a
981-- | You need this when using 'catches'
982data Handler a = forall e . Exception e => Handler (e -> Process a)
984instance Functor Handler where
985 fmap f (Handler h) = Handler (fmap f . h)
987-- | Lift 'Control.Exception.catches'
988catches :: Process a -> [Handler a] -> Process a
989catches proc handlers = proc `catch` catchesHandler handlers
991catchesHandler :: [Handler a] -> SomeException -> Process a
992catchesHandler handlers e = foldr tryHandler (throw e) handlers
993 where tryHandler (Handler handler) res
994 = case fromException e of
995 Just e' -> handler e'
998--------------------------------------------------------------------------------
1000--------------------------------------------------------------------------------
1002-- | Like 'expect' but with a timeout
1003expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
1004expectTimeout n = receiveTimeout n [match return]
1006-- | Asynchronous version of 'spawn'
1008-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
1009spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
1010spawnAsync nid proc = do
1011 spawnRef <- getSpawnRef
1014 then sendCtrlMsg Nothing $ Spawn proc spawnRef
1015 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
1018-- | Monitor a node (asynchronous)
1019monitorNode :: NodeId -> Process MonitorRef
1021 monitor' . NodeIdentifier
1023-- | Monitor a typed channel (asynchronous)
1024monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1025monitorPort (SendPort cid) =
1026 monitor' (SendPortIdentifier cid)
1028-- | Remove a monitor (asynchronous)
1029unmonitorAsync :: MonitorRef -> Process ()
1031 sendCtrlMsg Nothing . Unmonitor
1033-- | Link to a node (asynchronous)
1034linkNode :: NodeId -> Process ()
1035linkNode = link' . NodeIdentifier
1037-- | Link to a channel (asynchronous)
1038linkPort :: SendPort a -> Process ()
1039linkPort (SendPort cid) =
1040 link' (SendPortIdentifier cid)
1042-- | Remove a process link (asynchronous)
1043unlinkAsync :: ProcessId -> Process ()
1045 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1047-- | Remove a node link (asynchronous)
1048unlinkNodeAsync :: NodeId -> Process ()
1050 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1052-- | Remove a channel (send port) link (asynchronous)
1053unlinkPortAsync :: SendPort a -> Process ()
1054unlinkPortAsync (SendPort cid) =
1055 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1057--------------------------------------------------------------------------------
1059--------------------------------------------------------------------------------
1063-- @say message@ sends a message (time, pid of the current process, message)
1064-- to the process registered as 'logger'. By default, this process simply
1065-- sends the string to 'stderr'. Individual Cloud Haskell backends might
1066-- replace this with a different logger process, however.
1067say :: String -> Process ()
1069 now <- liftIO getCurrentTime
1071 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
1073--------------------------------------------------------------------------------
1075--------------------------------------------------------------------------------
1077-- | Register a process with the local registry (synchronous). The name must not
1078-- already be registered. The process need not be on this node. A bad
1079-- registration will result in a 'ProcessRegistrationException'
1081-- The process to be registered does not have to be local itself.
1082register :: String -> ProcessId -> Process ()
1083register = registerImpl False
1085-- | Like 'register', but will replace an existing registration.
1086-- The name must already be registered.
1087reregister :: String -> ProcessId -> Process ()
1088reregister = registerImpl True
1090registerImpl :: Bool -> String -> ProcessId -> Process ()
1091registerImpl force label pid = do
1092 mynid <- getSelfNode
1093 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1095 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1096 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1099-- | Register a process with a remote registry (asynchronous).
1101-- The process to be registered does not have to live on the same remote node.
1102-- Reply wil come in the form of a 'RegisterReply' message
1104-- See comments in 'whereisRemoteAsync'
1105registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1106registerRemoteAsync nid label pid =
1107 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
1109reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1110reregisterRemoteAsync nid label pid =
1111 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1113-- | Remove a process from the local registry (asynchronous).
1114-- This version will wait until a response is gotten from the
1115-- management process. The name must already be registered.
1116unregister :: String -> Process ()
1117unregister label = do
1118 mynid <- getSelfNode
1119 sendCtrlMsg Nothing (Register label mynid Nothing False)
1121 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1122 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1125-- | Deal with the result from an attempted registration or unregistration
1126-- by throwing an exception if necessary
1127handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()
1128handleRegistrationReply label ok owner =
1130 liftIO $ throwIO $ ProcessRegistrationException label owner
1132-- | Remove a process from a remote registry (asynchronous).
1134-- Reply wil come in the form of a 'RegisterReply' message
1136-- See comments in 'whereisRemoteAsync'
1137unregisterRemoteAsync :: NodeId -> String -> Process ()
1138unregisterRemoteAsync nid label =
1139 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1141-- | Query the local process registry
1142whereis :: String -> Process (Maybe ProcessId)
1144 sendCtrlMsg Nothing (WhereIs label)
1145 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1146 (\(WhereIsReply _ mPid) -> return mPid)
1149-- | Query a remote process registry (asynchronous)
1151-- Reply will come in the form of a 'WhereIsReply' message.
1153-- There is currently no synchronous version of 'whereisRemoteAsync': if
1154-- you implement one yourself, be sure to take into account that the remote
1155-- node might die or get disconnect before it can respond (i.e. you should
1156-- use 'monitorNode' and take appropriate action when you receive a
1157-- 'NodeMonitorNotification').
1158whereisRemoteAsync :: NodeId -> String -> Process ()
1159whereisRemoteAsync nid label =
1160 sendCtrlMsg (Just nid) (WhereIs label)
1162-- | Named send to a process in the local registry (asynchronous)
1163nsend :: Serializable a => String -> a -> Process ()
1165 sendCtrlMsg Nothing (NamedSend label (createMessage msg))
1167-- | Named send to a process in the local registry (asynchronous).
1168-- This function makes /no/ attempt to serialize and (in the case when the
1169-- destination process resides on the same local node) therefore ensure that
1170-- the payload is fully evaluated before it is delivered.
1171unsafeNSend :: Serializable a => String -> a -> Process ()
1172unsafeNSend = Unsafe.nsend
1174-- | Named send to a process in a remote registry (asynchronous)
1175nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1176nsendRemote nid label msg =
1177 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1179--------------------------------------------------------------------------------
1181--------------------------------------------------------------------------------
1183-- | Resolve a static value
1184unStatic :: Typeable a => Static a -> Process a
1186 rtable <- remoteTable . processNode <$> ask
1187 case Static.unstatic rtable static of
1188 Left err -> fail $ "Could not resolve static value: " ++ err
1191-- | Resolve a closure
1192unClosure :: Typeable a => Closure a -> Process a
1193unClosure closure = do
1194 rtable <- remoteTable . processNode <$> ask
1195 case Static.unclosure rtable closure of
1196 Left err -> fail $ "Could not resolve closure: " ++ err
1199--------------------------------------------------------------------------------
1201--------------------------------------------------------------------------------
1203-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1204-- message passing. However, when network connections get disrupted this
1205-- illusion cannot always be maintained. Once a network connection breaks (even
1206-- temporarily) no further communication on that connection will be possible.
1207-- For example, if process A sends a message to process B, and A is then
1208-- notified (by monitor notification) that it got disconnected from B, A will
1209-- not be able to send any further messages to B, /unless/ A explicitly
1210-- indicates that it is acceptable to attempt to reconnect to B using the
1211-- Cloud Haskell 'reconnect' primitive.
1213-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1214-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1215-- then receives a monitor notification that its connection to B has been lost,
1216-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1217-- and m3 but not m2.
1219-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1220-- to attempt to reconnect on the next send/. In particular, if no further
1221-- communication attempts are made to B then A can use reconnect to clean up
1222-- its connection to B.
1223reconnect :: ProcessId -> Process ()
1226 node <- processNode <$> ask
1227 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1229-- | Reconnect to a sendport. See 'reconnect' for more information.
1230reconnectPort :: SendPort a -> Process ()
1231reconnectPort them = do
1233 node <- processNode <$> ask
1234 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1236--------------------------------------------------------------------------------
1237-- Auxiliary functions --
1238--------------------------------------------------------------------------------
1240sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1242 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1244sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1245sendChanLocal spId msg =
1246 -- we *must* fully serialize/encode the message here, because
1247 -- attempting to use `unsafeCoerce' in the node controller
1248 -- won't work since we know nothing about the required type
1249 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1251getMonitorRefFor :: Identifier -> Process MonitorRef
1252getMonitorRefFor ident = do
1254 liftIO $ modifyMVar (processState proc) $ \st -> do
1255 let counter = st ^. monitorCounter
1256 return ( monitorCounter ^: (+ 1) $ st
1257 , MonitorRef ident counter
1260getSpawnRef :: Process SpawnRef
1263 liftIO $ modifyMVar (processState proc) $ \st -> do
1264 let counter = st ^. spawnCounter
1265 return ( spawnCounter ^: (+ 1) $ st
1269-- | Monitor a process/node/channel
1270monitor' :: Identifier -> Process MonitorRef
1272 monitorRef <- getMonitorRefFor ident
1273 sendCtrlMsg Nothing $ Monitor monitorRef
1276-- | Link to a process/node/channel
1277link' :: Identifier -> Process ()
1278link' = 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