1{-# LANGUAGE BangPatterns #-}
3-- | Cloud Haskell primitives
5-- We define these in a separate module so that we don't have to rely on
6-- the closure combinators
7module Control.Distributed.Process.Internal.Primitives
17 -- * Unsafe messaging variants
21 -- * Advanced messaging
46 -- * Process management
48 , ProcessTerminationException(..)
54 -- keep the exception constructor hidden, so that handling exit
55 -- reasons /must/ take place via the 'catchExit' family of primitives
56 , ProcessExitException()
64 -- * Monitoring and linking
79 , reregisterRemoteAsync
80 , unregisterRemoteAsync
86 -- * Exception handling
109 -- * Internal Exports
113#if ! MIN_VERSION_base(4,6,0)
114import Prelude hiding (catch)
117import Data.Binary (decode)
118import Data.Time.Clock (getCurrentTime)
119import Data.Time.Format (formatTime)
120import System.Locale (defaultTimeLocale)
121import System.Timeout (timeout)
122import Control.Monad (when)
123import Control.Monad.Reader (ask)
124import Control.Monad.IO.Class (MonadIO, liftIO)
125import Control.Applicative ((<$>))
126import Control.Exception (Exception(..), throw, throwIO, SomeException)
127import qualified Control.Exception as Ex (catch, mask, try)
128import Control.Distributed.Process.Internal.StrictMVar
133import Control.Concurrent.STM
142import Control.Distributed.Process.Internal.CQueue
147import Control.Distributed.Process.Serializable (Serializable, fingerprint)
148import Data.Accessor ((^.), (^:), (^=))
149import Control.Distributed.Static
153import Data.Rank1Typeable (Typeable)
154import qualified Control.Distributed.Static as Static (unstatic, unclosure)
155import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
156import Control.Distributed.Process.Internal.Types
166 , NodeMonitorNotification(..)
176 , ProcessExitException(..)
179 , DidUnlinkProcess(..)
184 , ProcessRegistrationException(..)
186 , ProcessInfoNone(..)
190 , createUnencodedMessage
192 , ImplicitReconnect( NoImplicitReconnect)
197import Control.Distributed.Process.Internal.Messaging
204import Control.Distributed.Process.Management.Types
207import Control.Distributed.Process.Management.Trace.Types
210import Control.Distributed.Process.Internal.WeakTQueue
218--------------------------------------------------------------------------------
220--------------------------------------------------------------------------------
223send :: Serializable a => ProcessId -> a -> Process ()
224-- This requires a lookup on every send. If we want to avoid that we need to
225-- modify serializable to allow for stateful (IO) deserialization.
228 let us = processId proc
229 node = processNode proc
230 nodeId = localNodeId node
231 destNode = (processNodeId them) in do
232 case destNode == nodeId of
233 True -> sendLocal them msg
234 False -> liftIO $ sendMessage (processNode proc)
235 (ProcessIdentifier (processId proc))
236 (ProcessIdentifier them)
239 -- We do not fire the trace event until after the sending is complete;
240 -- In the remote case, 'sendMessage' can block in the networking stack.
241 liftIO $ traceEvent (localEventBus node)
242 (MxSent them us (createUnencodedMessage msg))
244-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
245-- and (in the case when the destination process resides on the same local
246-- node) therefore ensure that the payload is fully evaluated before it is
248unsafeSend :: Serializable a => ProcessId -> a -> Process ()
249unsafeSend = Unsafe.send
251-- | Wait for a message of a specific type
252expect :: forall a. Serializable a => Process a
253expect = receiveWait [match return]
255--------------------------------------------------------------------------------
257--------------------------------------------------------------------------------
259-- | Create a new typed channel
260newChan :: Serializable a => Process (SendPort a, ReceivePort a)
263 liftIO . modifyMVar (processState proc) $ \st -> do
264 let lcid = st ^. channelCounter
265 let cid = SendPortId { sendPortProcessId = processId proc
266 , sendPortLocalId = lcid
268 let sport = SendPort cid
269 chan <- liftIO newTQueueIO
270 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
271 let rport = ReceivePort $ readTQueue chan
272 let tch = TypedChannel chan'
273 return ( (channelCounter ^: (+ 1))
274 . (typedChannelWithId lcid ^= Just tch)
279 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
280 finalizer st lcid = modifyMVar_ st $
281 return . (typedChannelWithId lcid ^= Nothing)
283-- | Send a message on a typed channel
284sendChan :: Serializable a => SendPort a -> a -> Process ()
285sendChan (SendPort cid) msg = do
287 let node = localNodeId (processNode proc)
288 destNode = processNodeId (sendPortProcessId cid) in do
289 case destNode == node of
290 True -> sendChanLocal cid msg
292 liftIO $ sendBinary (processNode proc)
293 (ProcessIdentifier (processId proc))
294 (SendPortIdentifier cid)
298-- | Send a message on a typed channel. This function makes /no/ attempt to
299-- serialize and (in the case when the @ReceivePort@ resides on the same local
300-- node) therefore ensure that the payload is fully evaluated before it is
302unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
303unsafeSendChan = Unsafe.sendChan
305-- | Wait for a message on a typed channel
306receiveChan :: Serializable a => ReceivePort a -> Process a
307receiveChan = liftIO . atomically . receiveSTM
309-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
310-- non-blocking check for a message.
311receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
312receiveChanTimeout 0 ch = liftIO . atomically $
313 (Just <$> receiveSTM ch) `orElse` return Nothing
314receiveChanTimeout n ch = liftIO . timeout n . atomically $
317-- | Merge a list of typed channels.
319-- The result port is left-biased: if there are messages available on more
320-- than one port, the first available message is returned.
321mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
322mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
324-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
326mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
327mergePortsRR = \ps -> do
328 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
329 return $ ReceivePort (rr psVar)
333 rotate (x:xs) = xs ++ [x]
335 rr :: TVar [STM a] -> STM a
338 a <- foldr1 orElse ps
339 writeTVar psVar (rotate ps)
342--------------------------------------------------------------------------------
343-- Advanced messaging --
344--------------------------------------------------------------------------------
346-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
347newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
349-- | Test the matches in order against each message in the queue
350receiveWait :: [Match b] -> Process b
352 queue <- processQueue <$> ask
353 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
356-- | Like 'receiveWait' but with a timeout.
358-- If the timeout is zero do a non-blocking check for matching messages. A
359-- non-zero timeout is applied only when waiting for incoming messages (that is,
360-- /after/ we have checked the messages that are already in the mailbox).
361receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
362receiveTimeout t ms = do
363 queue <- processQueue <$> ask
364 let blockSpec = if t == 0 then NonBlocking else Timeout t
365 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
367 Nothing -> return Nothing
368 Just proc -> Just <$> proc
370-- | Match on a typed channel
371matchChan :: ReceivePort a -> (a -> Process b) -> Match b
372matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
374-- | Match on an arbitrary STM action.
376-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
377-- actions with checks against our process' mailbox and/or any typed channel
378-- @ReceivePort@s we may hold.
380-- This allows us to process multiple input streams /along with our mailbox/,
381-- in just the same way that 'matchChan' supports checking both the mailbox
382-- /and/ an arbitrary set of typed channels in one atomic transaction.
384-- Note there are no ordering guarnatees with respect to these disparate input
387matchSTM :: STM a -> (a -> Process b) -> Match b
388matchSTM stm fn = Match $ MatchChan (fmap fn stm)
390-- | Match against any message of the right type
391match :: forall a b. Serializable a => (a -> Process b) -> Match b
392match = matchIf (const True)
394-- | Match against any message of the right type that satisfies a predicate
395matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
396matchIf c p = Match $ MatchMsg $ \msg ->
397 case messageFingerprint msg == fingerprint (undefined :: a) of
400 (UnencodedMessage _ m) ->
401 let m' = unsafeCoerce m :: a in
405 (EncodedMessage _ _) ->
406 if (c decoded) then Just (p decoded) else Nothing
409 -- Make sure the value is fully decoded so that we don't hang to
410 -- bytestrings when the process calling 'matchIf' doesn't process
411 -- the values immediately
412 !decoded = decode (messageEncoding msg)
414-- | Match against any message, regardless of the underlying (contained) type
415matchMessage :: (Message -> Process Message) -> Match Message
416matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
418-- | Match against any message (regardless of underlying type) that satisfies a predicate
419matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
420matchMessageIf c p = Match $ MatchMsg $ \msg ->
425-- | Forward a raw 'Message' to the given 'ProcessId'.
426forward :: Message -> ProcessId -> Process ()
429 let node = processNode proc
431 nid = localNodeId node
432 destNode = (processNodeId them) in do
433 case destNode == nid of
434 True -> sendCtrlMsg Nothing (LocalSend them msg)
435 False -> liftIO $ sendPayload (processNode proc)
436 (ProcessIdentifier (processId proc))
437 (ProcessIdentifier them)
439 (messageToPayload msg)
440 -- We do not fire the trace event until after the sending is complete;
441 -- In the remote case, 'sendMessage' can block in the networking stack.
442 liftIO $ traceEvent (localEventBus node)
446-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
447-- 'Serializable' - like the datum they contain - but also note, deserialising
448-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
449-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
454-- send self (wrapMessage "blah")
455-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
456-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
457-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
460wrapMessage :: Serializable a => a -> Message
461wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
463-- | This is the /unsafe/ variant of 'wrapMessage'. See
464-- "Control.Distributed.Process.UnsafePrimitives" for details.
465unsafeWrapMessage :: Serializable a => a -> Message
466unsafeWrapMessage = Unsafe.wrapMessage
468-- [note Serializable UnencodedMessage]
469-- if we attempt to serialise an UnencodedMessage, it will be converted to an
470-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
471-- Primitives.hs for the gory details
473-- | Attempt to unwrap a raw 'Message'.
474-- If the type of the decoded message payload matches the expected type, the
475-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
478-- This expression, for example, will evaluate to @Nothing@
479-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
481-- Whereas this expression, will yield @Just "foo"@
482-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
484unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
486 case messageFingerprint msg == fingerprint (undefined :: a) of
487 False -> return Nothing :: m (Maybe a)
489 (UnencodedMessage _ ms) ->
490 let ms' = unsafeCoerce ms :: a
492 (EncodedMessage _ _) ->
493 return (Just (decoded))
495 decoded :: a -- note [decoding]
496 !decoded = decode (messageEncoding msg)
498-- | Attempt to handle a raw 'Message'.
499-- If the type of the message matches the type of the first argument to
500-- the supplied expression, then the message will be decoded and the expression
501-- evaluated against its value. If this runtime type checking fails however,
502-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
503-- evaluation proceeds, the resulting value with be wrapped with @Just@.
505-- Intended for use in `catchesExit` and `matchAny` primitives.
507handleMessage :: forall m a b. (Monad m, Serializable a)
508 => Message -> (a -> m b) -> m (Maybe b)
509handleMessage msg proc = handleMessageIf msg (const True) proc
511-- | Conditionally handle a raw 'Message'.
512-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
513-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
514-- for further information about runtime type checking.
516handleMessageIf :: forall m a b . (Monad m, Serializable a)
521handleMessageIf msg c proc = do
522 case messageFingerprint msg == fingerprint (undefined :: a) of
523 False -> return Nothing :: m (Maybe b)
525 (UnencodedMessage _ ms) ->
526 let ms' = unsafeCoerce ms :: a in
528 True -> do { r <- proc ms'; return (Just r) }
529 False -> return Nothing :: m (Maybe b)
530 (EncodedMessage _ _) ->
532 True -> do { r <- proc (decoded :: a); return (Just r) }
533 False -> return Nothing :: m (Maybe b)
535 decoded :: a -- note [decoding]
536 !decoded = decode (messageEncoding msg)
538-- | As 'handleMessage' but ignores result, which is useful if you don't
539-- care whether or not the handler succeeded.
541handleMessage_ :: forall m a . (Monad m, Serializable a)
542 => Message -> (a -> m ()) -> m ()
543handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
545-- | Conditional version of 'handleMessage_'.
547handleMessageIf_ :: forall m a . (Monad m, Serializable a)
552handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
554-- | Match against an arbitrary message. 'matchAny' removes the first available
555-- message from the process mailbox. To handle arbitrary /raw/ messages once
556-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
558matchAny :: forall b. (Message -> Process b) -> Match b
559matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
561-- | Match against an arbitrary message. Intended for use with 'handleMessage'
562-- and 'unwrapMessage', this function /only/ removes a message from the process
563-- mailbox, /if/ the supplied condition matches. The success (or failure) of
564-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
565-- here, i.e., if the condition evaluates to @True@ then the message will
566-- be removed from the process mailbox and decoded, but that does /not/
567-- guarantee that an expression passed to @handleMessage@ will pass the
568-- runtime type checks and therefore be evaluated.
570matchAnyIf :: forall a b. (Serializable a)
572 -> (Message -> Process b)
574matchAnyIf c p = Match $ MatchMsg $ \msg ->
575 case messageFingerprint msg == fingerprint (undefined :: a) of
576 True | check -> Just (p msg)
581 (EncodedMessage _ _) -> c decoded
582 (UnencodedMessage _ m') -> c (unsafeCoerce m')
584 decoded :: a -- note [decoding]
585 !decoded = decode (messageEncoding msg)
589For an EncodedMessage, we need to ensure the value is fully decoded so that
590we don't hang to bytestrings if the calling process doesn't evaluate
591immediately. For UnencodedMessage we know (because the fingerprint comparison
592succeeds) that unsafeCoerce will not fail.
595-- | Remove any message from the queue
596matchUnknown :: Process b -> Match b
597matchUnknown p = Match $ MatchMsg (const (Just p))
599-- | Receives messages and forwards them to @pid@ if @p msg == True@.
600delegate :: ProcessId -> (Message -> Bool) -> Process ()
603 matchAny (\m -> case (p m) of
604 True -> forward m pid
609-- | A straight relay that forwards all messages to the supplied pid.
610relay :: ProcessId -> Process ()
611relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
613-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
614-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
615-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
616-- fails, no action is taken and the proxy process will continue running.
617proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
621 next <- handleMessage m proc
623 Just True -> forward m pid
624 Just False -> return () -- explicitly ignored
625 Nothing -> return ()) -- un-routable
629--------------------------------------------------------------------------------
630-- Process management --
631--------------------------------------------------------------------------------
633-- | Thrown by 'terminate'
634data ProcessTerminationException = ProcessTerminationException
635 deriving (Show, Typeable)
637instance Exception ProcessTerminationException
639-- | Terminate immediately (throws a ProcessTerminationException)
640terminate :: Process a
641terminate = liftIO $ throwIO ProcessTerminationException
644-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
645die :: Serializable a => a -> Process b
647-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
648-- could be decoded by a handler passed to 'catchExit', re-thrown or even
649-- passed to another process via the tracing mechanism.
651 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
653-- | Forceful request to kill a process. Where 'exit' provides an exception
654-- that can be caught and handled, 'kill' throws an unexposed exception type
655-- which cannot be handled explicitly (by type).
656kill :: ProcessId -> String -> Process ()
657-- NOTE: We send the message to our local node controller, which will then
658-- forward it to a remote node controller (if applicable). Sending it directly
659-- to a remote node controller means that that the message may overtake a
660-- 'monitor' or 'link' request.
661kill them reason = sendCtrlMsg Nothing (Kill them reason)
663-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
664-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
665-- manner can be handled using the 'catchExit' family of functions.
666exit :: Serializable a => ProcessId -> a -> Process ()
667-- NOTE: We send the message to our local node controller, which will then
668-- forward it to a remote node controller (if applicable). Sending it directly
669-- to a remote node controller means that that the message may overtake a
670-- 'monitor' or 'link' request.
671exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
673-- | Catches 'ProcessExitException'. The handler will not be applied unless its
674-- type matches the encoded data stored in the exception (see the /reason/
675-- argument given to the 'exit' primitive). If the handler cannot be applied,
676-- the exception will be re-thrown.
678-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
679-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
680catchExit :: forall a b . (Show a, Serializable a)
682 -> (ProcessId -> a -> Process b)
684catchExit act exitHandler = catch act handleExit
686 handleExit ex@(ProcessExitException from msg) =
687 if messageFingerprint msg == fingerprint (undefined :: a)
688 then exitHandler from decoded
689 else liftIO $ throwIO ex
692 -- Make sure the value is fully decoded so that we don't hang to
693 -- bytestrings if the caller doesn't use the value immediately
694 !decoded = decode (messageEncoding msg)
696-- | Lift 'Control.Exception.catches' (almost).
698-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
699-- message, a handler must accept inputs of the expected type. In order to
700-- handle a list of potentially different handlers (and therefore input types),
701-- a handler passed to 'catchesExit' must accept 'Message' and return
702-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
704-- See 'maybeHandleMessage' and 'Message' for more details.
705catchesExit :: Process b
706 -> [(ProcessId -> Message -> (Process (Maybe b)))]
708catchesExit act handlers = catch act ((flip handleExit) handlers)
710 handleExit :: ProcessExitException
711 -> [(ProcessId -> Message -> Process (Maybe b))]
713 handleExit ex [] = liftIO $ throwIO ex
714 handleExit ex@(ProcessExitException from msg) (h:hs) = do
717 Nothing -> handleExit ex hs
720-- | Our own process ID
721getSelfPid :: Process ProcessId
722getSelfPid = processId <$> ask
724-- | Get the node ID of our local node
725getSelfNode :: Process NodeId
726getSelfNode = localNodeId . processNode <$> ask
729-- | Get statistics about the specified node
730getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
732 selfNode <- getSelfNode
734 then Right `fmap` getLocalNodeStats -- optimisation
735 else getNodeStatsRemote nid
737 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
738 getNodeStatsRemote nid = do
739 sendCtrlMsg (Just nid) $ GetNodeStats nid
740 bracket (monitorNode nid) unmonitor $ \mRef ->
741 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
742 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
743 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
746-- | Get statistics about our local node
747getLocalNodeStats :: Process NodeStats
748getLocalNodeStats = do
750 sendCtrlMsg Nothing $ GetNodeStats self
751 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
753-- | Get information about the specified process
754getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
756 let them = processNodeId pid in do
758 dest <- mkNode them us
759 sendCtrlMsg dest $ GetInfo pid
761 match (\(p :: ProcessInfo) -> return $ Just p)
762 , match (\(_ :: ProcessInfoNone) -> return Nothing)
764 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
765 mkNode them us = case them == us of
766 True -> return Nothing
767 _ -> return $ Just them
769--------------------------------------------------------------------------------
770-- Monitoring and linking --
771--------------------------------------------------------------------------------
773-- | Link to a remote process (asynchronous)
775-- When process A links to process B (that is, process A calls
776-- @link pidB@) then an asynchronous exception will be thrown to process A
777-- when process B terminates (normally or abnormally), or when process A gets
778-- disconnected from process B. Although it is /technically/ possible to catch
779-- these exceptions, chances are if you find yourself trying to do so you should
780-- probably be using 'monitor' rather than 'link'. In particular, code such as
782-- > link pidB -- Link to process B
783-- > expect -- Wait for a message from process B
784-- > unlink pidB -- Unlink again
786-- doesn't quite do what one might expect: if process B sends a message to
787-- process A, and /subsequently terminates/, then process A might or might not
788-- be terminated too, depending on whether the exception is thrown before or
789-- after the 'unlink' (i.e., this code has a race condition).
791-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
792-- call to 'link' has no effect.
794-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
795-- Linking makes no distinction between normal and abnormal termination of
796-- the remote process.
797link :: ProcessId -> Process ()
798link = sendCtrlMsg Nothing . Link . ProcessIdentifier
800-- | Monitor another process (asynchronous)
802-- When process A monitors process B (that is, process A calls
803-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
804-- when process B terminates (normally or abnormally), or when process A gets
805-- disconnected from process B. You receive this message like any other (using
806-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
807-- 'DiedDisconnect', etc.).
809-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
810-- multiple monitors are set up, multiple notifications will be delivered
811-- and monitors can be disabled individually using 'unmonitor'.
812monitor :: ProcessId -> Process MonitorRef
813monitor = monitor' . ProcessIdentifier
815-- | Establishes temporary monitoring of another process.
817-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
818-- of @code@. Note: although monitoring is no longer active when
819-- @withMonitor@ returns, there might still be unreceived monitor
820-- messages in the queue.
822withMonitor :: ProcessId -> Process a -> Process a
823withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
824 -- unmonitor blocks waiting for the response, so there's a possibility
825 -- that an exception might interrupt withMonitor before the unmonitor
826 -- has completed. I think that's better than making the unmonitor
831-- This is synchronous in the sense that once it returns you are guaranteed
832-- that no exception will be raised if the remote process dies. However, it is
833-- asynchronous in the sense that we do not wait for a response from the remote
835unlink :: ProcessId -> Process ()
838 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
842-- | Remove a node link
844-- This has the same synchronous/asynchronous nature as 'unlink'.
845unlinkNode :: NodeId -> Process ()
848 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
852-- | Remove a channel (send port) link
854-- This has the same synchronous/asynchronous nature as 'unlink'.
855unlinkPort :: SendPort a -> Process ()
857 unlinkPortAsync sport
858 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
864-- This has the same synchronous/asynchronous nature as 'unlink'.
865unmonitor :: MonitorRef -> Process ()
868 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
872--------------------------------------------------------------------------------
873-- Exception handling --
874--------------------------------------------------------------------------------
876-- | Lift 'Control.Exception.catch'
877catch :: Exception e => Process a -> (e -> Process a) -> Process a
880 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
882-- | Lift 'Control.Exception.try'
883try :: Exception e => Process a -> Process (Either e a)
886 liftIO $ Ex.try (runLocalProcess lproc p)
888-- | Lift 'Control.Exception.mask'
889mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
892 liftIO $ Ex.mask $ \restore ->
893 runLocalProcess lproc (p (liftRestore restore))
895 liftRestore :: (forall a. IO a -> IO a)
896 -> (forall a. Process a -> Process a)
897 liftRestore restoreIO = \p2 -> do
899 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
901-- | Lift 'Control.Exception.onException'
902onException :: Process a -> Process b -> Process a
903onException p what = p `catch` \e -> do _ <- what
904 liftIO $ throwIO (e :: SomeException)
906-- | Lift 'Control.Exception.bracket'
907bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
908bracket before after thing =
909 mask $ \restore -> do
911 r <- restore (thing a) `onException` after a
915-- | Lift 'Control.Exception.bracket_'
916bracket_ :: Process a -> Process b -> Process c -> Process c
917bracket_ before after thing = bracket before (const after) (const thing)
919-- | Lift 'Control.Exception.finally'
920finally :: Process a -> Process b -> Process a
921finally a sequel = bracket_ (return ()) sequel a
923-- | You need this when using 'catches'
924data Handler a = forall e . Exception e => Handler (e -> Process a)
926instance Functor Handler where
927 fmap f (Handler h) = Handler (fmap f . h)
929-- | Lift 'Control.Exception.catches'
930catches :: Process a -> [Handler a] -> Process a
931catches proc handlers = proc `catch` catchesHandler handlers
933catchesHandler :: [Handler a] -> SomeException -> Process a
934catchesHandler handlers e = foldr tryHandler (throw e) handlers
935 where tryHandler (Handler handler) res
936 = case fromException e of
937 Just e' -> handler e'
940--------------------------------------------------------------------------------
942--------------------------------------------------------------------------------
944-- | Like 'expect' but with a timeout
945expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
946expectTimeout n = receiveTimeout n [match return]
948-- | Asynchronous version of 'spawn'
950-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
951spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
952spawnAsync nid proc = do
953 spawnRef <- getSpawnRef
954 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
957-- | Monitor a node (asynchronous)
958monitorNode :: NodeId -> Process MonitorRef
960 monitor' . NodeIdentifier
962-- | Monitor a typed channel (asynchronous)
963monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
964monitorPort (SendPort cid) =
965 monitor' (SendPortIdentifier cid)
967-- | Remove a monitor (asynchronous)
968unmonitorAsync :: MonitorRef -> Process ()
970 sendCtrlMsg Nothing . Unmonitor
972-- | Link to a node (asynchronous)
973linkNode :: NodeId -> Process ()
974linkNode = link' . NodeIdentifier
976-- | Link to a channel (asynchronous)
977linkPort :: SendPort a -> Process ()
978linkPort (SendPort cid) =
979 link' (SendPortIdentifier cid)
981-- | Remove a process link (asynchronous)
982unlinkAsync :: ProcessId -> Process ()
984 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
986-- | Remove a node link (asynchronous)
987unlinkNodeAsync :: NodeId -> Process ()
989 sendCtrlMsg Nothing . Unlink . NodeIdentifier
991-- | Remove a channel (send port) link (asynchronous)
992unlinkPortAsync :: SendPort a -> Process ()
993unlinkPortAsync (SendPort cid) =
994 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
996--------------------------------------------------------------------------------
998--------------------------------------------------------------------------------
1002-- @say message@ sends a message (time, pid of the current process, message)
1003-- to the process registered as 'logger'. By default, this process simply
1004-- sends the string to 'stderr'. Individual Cloud Haskell backends might
1005-- replace this with a different logger process, however.
1006say :: String -> Process ()
1008 now <- liftIO getCurrentTime
1010 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
1012--------------------------------------------------------------------------------
1014--------------------------------------------------------------------------------
1016-- | Register a process with the local registry (asynchronous).
1017-- This version will wait until a response is gotten from the
1018-- management process. The name must not already be registered.
1019-- The process need not be on this node.
1020-- A bad registration will result in a 'ProcessRegistrationException'
1022-- The process to be registered does not have to be local itself.
1023register :: String -> ProcessId -> Process ()
1024register = registerImpl False
1026-- | Like 'register', but will replace an existing registration.
1027-- The name must already be registered.
1028reregister :: String -> ProcessId -> Process ()
1029reregister = registerImpl True
1031registerImpl :: Bool -> String -> ProcessId -> Process ()
1032registerImpl force label pid = do
1033 mynid <- getSelfNode
1034 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1035 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
1036 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
1039-- | Register a process with a remote registry (asynchronous).
1041-- The process to be registered does not have to live on the same remote node.
1042-- Reply wil come in the form of a 'RegisterReply' message
1044-- See comments in 'whereisRemoteAsync'
1045registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1046registerRemoteAsync nid label pid =
1047 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
1049reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1050reregisterRemoteAsync nid label pid =
1051 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1053-- | Remove a process from the local registry (asynchronous).
1054-- This version will wait until a response is gotten from the
1055-- management process. The name must already be registered.
1056unregister :: String -> Process ()
1057unregister label = do
1058 mynid <- getSelfNode
1059 sendCtrlMsg Nothing (Register label mynid Nothing False)
1060 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
1061 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
1064-- | Deal with the result from an attempted registration or unregistration
1065-- by throwing an exception if necessary
1066handleRegistrationReply :: String -> Bool -> Process ()
1067handleRegistrationReply label ok =
1069 liftIO $ throwIO $ ProcessRegistrationException label
1071-- | Remove a process from a remote registry (asynchronous).
1073-- Reply wil come in the form of a 'RegisterReply' message
1075-- See comments in 'whereisRemoteAsync'
1076unregisterRemoteAsync :: NodeId -> String -> Process ()
1077unregisterRemoteAsync nid label =
1078 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1080-- | Query the local process registry
1081whereis :: String -> Process (Maybe ProcessId)
1083 sendCtrlMsg Nothing (WhereIs label)
1084 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1085 (\(WhereIsReply _ mPid) -> return mPid)
1088-- | Query a remote process registry (asynchronous)
1090-- Reply will come in the form of a 'WhereIsReply' message.
1092-- There is currently no synchronous version of 'whereisRemoteAsync': if
1093-- you implement one yourself, be sure to take into account that the remote
1094-- node might die or get disconnect before it can respond (i.e. you should
1095-- use 'monitorNode' and take appropriate action when you receive a
1096-- 'NodeMonitorNotification').
1097whereisRemoteAsync :: NodeId -> String -> Process ()
1098whereisRemoteAsync nid label =
1099 sendCtrlMsg (Just nid) (WhereIs label)
1101-- | Named send to a process in the local registry (asynchronous)
1102nsend :: Serializable a => String -> a -> Process ()
1104 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1106-- | Named send to a process in the local registry (asynchronous).
1107-- This function makes /no/ attempt to serialize and (in the case when the
1108-- destination process resides on the same local node) therefore ensure that
1109-- the payload is fully evaluated before it is delivered.
1110unsafeNSend :: Serializable a => String -> a -> Process ()
1111unsafeNSend = Unsafe.nsend
1113-- | Named send to a process in a remote registry (asynchronous)
1114nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1115nsendRemote nid label msg =
1116 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1118--------------------------------------------------------------------------------
1120--------------------------------------------------------------------------------
1122-- | Resolve a static value
1123unStatic :: Typeable a => Static a -> Process a
1125 rtable <- remoteTable . processNode <$> ask
1126 case Static.unstatic rtable static of
1127 Left err -> fail $ "Could not resolve static value: " ++ err
1130-- | Resolve a closure
1131unClosure :: Typeable a => Closure a -> Process a
1132unClosure closure = do
1133 rtable <- remoteTable . processNode <$> ask
1134 case Static.unclosure rtable closure of
1135 Left err -> fail $ "Could not resolve closure: " ++ err
1138--------------------------------------------------------------------------------
1140--------------------------------------------------------------------------------
1142-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1143-- message passing. However, when network connections get disrupted this
1144-- illusion cannot always be maintained. Once a network connection breaks (even
1145-- temporarily) no further communication on that connection will be possible.
1146-- For example, if process A sends a message to process B, and A is then
1147-- notified (by monitor notification) that it got disconnected from B, A will
1148-- not be able to send any further messages to B, /unless/ A explicitly
1149-- indicates that it is acceptable to attempt to reconnect to B using the
1150-- Cloud Haskell 'reconnect' primitive.
1152-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1153-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1154-- then receives a monitor notification that its connection to B has been lost,
1155-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1156-- and m3 but not m2.
1158-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1159-- to attempt to reconnect on the next send/. In particular, if no further
1160-- communication attempts are made to B then A can use reconnect to clean up
1161-- its connection to B.
1162reconnect :: ProcessId -> Process ()
1165 node <- processNode <$> ask
1166 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1168-- | Reconnect to a sendport. See 'reconnect' for more information.
1169reconnectPort :: SendPort a -> Process ()
1170reconnectPort them = do
1172 node <- processNode <$> ask
1173 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1175--------------------------------------------------------------------------------
1176-- Auxiliary functions --
1177--------------------------------------------------------------------------------
1179sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1181 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1183sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1184sendChanLocal spId msg =
1185 -- we *must* fully serialize/encode the message here, because
1186 -- attempting to use `unsafeCoerce' in the node controller
1187 -- won't work since we know nothing about the required type
1188 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1190getMonitorRefFor :: Identifier -> Process MonitorRef
1191getMonitorRefFor ident = do
1193 liftIO $ modifyMVar (processState proc) $ \st -> do
1194 let counter = st ^. monitorCounter
1195 return ( monitorCounter ^: (+ 1) $ st
1196 , MonitorRef ident counter
1199getSpawnRef :: Process SpawnRef
1202 liftIO $ modifyMVar (processState proc) $ \st -> do
1203 let counter = st ^. spawnCounter
1204 return ( spawnCounter ^: (+ 1) $ st
1208-- | Monitor a process/node/channel
1209monitor' :: Identifier -> Process MonitorRef
1211 monitorRef <- getMonitorRefFor ident
1212 sendCtrlMsg Nothing $ Monitor monitorRef
1215-- | Link to a process/node/channel
1216link' :: Identifier -> Process ()
1217link' = 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