2{-# LANGUAGE RankNTypes #-}
3{-# LANGUAGE DeriveDataTypeable #-}
4{-# LANGUAGE ScopedTypeVariables #-}
5{-# LANGUAGE ExistentialQuantification #-}
6{-# LANGUAGE BangPatterns #-}
8-- | Cloud Haskell primitives
10-- We define these in a separate module so that we don't have to rely on
11-- the closure combinators
12module Control.Distributed.Process.Internal.Primitives
13 ( -- * Basic messaging
23 -- * Unsafe messaging variants
27 -- * Advanced messaging
52 -- * Process management
54 , ProcessTerminationException(..)
60 -- keep the exception constructor hidden, so that handling exit
61 -- reasons /must/ take place via the 'catchExit' family of primitives
62 , ProcessExitException()
70 -- * Monitoring and linking
85 , reregisterRemoteAsync
86 , unregisterRemoteAsync
92 -- * Exception handling
116 -- * Internal Exports
120#if ! MIN_VERSION_base(4,6,0)
121import Prelude hiding (catch)
124import Data.Binary (decode)
125import Data.Time.Clock (getCurrentTime)
126import Data.Time.Format (formatTime)
127import System.Locale (defaultTimeLocale)
128import System.Timeout (timeout)
129import Control.Monad (when)
130import Control.Monad.Reader (ask)
131import Control.Monad.IO.Class (liftIO)
132import Control.Applicative ((<$>))
133import Control.Exception (Exception(..), throw, throwIO, SomeException)
134import qualified Control.Exception as Ex (catch, mask, mask_, try)
135import Control.Distributed.Process.Internal.StrictMVar
140import Control.Concurrent.STM
149import Control.Distributed.Process.Internal.CQueue
154import Control.Distributed.Process.Serializable (Serializable, fingerprint)
155import Data.Accessor ((^.), (^:), (^=))
156import Control.Distributed.Static
160import Data.Rank1Typeable (Typeable)
161import qualified Control.Distributed.Static as Static (unstatic, unclosure)
162import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
163import Control.Distributed.Process.Internal.Types
173 , NodeMonitorNotification(..)
183 , ProcessExitException(..)
186 , DidUnlinkProcess(..)
191 , ProcessRegistrationException(..)
193 , ProcessInfoNone(..)
197 , createUnencodedMessage
199 , ImplicitReconnect( NoImplicitReconnect)
204import Control.Distributed.Process.Internal.Messaging
211import Control.Distributed.Process.Management.Internal.Types
214import Control.Distributed.Process.Management.Internal.Trace.Types
217import Control.Distributed.Process.Internal.WeakTQueue
225--------------------------------------------------------------------------------
227--------------------------------------------------------------------------------
230send :: Serializable a => ProcessId -> a -> Process ()
231-- This requires a lookup on every send. If we want to avoid that we need to
232-- modify serializable to allow for stateful (IO) deserialization.
235 let us = processId proc
236 node = processNode proc
237 nodeId = localNodeId node
238 destNode = (processNodeId them) in do
239 case destNode == nodeId of
240 True -> sendLocal them msg
241 False -> liftIO $ sendMessage (processNode proc)
242 (ProcessIdentifier (processId proc))
243 (ProcessIdentifier them)
246 -- We do not fire the trace event until after the sending is complete;
247 -- In the remote case, 'sendMessage' can block in the networking stack.
248 liftIO $ traceEvent (localEventBus node)
249 (MxSent them us (createUnencodedMessage msg))
251-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
252-- and (in the case when the destination process resides on the same local
253-- node) therefore ensure that the payload is fully evaluated before it is
255unsafeSend :: Serializable a => ProcessId -> a -> Process ()
256unsafeSend = Unsafe.send
258-- | Send a message unreliably.
260-- Unlike 'send', this function is insensitive to 'reconnect'. It will
261-- try to send the message regardless of the history of connection failures
264-- Message passing with 'usend' is ordered for a given sender and receiver
265-- if the messages arrive at all.
267usend :: Serializable a => ProcessId -> a -> Process ()
270 let there = processNodeId them
272 then sendLocal them msg
273 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
276-- | Wait for a message of a specific type
277expect :: forall a. Serializable a => Process a
278expect = receiveWait [match return]
280--------------------------------------------------------------------------------
282--------------------------------------------------------------------------------
284-- | Create a new typed channel
285newChan :: Serializable a => Process (SendPort a, ReceivePort a)
288 liftIO . modifyMVar (processState proc) $ \st -> do
289 let lcid = st ^. channelCounter
290 let cid = SendPortId { sendPortProcessId = processId proc
291 , sendPortLocalId = lcid
293 let sport = SendPort cid
294 chan <- liftIO newTQueueIO
295 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
296 let rport = ReceivePort $ readTQueue chan
297 let tch = TypedChannel chan'
298 return ( (channelCounter ^: (+ 1))
299 . (typedChannelWithId lcid ^= Just tch)
304 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
305 finalizer st lcid = modifyMVar_ st $
306 return . (typedChannelWithId lcid ^= Nothing)
308-- | Send a message on a typed channel
309sendChan :: Serializable a => SendPort a -> a -> Process ()
310sendChan (SendPort cid) msg = do
312 let node = localNodeId (processNode proc)
313 destNode = processNodeId (sendPortProcessId cid) in do
314 case destNode == node of
315 True -> sendChanLocal cid msg
317 liftIO $ sendBinary (processNode proc)
318 (ProcessIdentifier (processId proc))
319 (SendPortIdentifier cid)
323-- | Send a message on a typed channel. This function makes /no/ attempt to
324-- serialize and (in the case when the @ReceivePort@ resides on the same local
325-- node) therefore ensure that the payload is fully evaluated before it is
327unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
328unsafeSendChan = Unsafe.sendChan
330-- | Wait for a message on a typed channel
331receiveChan :: Serializable a => ReceivePort a -> Process a
332receiveChan = liftIO . atomically . receiveSTM
334-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
335-- non-blocking check for a message.
336receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
337receiveChanTimeout 0 ch = liftIO . atomically $
338 (Just <$> receiveSTM ch) `orElse` return Nothing
339receiveChanTimeout n ch = liftIO . timeout n . atomically $
342-- | Merge a list of typed channels.
344-- The result port is left-biased: if there are messages available on more
345-- than one port, the first available message is returned.
346mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
347mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
349-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
351mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
352mergePortsRR = \ps -> do
353 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
354 return $ ReceivePort (rr psVar)
358 rotate (x:xs) = xs ++ [x]
360 rr :: TVar [STM a] -> STM a
363 a <- foldr1 orElse ps
364 writeTVar psVar (rotate ps)
367--------------------------------------------------------------------------------
368-- Advanced messaging --
369--------------------------------------------------------------------------------
371-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
372newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
374-- | Test the matches in order against each message in the queue
375receiveWait :: [Match b] -> Process b
377 queue <- processQueue <$> ask
378 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
381-- | Like 'receiveWait' but with a timeout.
383-- If the timeout is zero do a non-blocking check for matching messages. A
384-- non-zero timeout is applied only when waiting for incoming messages (that is,
385-- /after/ we have checked the messages that are already in the mailbox).
386receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
387receiveTimeout t ms = do
388 queue <- processQueue <$> ask
389 let blockSpec = if t == 0 then NonBlocking else Timeout t
390 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
392 Nothing -> return Nothing
393 Just proc -> Just <$> proc
395-- | Match on a typed channel
396matchChan :: ReceivePort a -> (a -> Process b) -> Match b
397matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
399-- | Match on an arbitrary STM action.
401-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
402-- actions with checks against our process' mailbox and/or any typed channel
403-- @ReceivePort@s we may hold.
405-- This allows us to process multiple input streams /along with our mailbox/,
406-- in just the same way that 'matchChan' supports checking both the mailbox
407-- /and/ an arbitrary set of typed channels in one atomic transaction.
409-- Note there are no ordering guarnatees with respect to these disparate input
412matchSTM :: STM a -> (a -> Process b) -> Match b
413matchSTM stm fn = Match $ MatchChan (fmap fn stm)
415-- | Match against any message of the right type
416match :: forall a b. Serializable a => (a -> Process b) -> Match b
417match = matchIf (const True)
419-- | Match against any message of the right type that satisfies a predicate
420matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
421matchIf c p = Match $ MatchMsg $ \msg ->
422 case messageFingerprint msg == fingerprint (undefined :: a) of
425 (UnencodedMessage _ m) ->
426 let m' = unsafeCoerce m :: a in
430 (EncodedMessage _ _) ->
431 if (c decoded) then Just (p decoded) else Nothing
434 -- Make sure the value is fully decoded so that we don't hang to
435 -- bytestrings when the process calling 'matchIf' doesn't process
436 -- the values immediately
437 !decoded = decode (messageEncoding msg)
439-- | Match against any message, regardless of the underlying (contained) type
440matchMessage :: (Message -> Process Message) -> Match Message
441matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
443-- | Match against any message (regardless of underlying type) that satisfies a predicate
444matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
445matchMessageIf c p = Match $ MatchMsg $ \msg ->
450-- | Forward a raw 'Message' to the given 'ProcessId'.
451forward :: Message -> ProcessId -> Process ()
454 let node = processNode proc
456 nid = localNodeId node
457 destNode = (processNodeId them) in do
458 case destNode == nid of
459 True -> sendCtrlMsg Nothing (LocalSend them msg)
460 False -> liftIO $ sendPayload (processNode proc)
461 (ProcessIdentifier (processId proc))
462 (ProcessIdentifier them)
464 (messageToPayload msg)
465 -- We do not fire the trace event until after the sending is complete;
466 -- In the remote case, 'sendMessage' can block in the networking stack.
467 liftIO $ traceEvent (localEventBus node)
471-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
472-- 'Serializable' - like the datum they contain - but also note, deserialising
473-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
474-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
479-- send self (wrapMessage "blah")
480-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
481-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
482-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
485wrapMessage :: Serializable a => a -> Message
486wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
488-- | This is the /unsafe/ variant of 'wrapMessage'. See
489-- "Control.Distributed.Process.UnsafePrimitives" for details.
490unsafeWrapMessage :: Serializable a => a -> Message
491unsafeWrapMessage = Unsafe.wrapMessage
493-- [note Serializable UnencodedMessage]
494-- if we attempt to serialise an UnencodedMessage, it will be converted to an
495-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
496-- Primitives.hs for the gory details
498-- | Attempt to unwrap a raw 'Message'.
499-- If the type of the decoded message payload matches the expected type, the
500-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
503-- This expression, for example, will evaluate to @Nothing@
504-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
506-- Whereas this expression, will yield @Just "foo"@
507-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
509unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
511 case messageFingerprint msg == fingerprint (undefined :: a) of
512 False -> return Nothing :: m (Maybe a)
514 (UnencodedMessage _ ms) ->
515 let ms' = unsafeCoerce ms :: a
517 (EncodedMessage _ _) ->
518 return (Just (decoded))
520 decoded :: a -- note [decoding]
521 !decoded = decode (messageEncoding msg)
523-- | Attempt to handle a raw 'Message'.
524-- If the type of the message matches the type of the first argument to
525-- the supplied expression, then the message will be decoded and the expression
526-- evaluated against its value. If this runtime type checking fails however,
527-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
528-- evaluation proceeds, the resulting value with be wrapped with @Just@.
530-- Intended for use in `catchesExit` and `matchAny` primitives.
532handleMessage :: forall m a b. (Monad m, Serializable a)
533 => Message -> (a -> m b) -> m (Maybe b)
534handleMessage msg proc = handleMessageIf msg (const True) proc
536-- | Conditionally handle a raw 'Message'.
537-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
538-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
539-- for further information about runtime type checking.
541handleMessageIf :: forall m a b . (Monad m, Serializable a)
546handleMessageIf msg c proc = do
547 case messageFingerprint msg == fingerprint (undefined :: a) of
548 False -> return Nothing :: m (Maybe b)
550 (UnencodedMessage _ ms) ->
551 let ms' = unsafeCoerce ms :: a in
553 True -> do { r <- proc ms'; return (Just r) }
554 False -> return Nothing :: m (Maybe b)
555 (EncodedMessage _ _) ->
557 True -> do { r <- proc (decoded :: a); return (Just r) }
558 False -> return Nothing :: m (Maybe b)
560 decoded :: a -- note [decoding]
561 !decoded = decode (messageEncoding msg)
563-- | As 'handleMessage' but ignores result, which is useful if you don't
564-- care whether or not the handler succeeded.
566handleMessage_ :: forall m a . (Monad m, Serializable a)
567 => Message -> (a -> m ()) -> m ()
568handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
570-- | Conditional version of 'handleMessage_'.
572handleMessageIf_ :: forall m a . (Monad m, Serializable a)
577handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
579-- | Match against an arbitrary message. 'matchAny' removes the first available
580-- message from the process mailbox. To handle arbitrary /raw/ messages once
581-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
583matchAny :: forall b. (Message -> Process b) -> Match b
584matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
586-- | Match against an arbitrary message. Intended for use with 'handleMessage'
587-- and 'unwrapMessage', this function /only/ removes a message from the process
588-- mailbox, /if/ the supplied condition matches. The success (or failure) of
589-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
590-- here, i.e., if the condition evaluates to @True@ then the message will
591-- be removed from the process mailbox and decoded, but that does /not/
592-- guarantee that an expression passed to @handleMessage@ will pass the
593-- runtime type checks and therefore be evaluated.
595matchAnyIf :: forall a b. (Serializable a)
597 -> (Message -> Process b)
599matchAnyIf c p = Match $ MatchMsg $ \msg ->
600 case messageFingerprint msg == fingerprint (undefined :: a) of
601 True | check -> Just (p msg)
606 (EncodedMessage _ _) -> c decoded
607 (UnencodedMessage _ m') -> c (unsafeCoerce m')
609 decoded :: a -- note [decoding]
610 !decoded = decode (messageEncoding msg)
614For an EncodedMessage, we need to ensure the value is fully decoded so that
615we don't hang to bytestrings if the calling process doesn't evaluate
616immediately. For UnencodedMessage we know (because the fingerprint comparison
617succeeds) that unsafeCoerce will not fail.
620-- | Remove any message from the queue
621matchUnknown :: Process b -> Match b
622matchUnknown p = Match $ MatchMsg (const (Just p))
624-- | Receives messages and forwards them to @pid@ if @p msg == True@.
625delegate :: ProcessId -> (Message -> Bool) -> Process ()
628 matchAny (\m -> case (p m) of
629 True -> forward m pid
634-- | A straight relay that forwards all messages to the supplied pid.
635relay :: ProcessId -> Process ()
636relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
638-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
639-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
640-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
641-- fails, no action is taken and the proxy process will continue running.
642proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
646 next <- handleMessage m proc
648 Just True -> forward m pid
649 Just False -> return () -- explicitly ignored
650 Nothing -> return ()) -- un-routable
654--------------------------------------------------------------------------------
655-- Process management --
656--------------------------------------------------------------------------------
658-- | Thrown by 'terminate'
659data ProcessTerminationException = ProcessTerminationException
660 deriving (Show, Typeable)
662instance Exception ProcessTerminationException
664-- | Terminate immediately (throws a ProcessTerminationException)
665terminate :: Process a
666terminate = liftIO $ throwIO ProcessTerminationException
669-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
670die :: Serializable a => a -> Process b
672-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
673-- could be decoded by a handler passed to 'catchExit', re-thrown or even
674-- passed to another process via the tracing mechanism.
676 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
678-- | Forceful request to kill a process. Where 'exit' provides an exception
679-- that can be caught and handled, 'kill' throws an unexposed exception type
680-- which cannot be handled explicitly (by type).
681kill :: ProcessId -> String -> Process ()
682-- NOTE: We send the message to our local node controller, which will then
683-- forward it to a remote node controller (if applicable). Sending it directly
684-- to a remote node controller means that that the message may overtake a
685-- 'monitor' or 'link' request.
686kill them reason = sendCtrlMsg Nothing (Kill them reason)
688-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
689-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
690-- manner can be handled using the 'catchExit' family of functions.
691exit :: Serializable a => ProcessId -> a -> Process ()
692-- NOTE: We send the message to our local node controller, which will then
693-- forward it to a remote node controller (if applicable). Sending it directly
694-- to a remote node controller means that that the message may overtake a
695-- 'monitor' or 'link' request.
696exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
698-- | Catches 'ProcessExitException'. The handler will not be applied unless its
699-- type matches the encoded data stored in the exception (see the /reason/
700-- argument given to the 'exit' primitive). If the handler cannot be applied,
701-- the exception will be re-thrown.
703-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
704-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
705catchExit :: forall a b . (Show a, Serializable a)
707 -> (ProcessId -> a -> Process b)
709catchExit act exitHandler = catch act handleExit
711 handleExit ex@(ProcessExitException from msg) =
712 if messageFingerprint msg == fingerprint (undefined :: a)
713 then exitHandler from decoded
714 else liftIO $ throwIO ex
717 -- Make sure the value is fully decoded so that we don't hang to
718 -- bytestrings if the caller doesn't use the value immediately
719 !decoded = decode (messageEncoding msg)
721-- | Lift 'Control.Exception.catches' (almost).
723-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
724-- message, a handler must accept inputs of the expected type. In order to
725-- handle a list of potentially different handlers (and therefore input types),
726-- a handler passed to 'catchesExit' must accept 'Message' and return
727-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
729-- See 'maybeHandleMessage' and 'Message' for more details.
730catchesExit :: Process b
731 -> [(ProcessId -> Message -> (Process (Maybe b)))]
733catchesExit act handlers = catch act ((flip handleExit) handlers)
735 handleExit :: ProcessExitException
736 -> [(ProcessId -> Message -> Process (Maybe b))]
738 handleExit ex [] = liftIO $ throwIO ex
739 handleExit ex@(ProcessExitException from msg) (h:hs) = do
742 Nothing -> handleExit ex hs
745-- | Our own process ID
746getSelfPid :: Process ProcessId
747getSelfPid = processId <$> ask
749-- | Get the node ID of our local node
750getSelfNode :: Process NodeId
751getSelfNode = localNodeId . processNode <$> ask
754-- | Get statistics about the specified node
755getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
757 selfNode <- getSelfNode
759 then Right `fmap` getLocalNodeStats -- optimisation
760 else getNodeStatsRemote selfNode
762 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
763 getNodeStatsRemote selfNode = do
764 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
765 bracket (monitorNode nid) unmonitor $ \mRef ->
766 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
767 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
768 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
771-- | Get statistics about our local node
772getLocalNodeStats :: Process NodeStats
773getLocalNodeStats = do
775 sendCtrlMsg Nothing $ GetNodeStats self
776 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
778-- | Get information about the specified process
779getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
781 let them = processNodeId pid in do
783 dest <- mkNode them us
784 sendCtrlMsg dest $ GetInfo pid
786 match (\(p :: ProcessInfo) -> return $ Just p)
787 , match (\(_ :: ProcessInfoNone) -> return Nothing)
789 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
790 mkNode them us = case them == us of
791 True -> return Nothing
792 _ -> return $ Just them
794--------------------------------------------------------------------------------
795-- Monitoring and linking --
796--------------------------------------------------------------------------------
798-- | Link to a remote process (asynchronous)
800-- When process A links to process B (that is, process A calls
801-- @link pidB@) then an asynchronous exception will be thrown to process A
802-- when process B terminates (normally or abnormally), or when process A gets
803-- disconnected from process B. Although it is /technically/ possible to catch
804-- these exceptions, chances are if you find yourself trying to do so you should
805-- probably be using 'monitor' rather than 'link'. In particular, code such as
807-- > link pidB -- Link to process B
808-- > expect -- Wait for a message from process B
809-- > unlink pidB -- Unlink again
811-- doesn't quite do what one might expect: if process B sends a message to
812-- process A, and /subsequently terminates/, then process A might or might not
813-- be terminated too, depending on whether the exception is thrown before or
814-- after the 'unlink' (i.e., this code has a race condition).
816-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
817-- call to 'link' has no effect.
819-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
820-- Linking makes no distinction between normal and abnormal termination of
821-- the remote process.
822link :: ProcessId -> Process ()
823link = sendCtrlMsg Nothing . Link . ProcessIdentifier
825-- | Monitor another process (asynchronous)
827-- When process A monitors process B (that is, process A calls
828-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
829-- when process B terminates (normally or abnormally), or when process A gets
830-- disconnected from process B. You receive this message like any other (using
831-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
832-- 'DiedDisconnect', etc.).
834-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
835-- multiple monitors are set up, multiple notifications will be delivered
836-- and monitors can be disabled individually using 'unmonitor'.
837monitor :: ProcessId -> Process MonitorRef
838monitor = monitor' . ProcessIdentifier
840-- | Establishes temporary monitoring of another process.
842-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
843-- of @code@. Note: although monitoring is no longer active when
844-- @withMonitor@ returns, there might still be unreceived monitor
845-- messages in the queue.
847withMonitor :: ProcessId -> Process a -> Process a
848withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
849 -- unmonitor blocks waiting for the response, so there's a possibility
850 -- that an exception might interrupt withMonitor before the unmonitor
851 -- has completed. I think that's better than making the unmonitor
856-- This is synchronous in the sense that once it returns you are guaranteed
857-- that no exception will be raised if the remote process dies. However, it is
858-- asynchronous in the sense that we do not wait for a response from the remote
860unlink :: ProcessId -> Process ()
863 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
867-- | Remove a node link
869-- This has the same synchronous/asynchronous nature as 'unlink'.
870unlinkNode :: NodeId -> Process ()
873 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
877-- | Remove a channel (send port) link
879-- This has the same synchronous/asynchronous nature as 'unlink'.
880unlinkPort :: SendPort a -> Process ()
882 unlinkPortAsync sport
883 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
889-- This has the same synchronous/asynchronous nature as 'unlink'.
890unmonitor :: MonitorRef -> Process ()
893 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
897--------------------------------------------------------------------------------
898-- Exception handling --
899--------------------------------------------------------------------------------
901-- | Lift 'Control.Exception.catch'
902catch :: Exception e => Process a -> (e -> Process a) -> Process a
905 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
907-- | Lift 'Control.Exception.try'
908try :: Exception e => Process a -> Process (Either e a)
911 liftIO $ Ex.try (runLocalProcess lproc p)
913-- | Lift 'Control.Exception.mask'
914mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
917 liftIO $ Ex.mask $ \restore ->
918 runLocalProcess lproc (p (liftRestore restore))
920 liftRestore :: (forall a. IO a -> IO a)
921 -> (forall a. Process a -> Process a)
922 liftRestore restoreIO = \p2 -> do
924 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
926-- | Lift 'Control.Exception.mask_'
927mask_ :: Process a -> Process a
930 liftIO $ Ex.mask_ $ runLocalProcess lproc p
932-- | Lift 'Control.Exception.onException'
933onException :: Process a -> Process b -> Process a
934onException p what = p `catch` \e -> do _ <- what
935 liftIO $ throwIO (e :: SomeException)
937-- | Lift 'Control.Exception.bracket'
938bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
939bracket before after thing =
940 mask $ \restore -> do
942 r <- restore (thing a) `onException` after a
946-- | Lift 'Control.Exception.bracket_'
947bracket_ :: Process a -> Process b -> Process c -> Process c
948bracket_ before after thing = bracket before (const after) (const thing)
950-- | Lift 'Control.Exception.finally'
951finally :: Process a -> Process b -> Process a
952finally a sequel = bracket_ (return ()) sequel a
954-- | You need this when using 'catches'
955data Handler a = forall e . Exception e => Handler (e -> Process a)
957instance Functor Handler where
958 fmap f (Handler h) = Handler (fmap f . h)
960-- | Lift 'Control.Exception.catches'
961catches :: Process a -> [Handler a] -> Process a
962catches proc handlers = proc `catch` catchesHandler handlers
964catchesHandler :: [Handler a] -> SomeException -> Process a
965catchesHandler handlers e = foldr tryHandler (throw e) handlers
966 where tryHandler (Handler handler) res
967 = case fromException e of
968 Just e' -> handler e'
971--------------------------------------------------------------------------------
973--------------------------------------------------------------------------------
975-- | Like 'expect' but with a timeout
976expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
977expectTimeout n = receiveTimeout n [match return]
979-- | Asynchronous version of 'spawn'
981-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
982spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
983spawnAsync nid proc = do
984 spawnRef <- getSpawnRef
985 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
988-- | Monitor a node (asynchronous)
989monitorNode :: NodeId -> Process MonitorRef
991 monitor' . NodeIdentifier
993-- | Monitor a typed channel (asynchronous)
994monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
995monitorPort (SendPort cid) =
996 monitor' (SendPortIdentifier cid)
998-- | Remove a monitor (asynchronous)
999unmonitorAsync :: MonitorRef -> Process ()
1001 sendCtrlMsg Nothing . Unmonitor
1003-- | Link to a node (asynchronous)
1004linkNode :: NodeId -> Process ()
1005linkNode = link' . NodeIdentifier
1007-- | Link to a channel (asynchronous)
1008linkPort :: SendPort a -> Process ()
1009linkPort (SendPort cid) =
1010 link' (SendPortIdentifier cid)
1012-- | Remove a process link (asynchronous)
1013unlinkAsync :: ProcessId -> Process ()
1015 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1017-- | Remove a node link (asynchronous)
1018unlinkNodeAsync :: NodeId -> Process ()
1020 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1022-- | Remove a channel (send port) link (asynchronous)
1023unlinkPortAsync :: SendPort a -> Process ()
1024unlinkPortAsync (SendPort cid) =
1025 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1027--------------------------------------------------------------------------------
1029--------------------------------------------------------------------------------
1033-- @say message@ sends a message (time, pid of the current process, message)
1034-- to the process registered as 'logger'. By default, this process simply
1035-- sends the string to 'stderr'. Individual Cloud Haskell backends might
1036-- replace this with a different logger process, however.
1037say :: String -> Process ()
1039 now <- liftIO getCurrentTime
1041 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
1043--------------------------------------------------------------------------------
1045--------------------------------------------------------------------------------
1047-- | Register a process with the local registry (synchronous). The name must not
1048-- already be registered. The process need not be on this node. A bad
1049-- registration will result in a 'ProcessRegistrationException'
1051-- The process to be registered does not have to be local itself.
1052register :: String -> ProcessId -> Process ()
1053register = registerImpl False
1055-- | Like 'register', but will replace an existing registration.
1056-- The name must already be registered.
1057reregister :: String -> ProcessId -> Process ()
1058reregister = registerImpl True
1060registerImpl :: Bool -> String -> ProcessId -> Process ()
1061registerImpl force label pid = do
1062 mynid <- getSelfNode
1063 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1064 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
1065 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
1068-- | Register a process with a remote registry (asynchronous).
1070-- The process to be registered does not have to live on the same remote node.
1071-- Reply wil come in the form of a 'RegisterReply' message
1073-- See comments in 'whereisRemoteAsync'
1074registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1075registerRemoteAsync nid label pid =
1076 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
1078reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1079reregisterRemoteAsync nid label pid =
1080 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1082-- | Remove a process from the local registry (asynchronous).
1083-- This version will wait until a response is gotten from the
1084-- management process. The name must already be registered.
1085unregister :: String -> Process ()
1086unregister label = do
1087 mynid <- getSelfNode
1088 sendCtrlMsg Nothing (Register label mynid Nothing False)
1089 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
1090 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
1093-- | Deal with the result from an attempted registration or unregistration
1094-- by throwing an exception if necessary
1095handleRegistrationReply :: String -> Bool -> Process ()
1096handleRegistrationReply label ok =
1098 liftIO $ throwIO $ ProcessRegistrationException label
1100-- | Remove a process from a remote registry (asynchronous).
1102-- Reply wil come in the form of a 'RegisterReply' message
1104-- See comments in 'whereisRemoteAsync'
1105unregisterRemoteAsync :: NodeId -> String -> Process ()
1106unregisterRemoteAsync nid label =
1107 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1109-- | Query the local process registry
1110whereis :: String -> Process (Maybe ProcessId)
1112 sendCtrlMsg Nothing (WhereIs label)
1113 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1114 (\(WhereIsReply _ mPid) -> return mPid)
1117-- | Query a remote process registry (asynchronous)
1119-- Reply will come in the form of a 'WhereIsReply' message.
1121-- There is currently no synchronous version of 'whereisRemoteAsync': if
1122-- you implement one yourself, be sure to take into account that the remote
1123-- node might die or get disconnect before it can respond (i.e. you should
1124-- use 'monitorNode' and take appropriate action when you receive a
1125-- 'NodeMonitorNotification').
1126whereisRemoteAsync :: NodeId -> String -> Process ()
1127whereisRemoteAsync nid label =
1128 sendCtrlMsg (Just nid) (WhereIs label)
1130-- | Named send to a process in the local registry (asynchronous)
1131nsend :: Serializable a => String -> a -> Process ()
1133 sendCtrlMsg Nothing (NamedSend label (createMessage msg))
1135-- | Named send to a process in the local registry (asynchronous).
1136-- This function makes /no/ attempt to serialize and (in the case when the
1137-- destination process resides on the same local node) therefore ensure that
1138-- the payload is fully evaluated before it is delivered.
1139unsafeNSend :: Serializable a => String -> a -> Process ()
1140unsafeNSend = Unsafe.nsend
1142-- | Named send to a process in a remote registry (asynchronous)
1143nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1144nsendRemote nid label msg =
1145 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1147--------------------------------------------------------------------------------
1149--------------------------------------------------------------------------------
1151-- | Resolve a static value
1152unStatic :: Typeable a => Static a -> Process a
1154 rtable <- remoteTable . processNode <$> ask
1155 case Static.unstatic rtable static of
1156 Left err -> fail $ "Could not resolve static value: " ++ err
1159-- | Resolve a closure
1160unClosure :: Typeable a => Closure a -> Process a
1161unClosure closure = do
1162 rtable <- remoteTable . processNode <$> ask
1163 case Static.unclosure rtable closure of
1164 Left err -> fail $ "Could not resolve closure: " ++ err
1167--------------------------------------------------------------------------------
1169--------------------------------------------------------------------------------
1171-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1172-- message passing. However, when network connections get disrupted this
1173-- illusion cannot always be maintained. Once a network connection breaks (even
1174-- temporarily) no further communication on that connection will be possible.
1175-- For example, if process A sends a message to process B, and A is then
1176-- notified (by monitor notification) that it got disconnected from B, A will
1177-- not be able to send any further messages to B, /unless/ A explicitly
1178-- indicates that it is acceptable to attempt to reconnect to B using the
1179-- Cloud Haskell 'reconnect' primitive.
1181-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1182-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1183-- then receives a monitor notification that its connection to B has been lost,
1184-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1185-- and m3 but not m2.
1187-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1188-- to attempt to reconnect on the next send/. In particular, if no further
1189-- communication attempts are made to B then A can use reconnect to clean up
1190-- its connection to B.
1191reconnect :: ProcessId -> Process ()
1194 node <- processNode <$> ask
1195 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1197-- | Reconnect to a sendport. See 'reconnect' for more information.
1198reconnectPort :: SendPort a -> Process ()
1199reconnectPort them = do
1201 node <- processNode <$> ask
1202 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1204--------------------------------------------------------------------------------
1205-- Auxiliary functions --
1206--------------------------------------------------------------------------------
1208sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1210 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1212sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1213sendChanLocal spId msg =
1214 -- we *must* fully serialize/encode the message here, because
1215 -- attempting to use `unsafeCoerce' in the node controller
1216 -- won't work since we know nothing about the required type
1217 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1219getMonitorRefFor :: Identifier -> Process MonitorRef
1220getMonitorRefFor ident = do
1222 liftIO $ modifyMVar (processState proc) $ \st -> do
1223 let counter = st ^. monitorCounter
1224 return ( monitorCounter ^: (+ 1) $ st
1225 , MonitorRef ident counter
1228getSpawnRef :: Process SpawnRef
1231 liftIO $ modifyMVar (processState proc) $ \st -> do
1232 let counter = st ^. spawnCounter
1233 return ( spawnCounter ^: (+ 1) $ st
1237-- | Monitor a process/node/channel
1238monitor' :: Identifier -> Process MonitorRef
1240 monitorRef <- getMonitorRefFor ident
1241 sendCtrlMsg Nothing $ Monitor monitorRef
1244-- | Link to a process/node/channel
1245link' :: Identifier -> Process ()
1246link' = 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