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
30 -- * Advanced messaging
56 -- * Process management
58 , ProcessTerminationException(..)
64 -- keep the exception constructor hidden, so that handling exit
65 -- reasons /must/ take place via the 'catchExit' family of primitives
66 , ProcessExitException()
74 -- * Monitoring and linking
89 , reregisterRemoteAsync
90 , unregisterRemoteAsync
96 -- * Exception handling
120 -- * Internal Exports
124#if ! MIN_VERSION_base(4,6,0)
125import Prelude hiding (catch)
128import Data.Binary (decode)
129import Data.Time.Clock (getCurrentTime)
130import Data.Time.Format (formatTime)
131#if MIN_VERSION_time(1,5,0)
132import Data.Time.Format (defaultTimeLocale)
134import System.Locale (defaultTimeLocale)
136import System.Timeout (timeout)
137import Control.Monad (when, void)
138import Control.Monad.Reader (ask)
139import Control.Monad.IO.Class (liftIO)
140import Control.Monad.Catch
146import qualified Control.Monad.Catch as Catch
147import Control.Applicative
148import Control.Distributed.Process.Internal.StrictMVar
153import Control.Concurrent.STM
162import Control.Distributed.Process.Internal.CQueue
167import Control.Distributed.Process.Serializable (Serializable, fingerprint)
168import Data.Accessor ((^.), (^:), (^=))
169import Control.Distributed.Static
173import Data.Rank1Typeable (Typeable)
174import qualified Control.Distributed.Static as Static (unstatic, unclosure)
175import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
176import Control.Distributed.Process.Internal.Types
186 , NodeMonitorNotification(..)
187 , ProcessMonitorNotification(..)
197 , ProcessExitException(..)
200 , DidUnlinkProcess(..)
205 , ProcessRegistrationException(..)
207 , ProcessInfoNone(..)
211 , createUnencodedMessage
212 , ImplicitReconnect( NoImplicitReconnect)
217import Control.Distributed.Process.Internal.Messaging
224import Control.Distributed.Process.Management.Internal.Types
227import Control.Distributed.Process.Management.Internal.Trace.Types
230import Control.Distributed.Process.Internal.WeakTQueue
239--------------------------------------------------------------------------------
241--------------------------------------------------------------------------------
244send :: Serializable a => ProcessId -> a -> Process ()
245-- This requires a lookup on every send. If we want to avoid that we need to
246-- modify serializable to allow for stateful (IO) deserialization.
249 let us = processId proc
250 node = processNode proc
251 nodeId = localNodeId node
252 destNode = (processNodeId them) in do
253 case destNode == nodeId of
254 True -> sendLocal them msg
255 False -> liftIO $ sendMessage (processNode proc)
256 (ProcessIdentifier (processId proc))
257 (ProcessIdentifier them)
260 -- We do not fire the trace event until after the sending is complete;
261 -- In the remote case, 'sendMessage' can block in the networking stack.
262 liftIO $ traceEvent (localEventBus node)
263 (MxSent them us (createUnencodedMessage msg))
265-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
266-- and (in the case when the destination process resides on the same local
267-- node) therefore ensure that the payload is fully evaluated before it is
269unsafeSend :: Serializable a => ProcessId -> a -> Process ()
270unsafeSend = Unsafe.send
272-- | Send a message unreliably.
274-- Unlike 'send', this function is insensitive to 'reconnect'. It will
275-- try to send the message regardless of the history of connection failures
278-- Message passing with 'usend' is ordered for a given sender and receiver
279-- if the messages arrive at all.
281usend :: Serializable a => ProcessId -> a -> Process ()
284 let there = processNodeId them
286 then sendLocal them msg
287 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
290-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize
291-- the message when the destination process resides on the same local
292-- node. Therefore, a local receiver would need to be prepared to cope with any
293-- errors resulting from evaluation of the message.
294unsafeUSend :: Serializable a => ProcessId -> a -> Process ()
295unsafeUSend = Unsafe.usend
297-- | Wait for a message of a specific type
298expect :: forall a. Serializable a => Process a
299expect = receiveWait [match return]
301--------------------------------------------------------------------------------
303--------------------------------------------------------------------------------
305-- | Create a new typed channel
306newChan :: Serializable a => Process (SendPort a, ReceivePort a)
309 liftIO . modifyMVar (processState proc) $ \st -> do
310 let lcid = st ^. channelCounter
311 let cid = SendPortId { sendPortProcessId = processId proc
312 , sendPortLocalId = lcid
314 let sport = SendPort cid
315 chan <- liftIO newTQueueIO
316 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
317 let rport = ReceivePort $ readTQueue chan
318 let tch = TypedChannel chan'
319 return ( (channelCounter ^: (+ 1))
320 . (typedChannelWithId lcid ^= Just tch)
325 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
326 finalizer st lcid = modifyMVar_ st $
327 return . (typedChannelWithId lcid ^= Nothing)
329-- | Send a message on a typed channel
330sendChan :: Serializable a => SendPort a -> a -> Process ()
331sendChan (SendPort cid) msg = do
333 let node = localNodeId (processNode proc)
334 destNode = processNodeId (sendPortProcessId cid) in do
335 case destNode == node of
336 True -> sendChanLocal cid msg
338 liftIO $ sendBinary (processNode proc)
339 (ProcessIdentifier (processId proc))
340 (SendPortIdentifier cid)
344-- | Send a message on a typed channel. This function makes /no/ attempt to
345-- serialize and (in the case when the @ReceivePort@ resides on the same local
346-- node) therefore ensure that the payload is fully evaluated before it is
348unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
349unsafeSendChan = Unsafe.sendChan
351-- | Wait for a message on a typed channel
352receiveChan :: Serializable a => ReceivePort a -> Process a
353receiveChan = liftIO . atomically . receiveSTM
355-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
356-- non-blocking check for a message.
357receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
358receiveChanTimeout 0 ch = liftIO . atomically $
359 (Just <$> receiveSTM ch) `orElse` return Nothing
360receiveChanTimeout n ch = liftIO . timeout n . atomically $
363-- | Merge a list of typed channels.
365-- The result port is left-biased: if there are messages available on more
366-- than one port, the first available message is returned.
367mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
368mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
370-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
372mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
373mergePortsRR = \ps -> do
374 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
375 return $ ReceivePort (rr psVar)
379 rotate (x:xs) = xs ++ [x]
381 rr :: TVar [STM a] -> STM a
384 a <- foldr1 orElse ps
385 writeTVar psVar (rotate ps)
388--------------------------------------------------------------------------------
389-- Advanced messaging --
390--------------------------------------------------------------------------------
392-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
393newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
396-- | Test the matches in order against each message in the queue
397receiveWait :: [Match b] -> Process b
399 queue <- processQueue <$> ask
400 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
403-- | Like 'receiveWait' but with a timeout.
405-- If the timeout is zero do a non-blocking check for matching messages. A
406-- non-zero timeout is applied only when waiting for incoming messages (that is,
407-- /after/ we have checked the messages that are already in the mailbox).
408receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
409receiveTimeout t ms = do
410 queue <- processQueue <$> ask
411 let blockSpec = if t == 0 then NonBlocking else Timeout t
412 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
414 Nothing -> return Nothing
415 Just proc -> Just <$> proc
417-- | Match on a typed channel
418matchChan :: ReceivePort a -> (a -> Process b) -> Match b
419matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
421-- | Match on an arbitrary STM action.
423-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
424-- actions with checks against our process' mailbox and/or any typed channel
425-- @ReceivePort@s we may hold.
427-- This allows us to process multiple input streams /along with our mailbox/,
428-- in just the same way that 'matchChan' supports checking both the mailbox
429-- /and/ an arbitrary set of typed channels in one atomic transaction.
431-- Note there are no ordering guarnatees with respect to these disparate input
434matchSTM :: STM a -> (a -> Process b) -> Match b
435matchSTM stm fn = Match $ MatchChan (fmap fn stm)
437-- | Match against any message of the right type
438match :: forall a b. Serializable a => (a -> Process b) -> Match b
439match = matchIf (const True)
441-- | Match against any message of the right type that satisfies a predicate
442matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
443matchIf c p = Match $ MatchMsg $ \msg ->
444 case messageFingerprint msg == fingerprint (undefined :: a) of
447 (UnencodedMessage _ m) ->
448 let m' = unsafeCoerce m :: a in
452 (EncodedMessage _ _) ->
453 if (c decoded) then Just (p decoded) else Nothing
456 -- Make sure the value is fully decoded so that we don't hang to
457 -- bytestrings when the process calling 'matchIf' doesn't process
458 -- the values immediately
459 !decoded = decode (messageEncoding msg)
461-- | Match against any message, regardless of the underlying (contained) type
462matchMessage :: (Message -> Process Message) -> Match Message
463matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
465-- | Match against any message (regardless of underlying type) that satisfies a predicate
466matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
467matchMessageIf c p = Match $ MatchMsg $ \msg ->
472-- | Forward a raw 'Message' to the given 'ProcessId'.
473forward :: Message -> ProcessId -> Process ()
476 let node = processNode proc
478 nid = localNodeId node
479 destNode = (processNodeId them) in do
480 case destNode == nid of
481 True -> sendCtrlMsg Nothing (LocalSend them msg)
482 False -> liftIO $ sendPayload (processNode proc)
483 (ProcessIdentifier (processId proc))
484 (ProcessIdentifier them)
486 (messageToPayload msg)
487 -- We do not fire the trace event until after the sending is complete;
488 -- In the remote case, 'sendMessage' can block in the networking stack.
489 liftIO $ traceEvent (localEventBus node)
492-- | Forward a raw 'Message' to the given 'ProcessId'.
494-- Unlike 'forward', this function is insensitive to 'reconnect'. It will
495-- try to send the message regardless of the history of connection failures
497uforward :: Message -> ProcessId -> Process ()
498uforward msg them = do
500 let node = processNode proc
502 nid = localNodeId node
503 destNode = (processNodeId them) in do
504 case destNode == nid of
505 True -> sendCtrlMsg Nothing (LocalSend them msg)
506 False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)
508 -- We do not fire the trace event until after the sending is complete;
509 -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
510 liftIO $ traceEvent (localEventBus node)
513-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
514-- 'Serializable' - like the datum they contain - but also note, deserialising
515-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
516-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
521-- send self (wrapMessage "blah")
522-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
523-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
524-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
527wrapMessage :: Serializable a => a -> Message
528wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
530-- | This is the /unsafe/ variant of 'wrapMessage'. See
531-- "Control.Distributed.Process.UnsafePrimitives" for details.
532unsafeWrapMessage :: Serializable a => a -> Message
533unsafeWrapMessage = Unsafe.wrapMessage
535-- [note Serializable UnencodedMessage]
536-- if we attempt to serialise an UnencodedMessage, it will be converted to an
537-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
538-- Primitives.hs for the gory details
540-- | Attempt to unwrap a raw 'Message'.
541-- If the type of the decoded message payload matches the expected type, the
542-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
545-- This expression, for example, will evaluate to @Nothing@
546-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
548-- Whereas this expression, will yield @Just "foo"@
549-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
551unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
553 case messageFingerprint msg == fingerprint (undefined :: a) of
554 False -> return Nothing :: m (Maybe a)
556 (UnencodedMessage _ ms) ->
557 let ms' = unsafeCoerce ms :: a
559 (EncodedMessage _ _) ->
560 return (Just (decoded))
562 decoded :: a -- note [decoding]
563 !decoded = decode (messageEncoding msg)
565-- | Attempt to handle a raw 'Message'.
566-- If the type of the message matches the type of the first argument to
567-- the supplied expression, then the message will be decoded and the expression
568-- evaluated against its value. If this runtime type checking fails however,
569-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
570-- evaluation proceeds, the resulting value with be wrapped with @Just@.
572-- Intended for use in `catchesExit` and `matchAny` primitives.
574handleMessage :: forall m a b. (Monad m, Serializable a)
575 => Message -> (a -> m b) -> m (Maybe b)
576handleMessage msg proc = handleMessageIf msg (const True) proc
578-- | Conditionally handle a raw 'Message'.
579-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
580-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
581-- for further information about runtime type checking.
583handleMessageIf :: forall m a b . (Monad m, Serializable a)
588handleMessageIf msg c proc = do
589 case messageFingerprint msg == fingerprint (undefined :: a) of
590 False -> return Nothing :: m (Maybe b)
592 (UnencodedMessage _ ms) ->
593 let ms' = unsafeCoerce ms :: a in
595 True -> do { r <- proc ms'; return (Just r) }
596 False -> return Nothing :: m (Maybe b)
597 (EncodedMessage _ _) ->
599 True -> do { r <- proc (decoded :: a); return (Just r) }
600 False -> return Nothing :: m (Maybe b)
602 decoded :: a -- note [decoding]
603 !decoded = decode (messageEncoding msg)
605-- | As 'handleMessage' but ignores result, which is useful if you don't
606-- care whether or not the handler succeeded.
608handleMessage_ :: forall m a . (Monad m, Serializable a)
609 => Message -> (a -> m ()) -> m ()
610handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
612-- | Conditional version of 'handleMessage_'.
614handleMessageIf_ :: forall m a . (Monad m, Serializable a)
619handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
621-- | Match against an arbitrary message. 'matchAny' removes the first available
622-- message from the process mailbox. To handle arbitrary /raw/ messages once
623-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
625matchAny :: forall b. (Message -> Process b) -> Match b
626matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
628-- | Match against an arbitrary message. Intended for use with 'handleMessage'
629-- and 'unwrapMessage', this function /only/ removes a message from the process
630-- mailbox, /if/ the supplied condition matches. The success (or failure) of
631-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
632-- here, i.e., if the condition evaluates to @True@ then the message will
633-- be removed from the process mailbox and decoded, but that does /not/
634-- guarantee that an expression passed to @handleMessage@ will pass the
635-- runtime type checks and therefore be evaluated.
637matchAnyIf :: forall a b. (Serializable a)
639 -> (Message -> Process b)
641matchAnyIf c p = Match $ MatchMsg $ \msg ->
642 case messageFingerprint msg == fingerprint (undefined :: a) of
643 True | check -> Just (p msg)
648 (EncodedMessage _ _) -> c decoded
649 (UnencodedMessage _ m') -> c (unsafeCoerce m')
651 decoded :: a -- note [decoding]
652 !decoded = decode (messageEncoding msg)
656For an EncodedMessage, we need to ensure the value is fully decoded so that
657we don't hang to bytestrings if the calling process doesn't evaluate
658immediately. For UnencodedMessage we know (because the fingerprint comparison
659succeeds) that unsafeCoerce will not fail.
662-- | Remove any message from the queue
663matchUnknown :: Process b -> Match b
664matchUnknown p = Match $ MatchMsg (const (Just p))
666-- | Receives messages and forwards them to @pid@ if @p msg == True@.
667delegate :: ProcessId -> (Message -> Bool) -> Process ()
670 matchAny (\m -> case (p m) of
671 True -> forward m pid
676-- | A straight relay that forwards all messages to the supplied pid.
677relay :: ProcessId -> Process ()
678relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
680-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
681-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
682-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
683-- fails, no action is taken and the proxy process will continue running.
684proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
688 next <- handleMessage m proc
690 Just True -> forward m pid
691 Just False -> return () -- explicitly ignored
692 Nothing -> return ()) -- un-routable
696--------------------------------------------------------------------------------
697-- Process management --
698--------------------------------------------------------------------------------
700-- | Thrown by 'terminate'
701data ProcessTerminationException = ProcessTerminationException
702 deriving (Show, Typeable)
704instance Exception ProcessTerminationException
706-- | Terminate immediately (throws a ProcessTerminationException)
707terminate :: Process a
708terminate = throwM ProcessTerminationException
711-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
712die :: Serializable a => a -> Process b
714-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
715-- could be decoded by a handler passed to 'catchExit', re-thrown or even
716-- passed to another process via the tracing mechanism.
718 throwM (ProcessExitException pid (createMessage reason))
720-- | Forceful request to kill a process. Where 'exit' provides an exception
721-- that can be caught and handled, 'kill' throws an unexposed exception type
722-- which cannot be handled explicitly (by type).
723kill :: ProcessId -> String -> Process ()
724-- NOTE: We send the message to our local node controller, which will then
725-- forward it to a remote node controller (if applicable). Sending it directly
726-- to a remote node controller means that that the message may overtake a
727-- 'monitor' or 'link' request.
728kill them reason = sendCtrlMsg Nothing (Kill them reason)
730-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
731-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
732-- manner can be handled using the 'catchExit' family of functions.
733exit :: Serializable a => ProcessId -> a -> Process ()
734-- NOTE: We send the message to our local node controller, which will then
735-- forward it to a remote node controller (if applicable). Sending it directly
736-- to a remote node controller means that that the message may overtake a
737-- 'monitor' or 'link' request.
738exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
740-- | Catches 'ProcessExitException'. The handler will not be applied unless its
741-- type matches the encoded data stored in the exception (see the /reason/
742-- argument given to the 'exit' primitive). If the handler cannot be applied,
743-- the exception will be re-thrown.
745-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
746-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
747catchExit :: forall a b . (Show a, Serializable a)
749 -> (ProcessId -> a -> Process b)
751catchExit act exitHandler = Catch.catch act handleExit
753 handleExit ex@(ProcessExitException from msg) =
754 if messageFingerprint msg == fingerprint (undefined :: a)
755 then exitHandler from decoded
759 -- Make sure the value is fully decoded so that we don't hang to
760 -- bytestrings if the caller doesn't use the value immediately
761 !decoded = decode (messageEncoding msg)
763-- | Lift 'Control.Exception.catches' (almost).
765-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
766-- message, a handler must accept inputs of the expected type. In order to
767-- handle a list of potentially different handlers (and therefore input types),
768-- a handler passed to 'catchesExit' must accept 'Message' and return
769-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
771-- See 'maybeHandleMessage' and 'Message' for more details.
772catchesExit :: Process b
773 -> [(ProcessId -> Message -> (Process (Maybe b)))]
775catchesExit act handlers = Catch.catch act ((flip handleExit) handlers)
777 handleExit :: ProcessExitException
778 -> [(ProcessId -> Message -> Process (Maybe b))]
780 handleExit ex [] = throwM ex
781 handleExit ex@(ProcessExitException from msg) (h:hs) = do
784 Nothing -> handleExit ex hs
787-- | Our own process ID
788getSelfPid :: Process ProcessId
789getSelfPid = processId <$> ask
791-- | Get the node ID of our local node
792getSelfNode :: Process NodeId
793getSelfNode = localNodeId . processNode <$> ask
796-- | Get statistics about the specified node
797getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
799 selfNode <- getSelfNode
801 then Right `fmap` getLocalNodeStats -- optimisation
802 else getNodeStatsRemote selfNode
804 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
805 getNodeStatsRemote selfNode = do
806 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
807 bracket (monitorNode nid) unmonitor $ \mRef ->
808 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
809 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
810 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
813-- | Get statistics about our local node
814getLocalNodeStats :: Process NodeStats
815getLocalNodeStats = do
817 sendCtrlMsg Nothing $ GetNodeStats self
818 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
820-- | Get information about the specified process
821getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
823 let them = processNodeId pid in do
825 dest <- mkNode them us
826 sendCtrlMsg dest $ GetInfo pid
828 match (\(p :: ProcessInfo) -> return $ Just p)
829 , match (\(_ :: ProcessInfoNone) -> return Nothing)
831 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
832 mkNode them us = case them == us of
833 True -> return Nothing
834 _ -> return $ Just them
836--------------------------------------------------------------------------------
837-- Monitoring and linking --
838--------------------------------------------------------------------------------
840-- | Link to a remote process (asynchronous)
842-- When process A links to process B (that is, process A calls
843-- @link pidB@) then an asynchronous exception will be thrown to process A
844-- when process B terminates (normally or abnormally), or when process A gets
845-- disconnected from process B. Although it is /technically/ possible to catch
846-- these exceptions, chances are if you find yourself trying to do so you should
847-- probably be using 'monitor' rather than 'link'. In particular, code such as
849-- > link pidB -- Link to process B
850-- > expect -- Wait for a message from process B
851-- > unlink pidB -- Unlink again
853-- doesn't quite do what one might expect: if process B sends a message to
854-- process A, and /subsequently terminates/, then process A might or might not
855-- be terminated too, depending on whether the exception is thrown before or
856-- after the 'unlink' (i.e., this code has a race condition).
858-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
859-- call to 'link' has no effect.
861-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
862-- Linking makes no distinction between normal and abnormal termination of
863-- the remote process.
864link :: ProcessId -> Process ()
865link = sendCtrlMsg Nothing . Link . ProcessIdentifier
867-- | Monitor another process (asynchronous)
869-- When process A monitors process B (that is, process A calls
870-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
871-- when process B terminates (normally or abnormally), or when process A gets
872-- disconnected from process B. You receive this message like any other (using
873-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
874-- 'DiedDisconnect', etc.).
876-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
877-- multiple monitors are set up, multiple notifications will be delivered
878-- and monitors can be disabled individually using 'unmonitor'.
879monitor :: ProcessId -> Process MonitorRef
880monitor = monitor' . ProcessIdentifier
882-- | Establishes temporary monitoring of another process.
884-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
885-- of @code@. Note: although monitoring is no longer active when
886-- @withMonitor@ returns, there might still be unreceived monitor
887-- messages in the queue.
889withMonitor :: ProcessId -> Process a -> Process a
890withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
891 -- unmonitor blocks waiting for the response, so there's a possibility
892 -- that an exception might interrupt withMonitor before the unmonitor
893 -- has completed. I think that's better than making the unmonitor
898-- This is synchronous in the sense that once it returns you are guaranteed
899-- that no exception will be raised if the remote process dies. However, it is
900-- asynchronous in the sense that we do not wait for a response from the remote
902unlink :: ProcessId -> Process ()
905 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
909-- | Remove a node link
911-- This has the same synchronous/asynchronous nature as 'unlink'.
912unlinkNode :: NodeId -> Process ()
915 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
919-- | Remove a channel (send port) link
921-- This has the same synchronous/asynchronous nature as 'unlink'.
922unlinkPort :: SendPort a -> Process ()
924 unlinkPortAsync sport
925 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
931-- This has the same synchronous/asynchronous nature as 'unlink'.
933-- ProcessMonitorNotification messages for the given MonitorRef are removed from
935unmonitor :: MonitorRef -> Process ()
938 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
941 -- Discard the notification if any. With the current NC implementation at most
942 -- one notification is in the mailbox for any given ref.
943 void $ receiveTimeout 0
944 [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref' == ref)
948--------------------------------------------------------------------------------
949-- Exception handling --
950--------------------------------------------------------------------------------
952-- | Lift 'Control.Exception.catch'
953catch :: Exception e => Process a -> (e -> Process a) -> Process a
955{-# DEPRECATED catch "Use Control.Monad.Catch.catch instead" #-}
957-- | Lift 'Control.Exception.try'
958try :: Exception e => Process a -> Process (Either e a)
960{-# DEPRECATED try "Use Control.Monad.Catch.mask_ instead" #-}
962-- | Lift 'Control.Exception.mask'
963mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
965{-# DEPRECATED mask "Use Control.Monad.Catch.mask_ instead" #-}
967-- | Lift 'Control.Exception.mask_'
968mask_ :: Process a -> Process a
970{-# DEPRECATED mask_ "Use Control.Monad.Catch.mask_ instead" #-}
972-- | Lift 'Control.Exception.onException'
973onException :: Process a -> Process b -> Process a
974onException = Catch.onException
975{-# DEPRECATED onException "Use Control.Monad.Catch.onException instead" #-}
977-- | Lift 'Control.Exception.bracket'
978bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
979bracket = Catch.bracket
980{-# DEPRECATED bracket "Use Control.Monad.Catch.bracket instead" #-}
982-- | Lift 'Control.Exception.bracket_'
983bracket_ :: Process a -> Process b -> Process c -> Process c
984bracket_ = Catch.bracket_
985{-# DEPRECATED bracket_ "Use Control.Monad.Catch.bracket_ instead" #-}
987-- | Lift 'Control.Exception.finally'
988finally :: Process a -> Process b -> Process a
989finally = Catch.finally
990{-# DEPRECATED finally "Use Control.Monad.Catch.finally instead" #-}
992-- | You need this when using 'catches'
993data Handler a = forall e . Exception e => Handler (e -> Process a)
995instance Functor Handler where
996 fmap f (Handler h) = Handler (fmap f . h)
998-- | Lift 'Control.Exception.catches'
999catches :: Process a -> [Handler a] -> Process a
1000catches proc handlers = proc `Catch.catch` catchesHandler handlers
1002catchesHandler :: [Handler a] -> SomeException -> Process a
1003catchesHandler handlers e = foldr tryHandler (throwM e) handlers
1004 where tryHandler (Handler handler) res
1005 = case fromException e of
1006 Just e' -> handler e'
1009--------------------------------------------------------------------------------
1011--------------------------------------------------------------------------------
1013-- | Like 'expect' but with a timeout
1014expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
1015expectTimeout n = receiveTimeout n [match return]
1017-- | Asynchronous version of 'spawn'
1019-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
1020spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
1021spawnAsync nid proc = do
1022 spawnRef <- getSpawnRef
1025 then sendCtrlMsg Nothing $ Spawn proc spawnRef
1026 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
1029-- | Monitor a node (asynchronous)
1030monitorNode :: NodeId -> Process MonitorRef
1032 monitor' . NodeIdentifier
1034-- | Monitor a typed channel (asynchronous)
1035monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1036monitorPort (SendPort cid) =
1037 monitor' (SendPortIdentifier cid)
1039-- | Remove a monitor (asynchronous)
1040unmonitorAsync :: MonitorRef -> Process ()
1042 sendCtrlMsg Nothing . Unmonitor
1044-- | Link to a node (asynchronous)
1045linkNode :: NodeId -> Process ()
1046linkNode = link' . NodeIdentifier
1048-- | Link to a channel (asynchronous)
1049linkPort :: SendPort a -> Process ()
1050linkPort (SendPort cid) =
1051 link' (SendPortIdentifier cid)
1053-- | Remove a process link (asynchronous)
1054unlinkAsync :: ProcessId -> Process ()
1056 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1058-- | Remove a node link (asynchronous)
1059unlinkNodeAsync :: NodeId -> Process ()
1061 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1063-- | Remove a channel (send port) link (asynchronous)
1064unlinkPortAsync :: SendPort a -> Process ()
1065unlinkPortAsync (SendPort cid) =
1066 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1068--------------------------------------------------------------------------------
1070--------------------------------------------------------------------------------
1074-- @say message@ sends a message (time, pid of the current process, message)
1075-- to the process registered as 'logger'. By default, this process simply
1076-- sends the string to 'stderr'. Individual Cloud Haskell backends might
1077-- replace this with a different logger process, however.
1078say :: String -> Process ()
1080 now <- liftIO getCurrentTime
1082 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
1084--------------------------------------------------------------------------------
1086--------------------------------------------------------------------------------
1088-- | Register a process with the local registry (synchronous). The name must not
1089-- already be registered. The process need not be on this node. A bad
1090-- registration will result in a 'ProcessRegistrationException'
1092-- The process to be registered does not have to be local itself.
1093register :: String -> ProcessId -> Process ()
1094register = registerImpl False
1096-- | Like 'register', but will replace an existing registration.
1097-- The name must already be registered.
1098reregister :: String -> ProcessId -> Process ()
1099reregister = registerImpl True
1101registerImpl :: Bool -> String -> ProcessId -> Process ()
1102registerImpl force label pid = do
1103 mynid <- getSelfNode
1104 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1106 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1107 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1110-- | Register a process with a remote registry (asynchronous).
1112-- The process to be registered does not have to live on the same remote node.
1113-- Reply wil come in the form of a 'RegisterReply' message
1115-- See comments in 'whereisRemoteAsync'
1116registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1117registerRemoteAsync nid label pid = do
1119 sendCtrlMsg (if nid == here then Nothing else Just nid)
1120 (Register label nid (Just pid) False)
1122reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1123reregisterRemoteAsync nid label pid =
1124 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1126-- | Remove a process from the local registry (asynchronous).
1127-- This version will wait until a response is gotten from the
1128-- management process. The name must already be registered.
1129unregister :: String -> Process ()
1130unregister label = do
1131 mynid <- getSelfNode
1132 sendCtrlMsg Nothing (Register label mynid Nothing False)
1134 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1135 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1138-- | Deal with the result from an attempted registration or unregistration
1139-- by throwing an exception if necessary
1140handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()
1141handleRegistrationReply label ok owner =
1143 throwM $ ProcessRegistrationException label owner
1145-- | Remove a process from a remote registry (asynchronous).
1147-- Reply wil come in the form of a 'RegisterReply' message
1149-- See comments in 'whereisRemoteAsync'
1150unregisterRemoteAsync :: NodeId -> String -> Process ()
1151unregisterRemoteAsync nid label =
1152 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1154-- | Query the local process registry
1155whereis :: String -> Process (Maybe ProcessId)
1157 sendCtrlMsg Nothing (WhereIs label)
1158 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1159 (\(WhereIsReply _ mPid) -> return mPid)
1162-- | Query a remote process registry (asynchronous)
1164-- Reply will come in the form of a 'WhereIsReply' message.
1166-- There is currently no synchronous version of 'whereisRemoteAsync': if
1167-- you implement one yourself, be sure to take into account that the remote
1168-- node might die or get disconnect before it can respond (i.e. you should
1169-- use 'monitorNode' and take appropriate action when you receive a
1170-- 'NodeMonitorNotification').
1171whereisRemoteAsync :: NodeId -> String -> Process ()
1172whereisRemoteAsync nid label = do
1174 sendCtrlMsg (if nid == here then Nothing else Just nid) (WhereIs label)
1176-- | Named send to a process in the local registry (asynchronous)
1177nsend :: Serializable a => String -> a -> Process ()
1179 sendCtrlMsg Nothing (NamedSend label (createMessage msg))
1181-- | Named send to a process in the local registry (asynchronous).
1182-- This function makes /no/ attempt to serialize and (in the case when the
1183-- destination process resides on the same local node) therefore ensure that
1184-- the payload is fully evaluated before it is delivered.
1185unsafeNSend :: Serializable a => String -> a -> Process ()
1186unsafeNSend = Unsafe.nsend
1188-- | Named send to a process in a remote registry (asynchronous)
1189nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1190nsendRemote nid label msg = do
1192 if here == nid then nsend label msg
1193 else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1195-- | Named send to a process in a remote registry (asynchronous)
1196-- This function makes /no/ attempt to serialize and (in the case when the
1197-- destination process resides on the same local node) therefore ensure that
1198-- the payload is fully evaluated before it is delivered.
1199unsafeNSendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1200unsafeNSendRemote = Unsafe.nsendRemote
1202--------------------------------------------------------------------------------
1204--------------------------------------------------------------------------------
1206-- | Resolve a static value
1207unStatic :: Typeable a => Static a -> Process a
1209 rtable <- remoteTable . processNode <$> ask
1210 case Static.unstatic rtable static of
1211 Left err -> fail $ "Could not resolve static value: " ++ err
1214-- | Resolve a closure
1215unClosure :: Typeable a => Closure a -> Process a
1216unClosure closure = do
1217 rtable <- remoteTable . processNode <$> ask
1218 case Static.unclosure rtable closure of
1219 Left err -> fail $ "Could not resolve closure: " ++ err
1222--------------------------------------------------------------------------------
1224--------------------------------------------------------------------------------
1226-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1227-- message passing. However, when network connections get disrupted this
1228-- illusion cannot always be maintained. Once a network connection breaks (even
1229-- temporarily) no further communication on that connection will be possible.
1230-- For example, if process A sends a message to process B, and A is then
1231-- notified (by monitor notification) that it got disconnected from B, A will
1232-- not be able to send any further messages to B, /unless/ A explicitly
1233-- indicates that it is acceptable to attempt to reconnect to B using the
1234-- Cloud Haskell 'reconnect' primitive.
1236-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1237-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1238-- then receives a monitor notification that its connection to B has been lost,
1239-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1240-- and m3 but not m2.
1242-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1243-- to attempt to reconnect on the next send/. In particular, if no further
1244-- communication attempts are made to B then A can use reconnect to clean up
1245-- its connection to B.
1246reconnect :: ProcessId -> Process ()
1249 node <- processNode <$> ask
1250 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1252-- | Reconnect to a sendport. See 'reconnect' for more information.
1253reconnectPort :: SendPort a -> Process ()
1254reconnectPort them = do
1256 node <- processNode <$> ask
1257 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1259--------------------------------------------------------------------------------
1260-- Auxiliary functions --
1261--------------------------------------------------------------------------------
1263sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1265 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1267sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1268sendChanLocal spId msg =
1269 -- we *must* fully serialize/encode the message here, because
1270 -- attempting to use `unsafeCoerce' in the node controller
1271 -- won't work since we know nothing about the required type
1272 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1274getMonitorRefFor :: Identifier -> Process MonitorRef
1275getMonitorRefFor ident = do
1277 liftIO $ modifyMVar (processState proc) $ \st -> do
1278 let counter = st ^. monitorCounter
1279 return ( monitorCounter ^: (+ 1) $ st
1280 , MonitorRef ident counter
1283getSpawnRef :: Process SpawnRef
1286 liftIO $ modifyMVar (processState proc) $ \st -> do
1287 let counter = st ^. spawnCounter
1288 return ( spawnCounter ^: (+ 1) $ st
1292-- | Monitor a process/node/channel
1293monitor' :: Identifier -> Process MonitorRef
1295 monitorRef <- getMonitorRefFor ident
1296 sendCtrlMsg Nothing $ Monitor monitorRef
1299-- | Link to a process/node/channel
1300link' :: Identifier -> Process ()
1301link' = 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