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
53 -- * Process management
55 , ProcessTerminationException(..)
61 -- keep the exception constructor hidden, so that handling exit
62 -- reasons /must/ take place via the 'catchExit' family of primitives
63 , ProcessExitException()
71 -- * Monitoring and linking
86 , reregisterRemoteAsync
87 , unregisterRemoteAsync
93 -- * Exception handling
117 -- * Internal Exports
121#if ! MIN_VERSION_base(4,6,0)
122import Prelude hiding (catch)
125import Data.Binary (decode)
126import Data.Time.Clock (getCurrentTime)
127import Data.Time.Format (formatTime)
128#if MIN_VERSION_time(1,5,0)
129import Data.Time.Format (defaultTimeLocale)
131import System.Locale (defaultTimeLocale)
133import System.Timeout (timeout)
134import Control.Monad (when)
135import Control.Monad.Reader (ask)
136import Control.Monad.IO.Class (liftIO)
137import Control.Applicative ((<$>))
138import Control.Exception (Exception(..), throw, throwIO, SomeException)
139import qualified Control.Exception as Ex (catch, mask, mask_, try)
140import Control.Distributed.Process.Internal.StrictMVar
145import Control.Concurrent.STM
154import Control.Distributed.Process.Internal.CQueue
159import Control.Distributed.Process.Serializable (Serializable, fingerprint)
160import Data.Accessor ((^.), (^:), (^=))
161import Control.Distributed.Static
165import Data.Rank1Typeable (Typeable)
166import qualified Control.Distributed.Static as Static (unstatic, unclosure)
167import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
168import Control.Distributed.Process.Internal.Types
178 , NodeMonitorNotification(..)
188 , ProcessExitException(..)
191 , DidUnlinkProcess(..)
196 , ProcessRegistrationException(..)
198 , ProcessInfoNone(..)
202 , createUnencodedMessage
204 , ImplicitReconnect( NoImplicitReconnect)
209import Control.Distributed.Process.Internal.Messaging
216import Control.Distributed.Process.Management.Internal.Types
219import Control.Distributed.Process.Management.Internal.Trace.Types
222import Control.Distributed.Process.Internal.WeakTQueue
230--------------------------------------------------------------------------------
232--------------------------------------------------------------------------------
235send :: Serializable a => ProcessId -> a -> Process ()
236-- This requires a lookup on every send. If we want to avoid that we need to
237-- modify serializable to allow for stateful (IO) deserialization.
240 let us = processId proc
241 node = processNode proc
242 nodeId = localNodeId node
243 destNode = (processNodeId them) in do
244 case destNode == nodeId of
245 True -> sendLocal them msg
246 False -> liftIO $ sendMessage (processNode proc)
247 (ProcessIdentifier (processId proc))
248 (ProcessIdentifier them)
251 -- We do not fire the trace event until after the sending is complete;
252 -- In the remote case, 'sendMessage' can block in the networking stack.
253 liftIO $ traceEvent (localEventBus node)
254 (MxSent them us (createUnencodedMessage msg))
256-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
257-- and (in the case when the destination process resides on the same local
258-- node) therefore ensure that the payload is fully evaluated before it is
260unsafeSend :: Serializable a => ProcessId -> a -> Process ()
261unsafeSend = Unsafe.send
263-- | Send a message unreliably.
265-- Unlike 'send', this function is insensitive to 'reconnect'. It will
266-- try to send the message regardless of the history of connection failures
269-- Message passing with 'usend' is ordered for a given sender and receiver
270-- if the messages arrive at all.
272usend :: Serializable a => ProcessId -> a -> Process ()
275 let there = processNodeId them
277 then sendLocal them msg
278 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
281-- | Wait for a message of a specific type
282expect :: forall a. Serializable a => Process a
283expect = receiveWait [match return]
285--------------------------------------------------------------------------------
287--------------------------------------------------------------------------------
289-- | Create a new typed channel
290newChan :: Serializable a => Process (SendPort a, ReceivePort a)
293 liftIO . modifyMVar (processState proc) $ \st -> do
294 let lcid = st ^. channelCounter
295 let cid = SendPortId { sendPortProcessId = processId proc
296 , sendPortLocalId = lcid
298 let sport = SendPort cid
299 chan <- liftIO newTQueueIO
300 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
301 let rport = ReceivePort $ readTQueue chan
302 let tch = TypedChannel chan'
303 return ( (channelCounter ^: (+ 1))
304 . (typedChannelWithId lcid ^= Just tch)
309 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
310 finalizer st lcid = modifyMVar_ st $
311 return . (typedChannelWithId lcid ^= Nothing)
313-- | Send a message on a typed channel
314sendChan :: Serializable a => SendPort a -> a -> Process ()
315sendChan (SendPort cid) msg = do
317 let node = localNodeId (processNode proc)
318 destNode = processNodeId (sendPortProcessId cid) in do
319 case destNode == node of
320 True -> sendChanLocal cid msg
322 liftIO $ sendBinary (processNode proc)
323 (ProcessIdentifier (processId proc))
324 (SendPortIdentifier cid)
328-- | Send a message on a typed channel. This function makes /no/ attempt to
329-- serialize and (in the case when the @ReceivePort@ resides on the same local
330-- node) therefore ensure that the payload is fully evaluated before it is
332unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
333unsafeSendChan = Unsafe.sendChan
335-- | Wait for a message on a typed channel
336receiveChan :: Serializable a => ReceivePort a -> Process a
337receiveChan = liftIO . atomically . receiveSTM
339-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
340-- non-blocking check for a message.
341receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
342receiveChanTimeout 0 ch = liftIO . atomically $
343 (Just <$> receiveSTM ch) `orElse` return Nothing
344receiveChanTimeout n ch = liftIO . timeout n . atomically $
347-- | Merge a list of typed channels.
349-- The result port is left-biased: if there are messages available on more
350-- than one port, the first available message is returned.
351mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
352mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
354-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
356mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
357mergePortsRR = \ps -> do
358 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
359 return $ ReceivePort (rr psVar)
363 rotate (x:xs) = xs ++ [x]
365 rr :: TVar [STM a] -> STM a
368 a <- foldr1 orElse ps
369 writeTVar psVar (rotate ps)
372--------------------------------------------------------------------------------
373-- Advanced messaging --
374--------------------------------------------------------------------------------
376-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
377newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
380-- | Test the matches in order against each message in the queue
381receiveWait :: [Match b] -> Process b
383 queue <- processQueue <$> ask
384 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
387-- | Like 'receiveWait' but with a timeout.
389-- If the timeout is zero do a non-blocking check for matching messages. A
390-- non-zero timeout is applied only when waiting for incoming messages (that is,
391-- /after/ we have checked the messages that are already in the mailbox).
392receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
393receiveTimeout t ms = do
394 queue <- processQueue <$> ask
395 let blockSpec = if t == 0 then NonBlocking else Timeout t
396 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
398 Nothing -> return Nothing
399 Just proc -> Just <$> proc
401-- | Match on a typed channel
402matchChan :: ReceivePort a -> (a -> Process b) -> Match b
403matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
405-- | Match on an arbitrary STM action.
407-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
408-- actions with checks against our process' mailbox and/or any typed channel
409-- @ReceivePort@s we may hold.
411-- This allows us to process multiple input streams /along with our mailbox/,
412-- in just the same way that 'matchChan' supports checking both the mailbox
413-- /and/ an arbitrary set of typed channels in one atomic transaction.
415-- Note there are no ordering guarnatees with respect to these disparate input
418matchSTM :: STM a -> (a -> Process b) -> Match b
419matchSTM stm fn = Match $ MatchChan (fmap fn stm)
421-- | Match against any message of the right type
422match :: forall a b. Serializable a => (a -> Process b) -> Match b
423match = matchIf (const True)
425-- | Match against any message of the right type that satisfies a predicate
426matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
427matchIf c p = Match $ MatchMsg $ \msg ->
428 case messageFingerprint msg == fingerprint (undefined :: a) of
431 (UnencodedMessage _ m) ->
432 let m' = unsafeCoerce m :: a in
436 (EncodedMessage _ _) ->
437 if (c decoded) then Just (p decoded) else Nothing
440 -- Make sure the value is fully decoded so that we don't hang to
441 -- bytestrings when the process calling 'matchIf' doesn't process
442 -- the values immediately
443 !decoded = decode (messageEncoding msg)
445-- | Match against any message, regardless of the underlying (contained) type
446matchMessage :: (Message -> Process Message) -> Match Message
447matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
449-- | Match against any message (regardless of underlying type) that satisfies a predicate
450matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
451matchMessageIf c p = Match $ MatchMsg $ \msg ->
456-- | Forward a raw 'Message' to the given 'ProcessId'.
457forward :: Message -> ProcessId -> Process ()
460 let node = processNode proc
462 nid = localNodeId node
463 destNode = (processNodeId them) in do
464 case destNode == nid of
465 True -> sendCtrlMsg Nothing (LocalSend them msg)
466 False -> liftIO $ sendPayload (processNode proc)
467 (ProcessIdentifier (processId proc))
468 (ProcessIdentifier them)
470 (messageToPayload msg)
471 -- We do not fire the trace event until after the sending is complete;
472 -- In the remote case, 'sendMessage' can block in the networking stack.
473 liftIO $ traceEvent (localEventBus node)
477-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
478-- 'Serializable' - like the datum they contain - but also note, deserialising
479-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
480-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
485-- send self (wrapMessage "blah")
486-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
487-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
488-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
491wrapMessage :: Serializable a => a -> Message
492wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
494-- | This is the /unsafe/ variant of 'wrapMessage'. See
495-- "Control.Distributed.Process.UnsafePrimitives" for details.
496unsafeWrapMessage :: Serializable a => a -> Message
497unsafeWrapMessage = Unsafe.wrapMessage
499-- [note Serializable UnencodedMessage]
500-- if we attempt to serialise an UnencodedMessage, it will be converted to an
501-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
502-- Primitives.hs for the gory details
504-- | Attempt to unwrap a raw 'Message'.
505-- If the type of the decoded message payload matches the expected type, the
506-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
509-- This expression, for example, will evaluate to @Nothing@
510-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
512-- Whereas this expression, will yield @Just "foo"@
513-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
515unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
517 case messageFingerprint msg == fingerprint (undefined :: a) of
518 False -> return Nothing :: m (Maybe a)
520 (UnencodedMessage _ ms) ->
521 let ms' = unsafeCoerce ms :: a
523 (EncodedMessage _ _) ->
524 return (Just (decoded))
526 decoded :: a -- note [decoding]
527 !decoded = decode (messageEncoding msg)
529-- | Attempt to handle a raw 'Message'.
530-- If the type of the message matches the type of the first argument to
531-- the supplied expression, then the message will be decoded and the expression
532-- evaluated against its value. If this runtime type checking fails however,
533-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
534-- evaluation proceeds, the resulting value with be wrapped with @Just@.
536-- Intended for use in `catchesExit` and `matchAny` primitives.
538handleMessage :: forall m a b. (Monad m, Serializable a)
539 => Message -> (a -> m b) -> m (Maybe b)
540handleMessage msg proc = handleMessageIf msg (const True) proc
542-- | Conditionally handle a raw 'Message'.
543-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
544-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
545-- for further information about runtime type checking.
547handleMessageIf :: forall m a b . (Monad m, Serializable a)
552handleMessageIf msg c proc = do
553 case messageFingerprint msg == fingerprint (undefined :: a) of
554 False -> return Nothing :: m (Maybe b)
556 (UnencodedMessage _ ms) ->
557 let ms' = unsafeCoerce ms :: a in
559 True -> do { r <- proc ms'; return (Just r) }
560 False -> return Nothing :: m (Maybe b)
561 (EncodedMessage _ _) ->
563 True -> do { r <- proc (decoded :: a); return (Just r) }
564 False -> return Nothing :: m (Maybe b)
566 decoded :: a -- note [decoding]
567 !decoded = decode (messageEncoding msg)
569-- | As 'handleMessage' but ignores result, which is useful if you don't
570-- care whether or not the handler succeeded.
572handleMessage_ :: forall m a . (Monad m, Serializable a)
573 => Message -> (a -> m ()) -> m ()
574handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
576-- | Conditional version of 'handleMessage_'.
578handleMessageIf_ :: forall m a . (Monad m, Serializable a)
583handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
585-- | Match against an arbitrary message. 'matchAny' removes the first available
586-- message from the process mailbox. To handle arbitrary /raw/ messages once
587-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
589matchAny :: forall b. (Message -> Process b) -> Match b
590matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
592-- | Match against an arbitrary message. Intended for use with 'handleMessage'
593-- and 'unwrapMessage', this function /only/ removes a message from the process
594-- mailbox, /if/ the supplied condition matches. The success (or failure) of
595-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
596-- here, i.e., if the condition evaluates to @True@ then the message will
597-- be removed from the process mailbox and decoded, but that does /not/
598-- guarantee that an expression passed to @handleMessage@ will pass the
599-- runtime type checks and therefore be evaluated.
601matchAnyIf :: forall a b. (Serializable a)
603 -> (Message -> Process b)
605matchAnyIf c p = Match $ MatchMsg $ \msg ->
606 case messageFingerprint msg == fingerprint (undefined :: a) of
607 True | check -> Just (p msg)
612 (EncodedMessage _ _) -> c decoded
613 (UnencodedMessage _ m') -> c (unsafeCoerce m')
615 decoded :: a -- note [decoding]
616 !decoded = decode (messageEncoding msg)
620For an EncodedMessage, we need to ensure the value is fully decoded so that
621we don't hang to bytestrings if the calling process doesn't evaluate
622immediately. For UnencodedMessage we know (because the fingerprint comparison
623succeeds) that unsafeCoerce will not fail.
626-- | Remove any message from the queue
627matchUnknown :: Process b -> Match b
628matchUnknown p = Match $ MatchMsg (const (Just p))
630-- | Receives messages and forwards them to @pid@ if @p msg == True@.
631delegate :: ProcessId -> (Message -> Bool) -> Process ()
634 matchAny (\m -> case (p m) of
635 True -> forward m pid
640-- | A straight relay that forwards all messages to the supplied pid.
641relay :: ProcessId -> Process ()
642relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
644-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
645-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
646-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
647-- fails, no action is taken and the proxy process will continue running.
648proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
652 next <- handleMessage m proc
654 Just True -> forward m pid
655 Just False -> return () -- explicitly ignored
656 Nothing -> return ()) -- un-routable
660--------------------------------------------------------------------------------
661-- Process management --
662--------------------------------------------------------------------------------
664-- | Thrown by 'terminate'
665data ProcessTerminationException = ProcessTerminationException
666 deriving (Show, Typeable)
668instance Exception ProcessTerminationException
670-- | Terminate immediately (throws a ProcessTerminationException)
671terminate :: Process a
672terminate = liftIO $ throwIO ProcessTerminationException
675-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
676die :: Serializable a => a -> Process b
678-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
679-- could be decoded by a handler passed to 'catchExit', re-thrown or even
680-- passed to another process via the tracing mechanism.
682 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
684-- | Forceful request to kill a process. Where 'exit' provides an exception
685-- that can be caught and handled, 'kill' throws an unexposed exception type
686-- which cannot be handled explicitly (by type).
687kill :: ProcessId -> String -> Process ()
688-- NOTE: We send the message to our local node controller, which will then
689-- forward it to a remote node controller (if applicable). Sending it directly
690-- to a remote node controller means that that the message may overtake a
691-- 'monitor' or 'link' request.
692kill them reason = sendCtrlMsg Nothing (Kill them reason)
694-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
695-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
696-- manner can be handled using the 'catchExit' family of functions.
697exit :: Serializable a => ProcessId -> a -> Process ()
698-- NOTE: We send the message to our local node controller, which will then
699-- forward it to a remote node controller (if applicable). Sending it directly
700-- to a remote node controller means that that the message may overtake a
701-- 'monitor' or 'link' request.
702exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
704-- | Catches 'ProcessExitException'. The handler will not be applied unless its
705-- type matches the encoded data stored in the exception (see the /reason/
706-- argument given to the 'exit' primitive). If the handler cannot be applied,
707-- the exception will be re-thrown.
709-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
710-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
711catchExit :: forall a b . (Show a, Serializable a)
713 -> (ProcessId -> a -> Process b)
715catchExit act exitHandler = catch act handleExit
717 handleExit ex@(ProcessExitException from msg) =
718 if messageFingerprint msg == fingerprint (undefined :: a)
719 then exitHandler from decoded
720 else liftIO $ throwIO ex
723 -- Make sure the value is fully decoded so that we don't hang to
724 -- bytestrings if the caller doesn't use the value immediately
725 !decoded = decode (messageEncoding msg)
727-- | Lift 'Control.Exception.catches' (almost).
729-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
730-- message, a handler must accept inputs of the expected type. In order to
731-- handle a list of potentially different handlers (and therefore input types),
732-- a handler passed to 'catchesExit' must accept 'Message' and return
733-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
735-- See 'maybeHandleMessage' and 'Message' for more details.
736catchesExit :: Process b
737 -> [(ProcessId -> Message -> (Process (Maybe b)))]
739catchesExit act handlers = catch act ((flip handleExit) handlers)
741 handleExit :: ProcessExitException
742 -> [(ProcessId -> Message -> Process (Maybe b))]
744 handleExit ex [] = liftIO $ throwIO ex
745 handleExit ex@(ProcessExitException from msg) (h:hs) = do
748 Nothing -> handleExit ex hs
751-- | Our own process ID
752getSelfPid :: Process ProcessId
753getSelfPid = processId <$> ask
755-- | Get the node ID of our local node
756getSelfNode :: Process NodeId
757getSelfNode = localNodeId . processNode <$> ask
760-- | Get statistics about the specified node
761getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
763 selfNode <- getSelfNode
765 then Right `fmap` getLocalNodeStats -- optimisation
766 else getNodeStatsRemote selfNode
768 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
769 getNodeStatsRemote selfNode = do
770 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
771 bracket (monitorNode nid) unmonitor $ \mRef ->
772 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
773 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
774 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
777-- | Get statistics about our local node
778getLocalNodeStats :: Process NodeStats
779getLocalNodeStats = do
781 sendCtrlMsg Nothing $ GetNodeStats self
782 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
784-- | Get information about the specified process
785getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
787 let them = processNodeId pid in do
789 dest <- mkNode them us
790 sendCtrlMsg dest $ GetInfo pid
792 match (\(p :: ProcessInfo) -> return $ Just p)
793 , match (\(_ :: ProcessInfoNone) -> return Nothing)
795 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
796 mkNode them us = case them == us of
797 True -> return Nothing
798 _ -> return $ Just them
800--------------------------------------------------------------------------------
801-- Monitoring and linking --
802--------------------------------------------------------------------------------
804-- | Link to a remote process (asynchronous)
806-- When process A links to process B (that is, process A calls
807-- @link pidB@) then an asynchronous exception will be thrown to process A
808-- when process B terminates (normally or abnormally), or when process A gets
809-- disconnected from process B. Although it is /technically/ possible to catch
810-- these exceptions, chances are if you find yourself trying to do so you should
811-- probably be using 'monitor' rather than 'link'. In particular, code such as
813-- > link pidB -- Link to process B
814-- > expect -- Wait for a message from process B
815-- > unlink pidB -- Unlink again
817-- doesn't quite do what one might expect: if process B sends a message to
818-- process A, and /subsequently terminates/, then process A might or might not
819-- be terminated too, depending on whether the exception is thrown before or
820-- after the 'unlink' (i.e., this code has a race condition).
822-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
823-- call to 'link' has no effect.
825-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
826-- Linking makes no distinction between normal and abnormal termination of
827-- the remote process.
828link :: ProcessId -> Process ()
829link = sendCtrlMsg Nothing . Link . ProcessIdentifier
831-- | Monitor another process (asynchronous)
833-- When process A monitors process B (that is, process A calls
834-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
835-- when process B terminates (normally or abnormally), or when process A gets
836-- disconnected from process B. You receive this message like any other (using
837-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
838-- 'DiedDisconnect', etc.).
840-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
841-- multiple monitors are set up, multiple notifications will be delivered
842-- and monitors can be disabled individually using 'unmonitor'.
843monitor :: ProcessId -> Process MonitorRef
844monitor = monitor' . ProcessIdentifier
846-- | Establishes temporary monitoring of another process.
848-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
849-- of @code@. Note: although monitoring is no longer active when
850-- @withMonitor@ returns, there might still be unreceived monitor
851-- messages in the queue.
853withMonitor :: ProcessId -> Process a -> Process a
854withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
855 -- unmonitor blocks waiting for the response, so there's a possibility
856 -- that an exception might interrupt withMonitor before the unmonitor
857 -- has completed. I think that's better than making the unmonitor
862-- This is synchronous in the sense that once it returns you are guaranteed
863-- that no exception will be raised if the remote process dies. However, it is
864-- asynchronous in the sense that we do not wait for a response from the remote
866unlink :: ProcessId -> Process ()
869 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
873-- | Remove a node link
875-- This has the same synchronous/asynchronous nature as 'unlink'.
876unlinkNode :: NodeId -> Process ()
879 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
883-- | Remove a channel (send port) link
885-- This has the same synchronous/asynchronous nature as 'unlink'.
886unlinkPort :: SendPort a -> Process ()
888 unlinkPortAsync sport
889 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
895-- This has the same synchronous/asynchronous nature as 'unlink'.
896unmonitor :: MonitorRef -> Process ()
899 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
903--------------------------------------------------------------------------------
904-- Exception handling --
905--------------------------------------------------------------------------------
907-- | Lift 'Control.Exception.catch'
908catch :: Exception e => Process a -> (e -> Process a) -> Process a
911 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
913-- | Lift 'Control.Exception.try'
914try :: Exception e => Process a -> Process (Either e a)
917 liftIO $ Ex.try (runLocalProcess lproc p)
919-- | Lift 'Control.Exception.mask'
920mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
923 liftIO $ Ex.mask $ \restore ->
924 runLocalProcess lproc (p (liftRestore restore))
926 liftRestore :: (forall a. IO a -> IO a)
927 -> (forall a. Process a -> Process a)
928 liftRestore restoreIO = \p2 -> do
930 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
932-- | Lift 'Control.Exception.mask_'
933mask_ :: Process a -> Process a
936 liftIO $ Ex.mask_ $ runLocalProcess lproc p
938-- | Lift 'Control.Exception.onException'
939onException :: Process a -> Process b -> Process a
940onException p what = p `catch` \e -> do _ <- what
941 liftIO $ throwIO (e :: SomeException)
943-- | Lift 'Control.Exception.bracket'
944bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
945bracket before after thing =
946 mask $ \restore -> do
948 r <- restore (thing a) `onException` after a
952-- | Lift 'Control.Exception.bracket_'
953bracket_ :: Process a -> Process b -> Process c -> Process c
954bracket_ before after thing = bracket before (const after) (const thing)
956-- | Lift 'Control.Exception.finally'
957finally :: Process a -> Process b -> Process a
958finally a sequel = bracket_ (return ()) sequel a
960-- | You need this when using 'catches'
961data Handler a = forall e . Exception e => Handler (e -> Process a)
963instance Functor Handler where
964 fmap f (Handler h) = Handler (fmap f . h)
966-- | Lift 'Control.Exception.catches'
967catches :: Process a -> [Handler a] -> Process a
968catches proc handlers = proc `catch` catchesHandler handlers
970catchesHandler :: [Handler a] -> SomeException -> Process a
971catchesHandler handlers e = foldr tryHandler (throw e) handlers
972 where tryHandler (Handler handler) res
973 = case fromException e of
974 Just e' -> handler e'
977--------------------------------------------------------------------------------
979--------------------------------------------------------------------------------
981-- | Like 'expect' but with a timeout
982expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
983expectTimeout n = receiveTimeout n [match return]
985-- | Asynchronous version of 'spawn'
987-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
988spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
989spawnAsync nid proc = do
990 spawnRef <- getSpawnRef
993 then sendCtrlMsg Nothing $ Spawn proc spawnRef
994 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
997-- | Monitor a node (asynchronous)
998monitorNode :: NodeId -> Process MonitorRef
1000 monitor' . NodeIdentifier
1002-- | Monitor a typed channel (asynchronous)
1003monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1004monitorPort (SendPort cid) =
1005 monitor' (SendPortIdentifier cid)
1007-- | Remove a monitor (asynchronous)
1008unmonitorAsync :: MonitorRef -> Process ()
1010 sendCtrlMsg Nothing . Unmonitor
1012-- | Link to a node (asynchronous)
1013linkNode :: NodeId -> Process ()
1014linkNode = link' . NodeIdentifier
1016-- | Link to a channel (asynchronous)
1017linkPort :: SendPort a -> Process ()
1018linkPort (SendPort cid) =
1019 link' (SendPortIdentifier cid)
1021-- | Remove a process link (asynchronous)
1022unlinkAsync :: ProcessId -> Process ()
1024 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1026-- | Remove a node link (asynchronous)
1027unlinkNodeAsync :: NodeId -> Process ()
1029 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1031-- | Remove a channel (send port) link (asynchronous)
1032unlinkPortAsync :: SendPort a -> Process ()
1033unlinkPortAsync (SendPort cid) =
1034 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1036--------------------------------------------------------------------------------
1038--------------------------------------------------------------------------------
1042-- @say message@ sends a message (time, pid of the current process, message)
1043-- to the process registered as 'logger'. By default, this process simply
1044-- sends the string to 'stderr'. Individual Cloud Haskell backends might
1045-- replace this with a different logger process, however.
1046say :: String -> Process ()
1048 now <- liftIO getCurrentTime
1050 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
1052--------------------------------------------------------------------------------
1054--------------------------------------------------------------------------------
1056-- | Register a process with the local registry (synchronous). The name must not
1057-- already be registered. The process need not be on this node. A bad
1058-- registration will result in a 'ProcessRegistrationException'
1060-- The process to be registered does not have to be local itself.
1061register :: String -> ProcessId -> Process ()
1062register = registerImpl False
1064-- | Like 'register', but will replace an existing registration.
1065-- The name must already be registered.
1066reregister :: String -> ProcessId -> Process ()
1067reregister = registerImpl True
1069registerImpl :: Bool -> String -> ProcessId -> Process ()
1070registerImpl force label pid = do
1071 mynid <- getSelfNode
1072 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1074 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1075 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1078-- | Register a process with a remote registry (asynchronous).
1080-- The process to be registered does not have to live on the same remote node.
1081-- Reply wil come in the form of a 'RegisterReply' message
1083-- See comments in 'whereisRemoteAsync'
1084registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1085registerRemoteAsync nid label pid =
1086 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
1088reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1089reregisterRemoteAsync nid label pid =
1090 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1092-- | Remove a process from the local registry (asynchronous).
1093-- This version will wait until a response is gotten from the
1094-- management process. The name must already be registered.
1095unregister :: String -> Process ()
1096unregister label = do
1097 mynid <- getSelfNode
1098 sendCtrlMsg Nothing (Register label mynid Nothing False)
1100 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1101 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1104-- | Deal with the result from an attempted registration or unregistration
1105-- by throwing an exception if necessary
1106handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()
1107handleRegistrationReply label ok owner =
1109 liftIO $ throwIO $ ProcessRegistrationException label owner
1111-- | Remove a process from a remote registry (asynchronous).
1113-- Reply wil come in the form of a 'RegisterReply' message
1115-- See comments in 'whereisRemoteAsync'
1116unregisterRemoteAsync :: NodeId -> String -> Process ()
1117unregisterRemoteAsync nid label =
1118 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1120-- | Query the local process registry
1121whereis :: String -> Process (Maybe ProcessId)
1123 sendCtrlMsg Nothing (WhereIs label)
1124 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1125 (\(WhereIsReply _ mPid) -> return mPid)
1128-- | Query a remote process registry (asynchronous)
1130-- Reply will come in the form of a 'WhereIsReply' message.
1132-- There is currently no synchronous version of 'whereisRemoteAsync': if
1133-- you implement one yourself, be sure to take into account that the remote
1134-- node might die or get disconnect before it can respond (i.e. you should
1135-- use 'monitorNode' and take appropriate action when you receive a
1136-- 'NodeMonitorNotification').
1137whereisRemoteAsync :: NodeId -> String -> Process ()
1138whereisRemoteAsync nid label =
1139 sendCtrlMsg (Just nid) (WhereIs label)
1141-- | Named send to a process in the local registry (asynchronous)
1142nsend :: Serializable a => String -> a -> Process ()
1144 sendCtrlMsg Nothing (NamedSend label (createMessage msg))
1146-- | Named send to a process in the local registry (asynchronous).
1147-- This function makes /no/ attempt to serialize and (in the case when the
1148-- destination process resides on the same local node) therefore ensure that
1149-- the payload is fully evaluated before it is delivered.
1150unsafeNSend :: Serializable a => String -> a -> Process ()
1151unsafeNSend = Unsafe.nsend
1153-- | Named send to a process in a remote registry (asynchronous)
1154nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1155nsendRemote nid label msg =
1156 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1158--------------------------------------------------------------------------------
1160--------------------------------------------------------------------------------
1162-- | Resolve a static value
1163unStatic :: Typeable a => Static a -> Process a
1165 rtable <- remoteTable . processNode <$> ask
1166 case Static.unstatic rtable static of
1167 Left err -> fail $ "Could not resolve static value: " ++ err
1170-- | Resolve a closure
1171unClosure :: Typeable a => Closure a -> Process a
1172unClosure closure = do
1173 rtable <- remoteTable . processNode <$> ask
1174 case Static.unclosure rtable closure of
1175 Left err -> fail $ "Could not resolve closure: " ++ err
1178--------------------------------------------------------------------------------
1180--------------------------------------------------------------------------------
1182-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1183-- message passing. However, when network connections get disrupted this
1184-- illusion cannot always be maintained. Once a network connection breaks (even
1185-- temporarily) no further communication on that connection will be possible.
1186-- For example, if process A sends a message to process B, and A is then
1187-- notified (by monitor notification) that it got disconnected from B, A will
1188-- not be able to send any further messages to B, /unless/ A explicitly
1189-- indicates that it is acceptable to attempt to reconnect to B using the
1190-- Cloud Haskell 'reconnect' primitive.
1192-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1193-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1194-- then receives a monitor notification that its connection to B has been lost,
1195-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1196-- and m3 but not m2.
1198-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1199-- to attempt to reconnect on the next send/. In particular, if no further
1200-- communication attempts are made to B then A can use reconnect to clean up
1201-- its connection to B.
1202reconnect :: ProcessId -> Process ()
1205 node <- processNode <$> ask
1206 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1208-- | Reconnect to a sendport. See 'reconnect' for more information.
1209reconnectPort :: SendPort a -> Process ()
1210reconnectPort them = do
1212 node <- processNode <$> ask
1213 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1215--------------------------------------------------------------------------------
1216-- Auxiliary functions --
1217--------------------------------------------------------------------------------
1219sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1221 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1223sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1224sendChanLocal spId msg =
1225 -- we *must* fully serialize/encode the message here, because
1226 -- attempting to use `unsafeCoerce' in the node controller
1227 -- won't work since we know nothing about the required type
1228 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1230getMonitorRefFor :: Identifier -> Process MonitorRef
1231getMonitorRefFor ident = do
1233 liftIO $ modifyMVar (processState proc) $ \st -> do
1234 let counter = st ^. monitorCounter
1235 return ( monitorCounter ^: (+ 1) $ st
1236 , MonitorRef ident counter
1239getSpawnRef :: Process SpawnRef
1242 liftIO $ modifyMVar (processState proc) $ \st -> do
1243 let counter = st ^. spawnCounter
1244 return ( spawnCounter ^: (+ 1) $ st
1248-- | Monitor a process/node/channel
1249monitor' :: Identifier -> Process MonitorRef
1251 monitorRef <- getMonitorRefFor ident
1252 sendCtrlMsg Nothing $ Monitor monitorRef
1255-- | Link to a process/node/channel
1256link' :: Identifier -> Process ()
1257link' = 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