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
90 , reregisterRemoteAsync
91 , unregisterRemoteAsync
97 -- * Exception handling
121 -- * Internal Exports
125#if ! MIN_VERSION_base(4,6,0)
126import Prelude hiding (catch)
129import Data.Binary (decode)
130import Data.Time.Clock (getCurrentTime)
131import Data.Time.Format (formatTime)
132#if MIN_VERSION_time(1,5,0)
133import Data.Time.Format (defaultTimeLocale)
135import System.Locale (defaultTimeLocale)
137import System.Timeout (timeout)
138import Control.Monad (when, void)
139import Control.Monad.Reader (ask)
140import Control.Monad.IO.Class (liftIO)
141import Control.Monad.Catch
147import qualified Control.Monad.Catch as Catch
148import Control.Applicative
149import Control.Distributed.Process.Internal.StrictMVar
154import Control.Concurrent.STM
163import Control.Distributed.Process.Internal.CQueue
168import Control.Distributed.Process.Serializable (Serializable, fingerprint)
169import Data.Accessor ((^.), (^:), (^=))
170import Control.Distributed.Static
174import Data.Rank1Typeable (Typeable)
175import qualified Control.Distributed.Static as Static (unstatic, unclosure)
176import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
177import Control.Distributed.Process.Internal.Types
187 , NodeMonitorNotification(..)
188 , ProcessMonitorNotification(..)
198 , ProcessExitException(..)
201 , DidUnlinkProcess(..)
206 , ProcessRegistrationException(..)
208 , ProcessInfoNone(..)
212 , createUnencodedMessage
213 , ImplicitReconnect( NoImplicitReconnect)
218import Control.Distributed.Process.Internal.Messaging
225import Control.Distributed.Process.Management.Internal.Types
228import Control.Distributed.Process.Management.Internal.Trace.Types
231import Control.Distributed.Process.Internal.WeakTQueue
240--------------------------------------------------------------------------------
242--------------------------------------------------------------------------------
245send :: Serializable a => ProcessId -> a -> Process ()
246-- This requires a lookup on every send. If we want to avoid that we need to
247-- modify serializable to allow for stateful (IO) deserialization.
250 let us = processId proc
251 node = processNode proc
252 nodeId = localNodeId node
253 destNode = (processNodeId them) in do
254 case destNode == nodeId of
255 True -> sendLocal them msg
256 False -> liftIO $ sendMessage (processNode proc)
257 (ProcessIdentifier (processId proc))
258 (ProcessIdentifier them)
261 -- We do not fire the trace event until after the sending is complete;
262 -- In the remote case, 'sendMessage' can block in the networking stack.
263 liftIO $ traceEvent (localEventBus node)
264 (MxSent them us (createUnencodedMessage msg))
266-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
267-- and (in the case when the destination process resides on the same local
268-- node) therefore ensure that the payload is fully evaluated before it is
270unsafeSend :: Serializable a => ProcessId -> a -> Process ()
271unsafeSend = Unsafe.send
273-- | Send a message unreliably.
275-- Unlike 'send', this function is insensitive to 'reconnect'. It will
276-- try to send the message regardless of the history of connection failures
279-- Message passing with 'usend' is ordered for a given sender and receiver
280-- if the messages arrive at all.
282usend :: Serializable a => ProcessId -> a -> Process ()
285 let there = processNodeId them
287 then sendLocal them msg
288 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
291-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize
292-- the message when the destination process resides on the same local
293-- node. Therefore, a local receiver would need to be prepared to cope with any
294-- errors resulting from evaluation of the message.
295unsafeUSend :: Serializable a => ProcessId -> a -> Process ()
296unsafeUSend = Unsafe.usend
298-- | Wait for a message of a specific type
299expect :: forall a. Serializable a => Process a
300expect = receiveWait [match return]
302--------------------------------------------------------------------------------
304--------------------------------------------------------------------------------
306-- | Create a new typed channel
307newChan :: Serializable a => Process (SendPort a, ReceivePort a)
310 liftIO . modifyMVar (processState proc) $ \st -> do
311 let lcid = st ^. channelCounter
312 let cid = SendPortId { sendPortProcessId = processId proc
313 , sendPortLocalId = lcid
315 let sport = SendPort cid
316 chan <- liftIO newTQueueIO
317 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
318 let rport = ReceivePort $ readTQueue chan
319 let tch = TypedChannel chan'
320 return ( (channelCounter ^: (+ 1))
321 . (typedChannelWithId lcid ^= Just tch)
326 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
327 finalizer st lcid = modifyMVar_ st $
328 return . (typedChannelWithId lcid ^= Nothing)
330-- | Send a message on a typed channel
331sendChan :: Serializable a => SendPort a -> a -> Process ()
332sendChan (SendPort cid) msg = do
334 let node = localNodeId (processNode proc)
335 destNode = processNodeId (sendPortProcessId cid) in do
336 case destNode == node of
337 True -> sendChanLocal cid msg
339 liftIO $ sendBinary (processNode proc)
340 (ProcessIdentifier (processId proc))
341 (SendPortIdentifier cid)
345-- | Send a message on a typed channel. This function makes /no/ attempt to
346-- serialize and (in the case when the @ReceivePort@ resides on the same local
347-- node) therefore ensure that the payload is fully evaluated before it is
349unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
350unsafeSendChan = Unsafe.sendChan
352-- | Wait for a message on a typed channel
353receiveChan :: Serializable a => ReceivePort a -> Process a
354receiveChan = liftIO . atomically . receiveSTM
356-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
357-- non-blocking check for a message.
358receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
359receiveChanTimeout 0 ch = liftIO . atomically $
360 (Just <$> receiveSTM ch) `orElse` return Nothing
361receiveChanTimeout n ch = liftIO . timeout n . atomically $
364-- | Merge a list of typed channels.
366-- The result port is left-biased: if there are messages available on more
367-- than one port, the first available message is returned.
368mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
369mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
371-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
373mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
374mergePortsRR = \ps -> do
375 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
376 return $ ReceivePort (rr psVar)
380 rotate (x:xs) = xs ++ [x]
382 rr :: TVar [STM a] -> STM a
385 a <- foldr1 orElse ps
386 writeTVar psVar (rotate ps)
389--------------------------------------------------------------------------------
390-- Advanced messaging --
391--------------------------------------------------------------------------------
393-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
394newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
397-- | Test the matches in order against each message in the queue
398receiveWait :: [Match b] -> Process b
400 queue <- processQueue <$> ask
401 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
404-- | Like 'receiveWait' but with a timeout.
406-- If the timeout is zero do a non-blocking check for matching messages. A
407-- non-zero timeout is applied only when waiting for incoming messages (that is,
408-- /after/ we have checked the messages that are already in the mailbox).
409receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
410receiveTimeout t ms = do
411 queue <- processQueue <$> ask
412 let blockSpec = if t == 0 then NonBlocking else Timeout t
413 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
415 Nothing -> return Nothing
416 Just proc -> Just <$> proc
418-- | Match on a typed channel
419matchChan :: ReceivePort a -> (a -> Process b) -> Match b
420matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
422-- | Match on an arbitrary STM action.
424-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
425-- actions with checks against our process' mailbox and/or any typed channel
426-- @ReceivePort@s we may hold.
428-- This allows us to process multiple input streams /along with our mailbox/,
429-- in just the same way that 'matchChan' supports checking both the mailbox
430-- /and/ an arbitrary set of typed channels in one atomic transaction.
432-- Note there are no ordering guarnatees with respect to these disparate input
435matchSTM :: STM a -> (a -> Process b) -> Match b
436matchSTM stm fn = Match $ MatchChan (fmap fn stm)
438-- | Match against any message of the right type
439match :: forall a b. Serializable a => (a -> Process b) -> Match b
440match = matchIf (const True)
442-- | Match against any message of the right type that satisfies a predicate
443matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
444matchIf c p = Match $ MatchMsg $ \msg ->
445 case messageFingerprint msg == fingerprint (undefined :: a) of
448 (UnencodedMessage _ m) ->
449 let m' = unsafeCoerce m :: a in
453 (EncodedMessage _ _) ->
454 if (c decoded) then Just (p decoded) else Nothing
457 -- Make sure the value is fully decoded so that we don't hang to
458 -- bytestrings when the process calling 'matchIf' doesn't process
459 -- the values immediately
460 !decoded = decode (messageEncoding msg)
462-- | Match against any message, regardless of the underlying (contained) type
463matchMessage :: (Message -> Process Message) -> Match Message
464matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
466-- | Match against any message (regardless of underlying type) that satisfies a predicate
467matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
468matchMessageIf c p = Match $ MatchMsg $ \msg ->
473-- | Forward a raw 'Message' to the given 'ProcessId'.
474forward :: Message -> ProcessId -> Process ()
477 let node = processNode proc
479 nid = localNodeId node
480 destNode = (processNodeId them) in do
481 case destNode == nid of
482 True -> sendCtrlMsg Nothing (LocalSend them msg)
483 False -> liftIO $ sendPayload (processNode proc)
484 (ProcessIdentifier (processId proc))
485 (ProcessIdentifier them)
487 (messageToPayload msg)
488 -- We do not fire the trace event until after the sending is complete;
489 -- In the remote case, 'sendMessage' can block in the networking stack.
490 liftIO $ traceEvent (localEventBus node)
493-- | Forward a raw 'Message' to the given 'ProcessId'.
495-- Unlike 'forward', this function is insensitive to 'reconnect'. It will
496-- try to send the message regardless of the history of connection failures
498uforward :: Message -> ProcessId -> Process ()
499uforward msg them = do
501 let node = processNode proc
503 nid = localNodeId node
504 destNode = (processNodeId them) in do
505 case destNode == nid of
506 True -> sendCtrlMsg Nothing (LocalSend them msg)
507 False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)
509 -- We do not fire the trace event until after the sending is complete;
510 -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
511 liftIO $ traceEvent (localEventBus node)
514-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
515-- 'Serializable' - like the datum they contain - but also note, deserialising
516-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
517-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
522-- send self (wrapMessage "blah")
523-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
524-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
525-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
528wrapMessage :: Serializable a => a -> Message
529wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
531-- | This is the /unsafe/ variant of 'wrapMessage'. See
532-- "Control.Distributed.Process.UnsafePrimitives" for details.
533unsafeWrapMessage :: Serializable a => a -> Message
534unsafeWrapMessage = Unsafe.wrapMessage
536-- [note Serializable UnencodedMessage]
537-- if we attempt to serialise an UnencodedMessage, it will be converted to an
538-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
539-- Primitives.hs for the gory details
541-- | Attempt to unwrap a raw 'Message'.
542-- If the type of the decoded message payload matches the expected type, the
543-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
546-- This expression, for example, will evaluate to @Nothing@
547-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
549-- Whereas this expression, will yield @Just "foo"@
550-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
552unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
554 case messageFingerprint msg == fingerprint (undefined :: a) of
555 False -> return Nothing :: m (Maybe a)
557 (UnencodedMessage _ ms) ->
558 let ms' = unsafeCoerce ms :: a
560 (EncodedMessage _ _) ->
561 return (Just (decoded))
563 decoded :: a -- note [decoding]
564 !decoded = decode (messageEncoding msg)
566-- | Attempt to handle a raw 'Message'.
567-- If the type of the message matches the type of the first argument to
568-- the supplied expression, then the message will be decoded and the expression
569-- evaluated against its value. If this runtime type checking fails however,
570-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
571-- evaluation proceeds, the resulting value with be wrapped with @Just@.
573-- Intended for use in `catchesExit` and `matchAny` primitives.
575handleMessage :: forall m a b. (Monad m, Serializable a)
576 => Message -> (a -> m b) -> m (Maybe b)
577handleMessage msg proc = handleMessageIf msg (const True) proc
579-- | Conditionally handle a raw 'Message'.
580-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
581-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
582-- for further information about runtime type checking.
584handleMessageIf :: forall m a b . (Monad m, Serializable a)
589handleMessageIf msg c proc = do
590 case messageFingerprint msg == fingerprint (undefined :: a) of
591 False -> return Nothing :: m (Maybe b)
593 (UnencodedMessage _ ms) ->
594 let ms' = unsafeCoerce ms :: a in
596 True -> do { r <- proc ms'; return (Just r) }
597 False -> return Nothing :: m (Maybe b)
598 (EncodedMessage _ _) ->
600 True -> do { r <- proc (decoded :: a); return (Just r) }
601 False -> return Nothing :: m (Maybe b)
603 decoded :: a -- note [decoding]
604 !decoded = decode (messageEncoding msg)
606-- | As 'handleMessage' but ignores result, which is useful if you don't
607-- care whether or not the handler succeeded.
609handleMessage_ :: forall m a . (Monad m, Serializable a)
610 => Message -> (a -> m ()) -> m ()
611handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
613-- | Conditional version of 'handleMessage_'.
615handleMessageIf_ :: forall m a . (Monad m, Serializable a)
620handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
622-- | Match against an arbitrary message. 'matchAny' removes the first available
623-- message from the process mailbox. To handle arbitrary /raw/ messages once
624-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
626matchAny :: forall b. (Message -> Process b) -> Match b
627matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
629-- | Match against an arbitrary message. Intended for use with 'handleMessage'
630-- and 'unwrapMessage', this function /only/ removes a message from the process
631-- mailbox, /if/ the supplied condition matches. The success (or failure) of
632-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
633-- here, i.e., if the condition evaluates to @True@ then the message will
634-- be removed from the process mailbox and decoded, but that does /not/
635-- guarantee that an expression passed to @handleMessage@ will pass the
636-- runtime type checks and therefore be evaluated.
638matchAnyIf :: forall a b. (Serializable a)
640 -> (Message -> Process b)
642matchAnyIf c p = Match $ MatchMsg $ \msg ->
643 case messageFingerprint msg == fingerprint (undefined :: a) of
644 True | check -> Just (p msg)
649 (EncodedMessage _ _) -> c decoded
650 (UnencodedMessage _ m') -> c (unsafeCoerce m')
652 decoded :: a -- note [decoding]
653 !decoded = decode (messageEncoding msg)
657For an EncodedMessage, we need to ensure the value is fully decoded so that
658we don't hang to bytestrings if the calling process doesn't evaluate
659immediately. For UnencodedMessage we know (because the fingerprint comparison
660succeeds) that unsafeCoerce will not fail.
663-- | Remove any message from the queue
664matchUnknown :: Process b -> Match b
665matchUnknown p = Match $ MatchMsg (const (Just p))
667-- | Receives messages and forwards them to @pid@ if @p msg == True@.
668delegate :: ProcessId -> (Message -> Bool) -> Process ()
671 matchAny (\m -> case (p m) of
672 True -> forward m pid
677-- | A straight relay that forwards all messages to the supplied pid.
678relay :: ProcessId -> Process ()
679relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
681-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
682-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
683-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
684-- fails, no action is taken and the proxy process will continue running.
685proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
689 next <- handleMessage m proc
691 Just True -> forward m pid
692 Just False -> return () -- explicitly ignored
693 Nothing -> return ()) -- un-routable
697--------------------------------------------------------------------------------
698-- Process management --
699--------------------------------------------------------------------------------
701-- | Thrown by 'terminate'
702data ProcessTerminationException = ProcessTerminationException
703 deriving (Show, Typeable)
705instance Exception ProcessTerminationException
707-- | Terminate immediately (throws a ProcessTerminationException)
708terminate :: Process a
709terminate = throwM ProcessTerminationException
712-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
713die :: Serializable a => a -> Process b
715-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
716-- could be decoded by a handler passed to 'catchExit', re-thrown or even
717-- passed to another process via the tracing mechanism.
719 throwM (ProcessExitException pid (createMessage reason))
721-- | Forceful request to kill a process. Where 'exit' provides an exception
722-- that can be caught and handled, 'kill' throws an unexposed exception type
723-- which cannot be handled explicitly (by type).
724kill :: ProcessId -> String -> Process ()
725-- NOTE: We send the message to our local node controller, which will then
726-- forward it to a remote node controller (if applicable). Sending it directly
727-- to a remote node controller means that that the message may overtake a
728-- 'monitor' or 'link' request.
729kill them reason = sendCtrlMsg Nothing (Kill them reason)
731-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
732-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
733-- manner can be handled using the 'catchExit' family of functions.
734exit :: Serializable a => ProcessId -> a -> Process ()
735-- NOTE: We send the message to our local node controller, which will then
736-- forward it to a remote node controller (if applicable). Sending it directly
737-- to a remote node controller means that that the message may overtake a
738-- 'monitor' or 'link' request.
739exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
741-- | Catches 'ProcessExitException'. The handler will not be applied unless its
742-- type matches the encoded data stored in the exception (see the /reason/
743-- argument given to the 'exit' primitive). If the handler cannot be applied,
744-- the exception will be re-thrown.
746-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
747-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
748catchExit :: forall a b . (Show a, Serializable a)
750 -> (ProcessId -> a -> Process b)
752catchExit act exitHandler = Catch.catch act handleExit
754 handleExit ex@(ProcessExitException from msg) =
755 if messageFingerprint msg == fingerprint (undefined :: a)
756 then exitHandler from decoded
760 -- Make sure the value is fully decoded so that we don't hang to
761 -- bytestrings if the caller doesn't use the value immediately
762 !decoded = decode (messageEncoding msg)
764-- | Lift 'Control.Exception.catches' (almost).
766-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
767-- message, a handler must accept inputs of the expected type. In order to
768-- handle a list of potentially different handlers (and therefore input types),
769-- a handler passed to 'catchesExit' must accept 'Message' and return
770-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
772-- See 'maybeHandleMessage' and 'Message' for more details.
773catchesExit :: Process b
774 -> [(ProcessId -> Message -> (Process (Maybe b)))]
776catchesExit act handlers = Catch.catch act ((flip handleExit) handlers)
778 handleExit :: ProcessExitException
779 -> [(ProcessId -> Message -> Process (Maybe b))]
781 handleExit ex [] = throwM ex
782 handleExit ex@(ProcessExitException from msg) (h:hs) = do
785 Nothing -> handleExit ex hs
788-- | Our own process ID
789getSelfPid :: Process ProcessId
790getSelfPid = processId <$> ask
792-- | Get the node ID of our local node
793getSelfNode :: Process NodeId
794getSelfNode = localNodeId . processNode <$> ask
797-- | Get statistics about the specified node
798getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
800 selfNode <- getSelfNode
802 then Right `fmap` getLocalNodeStats -- optimisation
803 else getNodeStatsRemote selfNode
805 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
806 getNodeStatsRemote selfNode = do
807 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
808 bracket (monitorNode nid) unmonitor $ \mRef ->
809 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
810 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
811 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
814-- | Get statistics about our local node
815getLocalNodeStats :: Process NodeStats
816getLocalNodeStats = do
818 sendCtrlMsg Nothing $ GetNodeStats self
819 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
821-- | Get information about the specified process
822getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
824 let them = processNodeId pid in do
826 dest <- mkNode them us
827 sendCtrlMsg dest $ GetInfo pid
829 match (\(p :: ProcessInfo) -> return $ Just p)
830 , match (\(_ :: ProcessInfoNone) -> return Nothing)
832 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
833 mkNode them us = case them == us of
834 True -> return Nothing
835 _ -> return $ Just them
837--------------------------------------------------------------------------------
838-- Monitoring and linking --
839--------------------------------------------------------------------------------
841-- | Link to a remote process (asynchronous)
843-- When process A links to process B (that is, process A calls
844-- @link pidB@) then an asynchronous exception will be thrown to process A
845-- when process B terminates (normally or abnormally), or when process A gets
846-- disconnected from process B. Although it is /technically/ possible to catch
847-- these exceptions, chances are if you find yourself trying to do so you should
848-- probably be using 'monitor' rather than 'link'. In particular, code such as
850-- > link pidB -- Link to process B
851-- > expect -- Wait for a message from process B
852-- > unlink pidB -- Unlink again
854-- doesn't quite do what one might expect: if process B sends a message to
855-- process A, and /subsequently terminates/, then process A might or might not
856-- be terminated too, depending on whether the exception is thrown before or
857-- after the 'unlink' (i.e., this code has a race condition).
859-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
860-- call to 'link' has no effect.
862-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
863-- Linking makes no distinction between normal and abnormal termination of
864-- the remote process.
865link :: ProcessId -> Process ()
866link = sendCtrlMsg Nothing . Link . ProcessIdentifier
868-- | Monitor another process (asynchronous)
870-- When process A monitors process B (that is, process A calls
871-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
872-- when process B terminates (normally or abnormally), or when process A gets
873-- disconnected from process B. You receive this message like any other (using
874-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
875-- 'DiedDisconnect', etc.).
877-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
878-- multiple monitors are set up, multiple notifications will be delivered
879-- and monitors can be disabled individually using 'unmonitor'.
880monitor :: ProcessId -> Process MonitorRef
881monitor = monitor' . ProcessIdentifier
883-- | Establishes temporary monitoring of another process.
885-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
886-- of @code@. Note: although monitoring is no longer active when
887-- @withMonitor@ returns, there might still be unreceived monitor
888-- messages in the queue.
890withMonitor :: ProcessId -> Process a -> Process a
891withMonitor pid = withMonitorRef pid . const
892 -- unmonitor blocks waiting for the response, so there's a possibility
893 -- that an exception might interrupt withMonitor before the unmonitor
894 -- has completed. I think that's better than making the unmonitor
897-- | Establishes temporary monitoring of another process.
899-- @withMonitorRef pid code@ sets up monitoring of @pid@ for the duration
900-- of @code@. Note: although monitoring is no longer active when
901-- @withMonitorRef@ returns, there might still be unreceived monitor
902-- messages in the queue.
903withMonitorRef :: ProcessId -> (MonitorRef -> Process a) -> Process a
904withMonitorRef pid code = bracket (monitor pid) unmonitor code
908-- This is synchronous in the sense that once it returns you are guaranteed
909-- that no exception will be raised if the remote process dies. However, it is
910-- asynchronous in the sense that we do not wait for a response from the remote
912unlink :: ProcessId -> Process ()
915 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
919-- | Remove a node link
921-- This has the same synchronous/asynchronous nature as 'unlink'.
922unlinkNode :: NodeId -> Process ()
925 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
929-- | Remove a channel (send port) link
931-- This has the same synchronous/asynchronous nature as 'unlink'.
932unlinkPort :: SendPort a -> Process ()
934 unlinkPortAsync sport
935 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
941-- This has the same synchronous/asynchronous nature as 'unlink'.
943-- ProcessMonitorNotification messages for the given MonitorRef are removed from
945unmonitor :: MonitorRef -> Process ()
948 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
951 -- Discard the notification if any. With the current NC implementation at most
952 -- one notification is in the mailbox for any given ref.
953 void $ receiveTimeout 0
954 [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref' == ref)
958--------------------------------------------------------------------------------
959-- Exception handling --
960--------------------------------------------------------------------------------
962-- | Lift 'Control.Exception.catch'
963catch :: Exception e => Process a -> (e -> Process a) -> Process a
965{-# DEPRECATED catch "Use Control.Monad.Catch.catch instead" #-}
967-- | Lift 'Control.Exception.try'
968try :: Exception e => Process a -> Process (Either e a)
970{-# DEPRECATED try "Use Control.Monad.Catch.mask_ instead" #-}
972-- | Lift 'Control.Exception.mask'
973mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
975{-# DEPRECATED mask "Use Control.Monad.Catch.mask_ instead" #-}
977-- | Lift 'Control.Exception.mask_'
978mask_ :: Process a -> Process a
980{-# DEPRECATED mask_ "Use Control.Monad.Catch.mask_ instead" #-}
982-- | Lift 'Control.Exception.onException'
983onException :: Process a -> Process b -> Process a
984onException = Catch.onException
985{-# DEPRECATED onException "Use Control.Monad.Catch.onException instead" #-}
987-- | Lift 'Control.Exception.bracket'
988bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
989bracket = Catch.bracket
990{-# DEPRECATED bracket "Use Control.Monad.Catch.bracket instead" #-}
992-- | Lift 'Control.Exception.bracket_'
993bracket_ :: Process a -> Process b -> Process c -> Process c
994bracket_ = Catch.bracket_
995{-# DEPRECATED bracket_ "Use Control.Monad.Catch.bracket_ instead" #-}
997-- | Lift 'Control.Exception.finally'
998finally :: Process a -> Process b -> Process a
999finally = Catch.finally
1000{-# DEPRECATED finally "Use Control.Monad.Catch.finally instead" #-}
1002-- | You need this when using 'catches'
1003data Handler a = forall e . Exception e => Handler (e -> Process a)
1005instance Functor Handler where
1006 fmap f (Handler h) = Handler (fmap f . h)
1008-- | Lift 'Control.Exception.catches'
1009catches :: Process a -> [Handler a] -> Process a
1010catches proc handlers = proc `Catch.catch` catchesHandler handlers
1012catchesHandler :: [Handler a] -> SomeException -> Process a
1013catchesHandler handlers e = foldr tryHandler (throwM e) handlers
1014 where tryHandler (Handler handler) res
1015 = case fromException e of
1016 Just e' -> handler e'
1019--------------------------------------------------------------------------------
1021--------------------------------------------------------------------------------
1023-- | Like 'expect' but with a timeout
1024expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
1025expectTimeout n = receiveTimeout n [match return]
1027-- | Asynchronous version of 'spawn'
1029-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
1030spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
1031spawnAsync nid proc = do
1032 spawnRef <- getSpawnRef
1035 then sendCtrlMsg Nothing $ Spawn proc spawnRef
1036 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
1039-- | Monitor a node (asynchronous)
1040monitorNode :: NodeId -> Process MonitorRef
1042 monitor' . NodeIdentifier
1044-- | Monitor a typed channel (asynchronous)
1045monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1046monitorPort (SendPort cid) =
1047 monitor' (SendPortIdentifier cid)
1049-- | Remove a monitor (asynchronous)
1050unmonitorAsync :: MonitorRef -> Process ()
1052 sendCtrlMsg Nothing . Unmonitor
1054-- | Link to a node (asynchronous)
1055linkNode :: NodeId -> Process ()
1056linkNode = link' . NodeIdentifier
1058-- | Link to a channel (asynchronous)
1059linkPort :: SendPort a -> Process ()
1060linkPort (SendPort cid) =
1061 link' (SendPortIdentifier cid)
1063-- | Remove a process link (asynchronous)
1064unlinkAsync :: ProcessId -> Process ()
1066 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1068-- | Remove a node link (asynchronous)
1069unlinkNodeAsync :: NodeId -> Process ()
1071 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1073-- | Remove a channel (send port) link (asynchronous)
1074unlinkPortAsync :: SendPort a -> Process ()
1075unlinkPortAsync (SendPort cid) =
1076 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1078--------------------------------------------------------------------------------
1080--------------------------------------------------------------------------------
1084-- @say message@ sends a message (time, pid of the current process, message)
1085-- to the process registered as 'logger'. By default, this process simply
1086-- sends the string to 'stderr'. Individual Cloud Haskell backends might
1087-- replace this with a different logger process, however.
1088say :: String -> Process ()
1090 now <- liftIO getCurrentTime
1092 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
1094--------------------------------------------------------------------------------
1096--------------------------------------------------------------------------------
1098-- | Register a process with the local registry (synchronous). The name must not
1099-- already be registered. The process need not be on this node. A bad
1100-- registration will result in a 'ProcessRegistrationException'
1102-- The process to be registered does not have to be local itself.
1103register :: String -> ProcessId -> Process ()
1104register = registerImpl False
1106-- | Like 'register', but will replace an existing registration.
1107-- The name must already be registered.
1108reregister :: String -> ProcessId -> Process ()
1109reregister = registerImpl True
1111registerImpl :: Bool -> String -> ProcessId -> Process ()
1112registerImpl force label pid = do
1113 mynid <- getSelfNode
1114 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1116 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1117 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1120-- | Register a process with a remote registry (asynchronous).
1122-- The process to be registered does not have to live on the same remote node.
1123-- Reply wil come in the form of a 'RegisterReply' message
1125-- See comments in 'whereisRemoteAsync'
1126registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1127registerRemoteAsync nid label pid = do
1129 sendCtrlMsg (if nid == here then Nothing else Just nid)
1130 (Register label nid (Just pid) False)
1132reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1133reregisterRemoteAsync nid label pid =
1134 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1136-- | Remove a process from the local registry (asynchronous).
1137-- This version will wait until a response is gotten from the
1138-- management process. The name must already be registered.
1139unregister :: String -> Process ()
1140unregister label = do
1141 mynid <- getSelfNode
1142 sendCtrlMsg Nothing (Register label mynid Nothing False)
1144 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1145 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1148-- | Deal with the result from an attempted registration or unregistration
1149-- by throwing an exception if necessary
1150handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()
1151handleRegistrationReply label ok owner =
1153 throwM $ ProcessRegistrationException label owner
1155-- | Remove a process from a remote registry (asynchronous).
1157-- Reply wil come in the form of a 'RegisterReply' message
1159-- See comments in 'whereisRemoteAsync'
1160unregisterRemoteAsync :: NodeId -> String -> Process ()
1161unregisterRemoteAsync nid label =
1162 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1164-- | Query the local process registry
1165whereis :: String -> Process (Maybe ProcessId)
1167 sendCtrlMsg Nothing (WhereIs label)
1168 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1169 (\(WhereIsReply _ mPid) -> return mPid)
1172-- | Query a remote process registry (asynchronous)
1174-- Reply will come in the form of a 'WhereIsReply' message.
1176-- There is currently no synchronous version of 'whereisRemoteAsync': if
1177-- you implement one yourself, be sure to take into account that the remote
1178-- node might die or get disconnect before it can respond (i.e. you should
1179-- use 'monitorNode' and take appropriate action when you receive a
1180-- 'NodeMonitorNotification').
1181whereisRemoteAsync :: NodeId -> String -> Process ()
1182whereisRemoteAsync nid label = do
1184 sendCtrlMsg (if nid == here then Nothing else Just nid) (WhereIs label)
1186-- | Named send to a process in the local registry (asynchronous)
1187nsend :: Serializable a => String -> a -> Process ()
1189 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1191-- | Named send to a process in the local registry (asynchronous).
1192-- This function makes /no/ attempt to serialize and (in the case when the
1193-- destination process resides on the same local node) therefore ensure that
1194-- the payload is fully evaluated before it is delivered.
1195unsafeNSend :: Serializable a => String -> a -> Process ()
1196unsafeNSend = Unsafe.nsend
1198-- | Named send to a process in a remote registry (asynchronous)
1199nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1200nsendRemote nid label msg = do
1202 if here == nid then nsend label msg
1203 else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1205-- | Named send to a process in a remote registry (asynchronous)
1206-- This function makes /no/ attempt to serialize and (in the case when the
1207-- destination process resides on the same local node) therefore ensure that
1208-- the payload is fully evaluated before it is delivered.
1209unsafeNSendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1210unsafeNSendRemote = Unsafe.nsendRemote
1212--------------------------------------------------------------------------------
1214--------------------------------------------------------------------------------
1216-- | Resolve a static value
1217unStatic :: Typeable a => Static a -> Process a
1219 rtable <- remoteTable . processNode <$> ask
1220 case Static.unstatic rtable static of
1221 Left err -> fail $ "Could not resolve static value: " ++ err
1224-- | Resolve a closure
1225unClosure :: Typeable a => Closure a -> Process a
1226unClosure closure = do
1227 rtable <- remoteTable . processNode <$> ask
1228 case Static.unclosure rtable closure of
1229 Left err -> fail $ "Could not resolve closure: " ++ err
1232--------------------------------------------------------------------------------
1234--------------------------------------------------------------------------------
1236-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1237-- message passing. However, when network connections get disrupted this
1238-- illusion cannot always be maintained. Once a network connection breaks (even
1239-- temporarily) no further communication on that connection will be possible.
1240-- For example, if process A sends a message to process B, and A is then
1241-- notified (by monitor notification) that it got disconnected from B, A will
1242-- not be able to send any further messages to B, /unless/ A explicitly
1243-- indicates that it is acceptable to attempt to reconnect to B using the
1244-- Cloud Haskell 'reconnect' primitive.
1246-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1247-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1248-- then receives a monitor notification that its connection to B has been lost,
1249-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1250-- and m3 but not m2.
1252-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1253-- to attempt to reconnect on the next send/. In particular, if no further
1254-- communication attempts are made to B then A can use reconnect to clean up
1255-- its connection to B.
1256reconnect :: ProcessId -> Process ()
1259 node <- processNode <$> ask
1260 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1262-- | Reconnect to a sendport. See 'reconnect' for more information.
1263reconnectPort :: SendPort a -> Process ()
1264reconnectPort them = do
1266 node <- processNode <$> ask
1267 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1269--------------------------------------------------------------------------------
1270-- Auxiliary functions --
1271--------------------------------------------------------------------------------
1273sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1275 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1277sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1278sendChanLocal spId msg =
1279 -- we *must* fully serialize/encode the message here, because
1280 -- attempting to use `unsafeCoerce' in the node controller
1281 -- won't work since we know nothing about the required type
1282 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1284getMonitorRefFor :: Identifier -> Process MonitorRef
1285getMonitorRefFor ident = do
1287 liftIO $ modifyMVar (processState proc) $ \st -> do
1288 let counter = st ^. monitorCounter
1289 return ( monitorCounter ^: (+ 1) $ st
1290 , MonitorRef ident counter
1293getSpawnRef :: Process SpawnRef
1296 liftIO $ modifyMVar (processState proc) $ \st -> do
1297 let counter = st ^. spawnCounter
1298 return ( spawnCounter ^: (+ 1) $ st
1302-- | Monitor a process/node/channel
1303monitor' :: Identifier -> Process MonitorRef
1305 monitorRef <- getMonitorRefFor ident
1306 sendCtrlMsg Nothing $ Monitor monitorRef
1309-- | Link to a process/node/channel
1310link' :: Identifier -> Process ()
1311link' = 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