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
91 , reregisterRemoteAsync
92 , unregisterRemoteAsync
98 -- * Exception handling
122 -- * Internal Exports
126#if ! MIN_VERSION_base(4,6,0)
127import Prelude hiding (catch)
130import Data.Binary (Binary(..), Put, Get, decode)
131import Data.Time.Clock (getCurrentTime, UTCTime(..))
132import Data.Time.Calendar (Day(..))
133import Data.Time.Format (formatTime)
134#if MIN_VERSION_time(1,5,0)
135import Data.Time.Format (defaultTimeLocale)
137import System.Locale (defaultTimeLocale)
139import System.Timeout (timeout)
140import Control.Monad (when, void)
141import Control.Monad.Reader (ask)
142import Control.Monad.IO.Class (liftIO)
143import Control.Monad.Catch
149import qualified Control.Monad.Catch as Catch
150import Control.Applicative
151import Control.Distributed.Process.Internal.StrictMVar
156import Control.Concurrent.STM
165import Control.Distributed.Process.Internal.CQueue
170import Control.Distributed.Process.Serializable (Serializable, fingerprint)
171import Data.Accessor ((^.), (^:), (^=))
172import Control.Distributed.Static
176import Data.Rank1Typeable (Typeable)
177import qualified Control.Distributed.Static as Static (unstatic, unclosure)
178import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
179import Control.Distributed.Process.Internal.Types
189 , NodeMonitorNotification(..)
190 , ProcessMonitorNotification(..)
200 , ProcessExitException(..)
203 , DidUnlinkProcess(..)
208 , ProcessRegistrationException(..)
210 , ProcessInfoNone(..)
214 , createUnencodedMessage
215 , ImplicitReconnect( NoImplicitReconnect)
220import Control.Distributed.Process.Internal.Messaging
227import Control.Distributed.Process.Management.Internal.Types
230import Control.Distributed.Process.Management.Internal.Trace.Types
233import Control.Distributed.Process.Internal.WeakTQueue
242--------------------------------------------------------------------------------
244--------------------------------------------------------------------------------
247send :: Serializable a => ProcessId -> a -> Process ()
248-- This requires a lookup on every send. If we want to avoid that we need to
249-- modify serializable to allow for stateful (IO) deserialization.
252 let us = processId proc
253 node = processNode proc
254 nodeId = localNodeId node
255 destNode = (processNodeId them) in do
256 case destNode == nodeId of
257 True -> sendLocal them msg
258 False -> liftIO $ sendMessage (processNode proc)
259 (ProcessIdentifier (processId proc))
260 (ProcessIdentifier them)
263 -- We do not fire the trace event until after the sending is complete;
264 -- In the remote case, 'sendMessage' can block in the networking stack.
265 liftIO $ traceEvent (localEventBus node)
266 (MxSent them us (createUnencodedMessage msg))
268-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
269-- and (in the case when the destination process resides on the same local
270-- node) therefore ensure that the payload is fully evaluated before it is
272unsafeSend :: Serializable a => ProcessId -> a -> Process ()
273unsafeSend = Unsafe.send
275-- | Send a message unreliably.
277-- Unlike 'send', this function is insensitive to 'reconnect'. It will
278-- try to send the message regardless of the history of connection failures
281-- Message passing with 'usend' is ordered for a given sender and receiver
282-- if the messages arrive at all.
284usend :: Serializable a => ProcessId -> a -> Process ()
287 let there = processNodeId them
289 then sendLocal them msg
290 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
293-- | /Unsafe/ variant of 'usend'. This function makes /no/ attempt to serialize
294-- the message when the destination process resides on the same local
295-- node. Therefore, a local receiver would need to be prepared to cope with any
296-- errors resulting from evaluation of the message.
297unsafeUSend :: Serializable a => ProcessId -> a -> Process ()
298unsafeUSend = Unsafe.usend
300-- | Wait for a message of a specific type
301expect :: forall a. Serializable a => Process a
302expect = receiveWait [match return]
304--------------------------------------------------------------------------------
306--------------------------------------------------------------------------------
308-- | Create a new typed channel
309newChan :: Serializable a => Process (SendPort a, ReceivePort a)
312 liftIO . modifyMVar (processState proc) $ \st -> do
313 let lcid = st ^. channelCounter
314 let cid = SendPortId { sendPortProcessId = processId proc
315 , sendPortLocalId = lcid
317 let sport = SendPort cid
318 chan <- liftIO newTQueueIO
319 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
320 let rport = ReceivePort $ readTQueue chan
321 let tch = TypedChannel chan'
322 return ( (channelCounter ^: (+ 1))
323 . (typedChannelWithId lcid ^= Just tch)
328 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
329 finalizer st lcid = modifyMVar_ st $
330 return . (typedChannelWithId lcid ^= Nothing)
332-- | Send a message on a typed channel
333sendChan :: Serializable a => SendPort a -> a -> Process ()
334sendChan (SendPort cid) msg = do
336 let node = localNodeId (processNode proc)
337 destNode = processNodeId (sendPortProcessId cid) in do
338 case destNode == node of
339 True -> sendChanLocal cid msg
341 liftIO $ sendBinary (processNode proc)
342 (ProcessIdentifier (processId proc))
343 (SendPortIdentifier cid)
347-- | Send a message on a typed channel. This function makes /no/ attempt to
348-- serialize and (in the case when the @ReceivePort@ resides on the same local
349-- node) therefore ensure that the payload is fully evaluated before it is
351unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
352unsafeSendChan = Unsafe.sendChan
354-- | Wait for a message on a typed channel
355receiveChan :: Serializable a => ReceivePort a -> Process a
356receiveChan = liftIO . atomically . receiveSTM
358-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
359-- non-blocking check for a message.
360receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
361receiveChanTimeout 0 ch = liftIO . atomically $
362 (Just <$> receiveSTM ch) `orElse` return Nothing
363receiveChanTimeout n ch = liftIO . timeout n . atomically $
366-- | Merge a list of typed channels.
368-- The result port is left-biased: if there are messages available on more
369-- than one port, the first available message is returned.
370mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
371mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
373-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
375mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
376mergePortsRR = \ps -> do
377 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
378 return $ ReceivePort (rr psVar)
382 rotate (x:xs) = xs ++ [x]
384 rr :: TVar [STM a] -> STM a
387 a <- foldr1 orElse ps
388 writeTVar psVar (rotate ps)
391--------------------------------------------------------------------------------
392-- Advanced messaging --
393--------------------------------------------------------------------------------
395-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
396newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
399-- | Test the matches in order against each message in the queue
400receiveWait :: [Match b] -> Process b
402 queue <- processQueue <$> ask
403 mProc <- liftIO $ dequeue queue Blocking (map unMatch ms)
406 Nothing -> die $ "System Invariant Violation: CQueue.hs returned `Nothing` "
407 ++ "in the absence of a timeout value."
409-- | Like 'receiveWait' but with a timeout.
411-- If the timeout is zero do a non-blocking check for matching messages. A
412-- non-zero timeout is applied only when waiting for incoming messages (that is,
413-- /after/ we have checked the messages that are already in the mailbox).
414receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
415receiveTimeout t ms = do
416 queue <- processQueue <$> ask
417 let blockSpec = if t == 0 then NonBlocking else Timeout t
418 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
420 Nothing -> return Nothing
421 Just proc -> Just <$> proc
423-- | Match on a typed channel
424matchChan :: ReceivePort a -> (a -> Process b) -> Match b
425matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
427-- | Match on an arbitrary STM action.
429-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
430-- actions with checks against our process' mailbox and/or any typed channel
431-- @ReceivePort@s we may hold.
433-- This allows us to process multiple input streams /along with our mailbox/,
434-- in just the same way that 'matchChan' supports checking both the mailbox
435-- /and/ an arbitrary set of typed channels in one atomic transaction.
437-- Note there are no ordering guarnatees with respect to these disparate input
440matchSTM :: STM a -> (a -> Process b) -> Match b
441matchSTM stm fn = Match $ MatchChan (fmap fn stm)
443-- | Match against any message of the right type
444match :: forall a b. Serializable a => (a -> Process b) -> Match b
445match = matchIf (const True)
447-- | Match against any message of the right type that satisfies a predicate
448matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
449matchIf c p = Match $ MatchMsg $ \msg ->
450 case messageFingerprint msg == fingerprint (undefined :: a) of
453 (UnencodedMessage _ m) ->
454 let m' = unsafeCoerce m :: a in
458 (EncodedMessage _ _) ->
459 if (c decoded) then Just (p decoded) else Nothing
462 -- Make sure the value is fully decoded so that we don't hang to
463 -- bytestrings when the process calling 'matchIf' doesn't process
464 -- the values immediately
465 !decoded = decode (messageEncoding msg)
467-- | Match against any message, regardless of the underlying (contained) type
468matchMessage :: (Message -> Process Message) -> Match Message
469matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
471-- | Match against any message (regardless of underlying type) that satisfies a predicate
472matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
473matchMessageIf c p = Match $ MatchMsg $ \msg ->
478-- | Forward a raw 'Message' to the given 'ProcessId'.
479forward :: Message -> ProcessId -> Process ()
482 let node = processNode proc
484 nid = localNodeId node
485 destNode = (processNodeId them) in do
486 case destNode == nid of
487 True -> sendCtrlMsg Nothing (LocalSend them msg)
488 False -> liftIO $ sendPayload (processNode proc)
489 (ProcessIdentifier (processId proc))
490 (ProcessIdentifier them)
492 (messageToPayload msg)
493 -- We do not fire the trace event until after the sending is complete;
494 -- In the remote case, 'sendMessage' can block in the networking stack.
495 liftIO $ traceEvent (localEventBus node)
498-- | Forward a raw 'Message' to the given 'ProcessId'.
500-- Unlike 'forward', this function is insensitive to 'reconnect'. It will
501-- try to send the message regardless of the history of connection failures
503uforward :: Message -> ProcessId -> Process ()
504uforward msg them = do
506 let node = processNode proc
508 nid = localNodeId node
509 destNode = (processNodeId them) in do
510 case destNode == nid of
511 True -> sendCtrlMsg Nothing (LocalSend them msg)
512 False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)
514 -- We do not fire the trace event until after the sending is complete;
515 -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
516 liftIO $ traceEvent (localEventBus node)
519-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
520-- 'Serializable' - like the datum they contain - but also note, deserialising
521-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
522-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
527-- send self (wrapMessage "blah")
528-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
529-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
530-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
533wrapMessage :: Serializable a => a -> Message
534wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
536-- | This is the /unsafe/ variant of 'wrapMessage'. See
537-- "Control.Distributed.Process.UnsafePrimitives" for details.
538unsafeWrapMessage :: Serializable a => a -> Message
539unsafeWrapMessage = Unsafe.wrapMessage
541-- [note Serializable UnencodedMessage]
542-- if we attempt to serialise an UnencodedMessage, it will be converted to an
543-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
544-- Primitives.hs for the gory details
546-- | Attempt to unwrap a raw 'Message'.
547-- If the type of the decoded message payload matches the expected type, the
548-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
551-- This expression, for example, will evaluate to @Nothing@
552-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
554-- Whereas this expression, will yield @Just "foo"@
555-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
557unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
559 case messageFingerprint msg == fingerprint (undefined :: a) of
560 False -> return Nothing :: m (Maybe a)
562 (UnencodedMessage _ ms) ->
563 let ms' = unsafeCoerce ms :: a
565 (EncodedMessage _ _) ->
566 return (Just (decoded))
568 decoded :: a -- note [decoding]
569 !decoded = decode (messageEncoding msg)
571-- | Attempt to handle a raw 'Message'.
572-- If the type of the message matches the type of the first argument to
573-- the supplied expression, then the message will be decoded and the expression
574-- evaluated against its value. If this runtime type checking fails however,
575-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
576-- evaluation proceeds, the resulting value with be wrapped with @Just@.
578-- Intended for use in `catchesExit` and `matchAny` primitives.
580handleMessage :: forall m a b. (Monad m, Serializable a)
581 => Message -> (a -> m b) -> m (Maybe b)
582handleMessage msg proc = handleMessageIf msg (const True) proc
584-- | Conditionally handle a raw 'Message'.
585-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
586-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
587-- for further information about runtime type checking.
589handleMessageIf :: forall m a b . (Monad m, Serializable a)
594handleMessageIf msg c proc = do
595 case messageFingerprint msg == fingerprint (undefined :: a) of
596 False -> return Nothing :: m (Maybe b)
598 (UnencodedMessage _ ms) ->
599 let ms' = unsafeCoerce ms :: a in
601 True -> do { r <- proc ms'; return (Just r) }
602 False -> return Nothing :: m (Maybe b)
603 (EncodedMessage _ _) ->
605 True -> do { r <- proc (decoded :: a); return (Just r) }
606 False -> return Nothing :: m (Maybe b)
608 decoded :: a -- note [decoding]
609 !decoded = decode (messageEncoding msg)
611-- | As 'handleMessage' but ignores result, which is useful if you don't
612-- care whether or not the handler succeeded.
614handleMessage_ :: forall m a . (Monad m, Serializable a)
615 => Message -> (a -> m ()) -> m ()
616handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
618-- | Conditional version of 'handleMessage_'.
620handleMessageIf_ :: forall m a . (Monad m, Serializable a)
625handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
627-- | Match against an arbitrary message. 'matchAny' removes the first available
628-- message from the process mailbox. To handle arbitrary /raw/ messages once
629-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
631matchAny :: forall b. (Message -> Process b) -> Match b
632matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
634-- | Match against an arbitrary message. Intended for use with 'handleMessage'
635-- and 'unwrapMessage', this function /only/ removes a message from the process
636-- mailbox, /if/ the supplied condition matches. The success (or failure) of
637-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
638-- here, i.e., if the condition evaluates to @True@ then the message will
639-- be removed from the process mailbox and decoded, but that does /not/
640-- guarantee that an expression passed to @handleMessage@ will pass the
641-- runtime type checks and therefore be evaluated.
643matchAnyIf :: forall a b. (Serializable a)
645 -> (Message -> Process b)
647matchAnyIf c p = Match $ MatchMsg $ \msg ->
648 case messageFingerprint msg == fingerprint (undefined :: a) of
649 True | check -> Just (p msg)
654 (EncodedMessage _ _) -> c decoded
655 (UnencodedMessage _ m') -> c (unsafeCoerce m')
657 decoded :: a -- note [decoding]
658 !decoded = decode (messageEncoding msg)
662For an EncodedMessage, we need to ensure the value is fully decoded so that
663we don't hang to bytestrings if the calling process doesn't evaluate
664immediately. For UnencodedMessage we know (because the fingerprint comparison
665succeeds) that unsafeCoerce will not fail.
668-- | Remove any message from the queue
669matchUnknown :: Process b -> Match b
670matchUnknown p = Match $ MatchMsg (const (Just p))
672-- | Receives messages and forwards them to @pid@ if @p msg == True@.
673delegate :: ProcessId -> (Message -> Bool) -> Process ()
676 matchAny (\m -> case (p m) of
677 True -> forward m pid
682-- | A straight relay that forwards all messages to the supplied pid.
683relay :: ProcessId -> Process ()
684relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
686-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
687-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
688-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
689-- fails, no action is taken and the proxy process will continue running.
690proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
694 next <- handleMessage m proc
696 Just True -> forward m pid
697 Just False -> return () -- explicitly ignored
698 Nothing -> return ()) -- un-routable
702--------------------------------------------------------------------------------
703-- Process management --
704--------------------------------------------------------------------------------
706-- | Thrown by 'terminate'
707data ProcessTerminationException = ProcessTerminationException
708 deriving (Show, Typeable)
710instance Exception ProcessTerminationException
712-- | Terminate immediately (throws a ProcessTerminationException)
713terminate :: Process a
714terminate = throwM ProcessTerminationException
717-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
718die :: Serializable a => a -> Process b
720-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
721-- could be decoded by a handler passed to 'catchExit', re-thrown or even
722-- passed to another process via the tracing mechanism.
724 throwM (ProcessExitException pid (createMessage reason))
726-- | Forceful request to kill a process. Where 'exit' provides an exception
727-- that can be caught and handled, 'kill' throws an unexposed exception type
728-- which cannot be handled explicitly (by type).
729kill :: ProcessId -> String -> Process ()
730-- NOTE: We send the message to our local node controller, which will then
731-- forward it to a remote node controller (if applicable). Sending it directly
732-- to a remote node controller means that that the message may overtake a
733-- 'monitor' or 'link' request.
734kill them reason = sendCtrlMsg Nothing (Kill them reason)
736-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
737-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
738-- manner can be handled using the 'catchExit' family of functions.
739exit :: Serializable a => ProcessId -> a -> Process ()
740-- NOTE: We send the message to our local node controller, which will then
741-- forward it to a remote node controller (if applicable). Sending it directly
742-- to a remote node controller means that that the message may overtake a
743-- 'monitor' or 'link' request.
744exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
746-- | Catches 'ProcessExitException'. The handler will not be applied unless its
747-- type matches the encoded data stored in the exception (see the /reason/
748-- argument given to the 'exit' primitive). If the handler cannot be applied,
749-- the exception will be re-thrown.
751-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
752-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
753catchExit :: forall a b . (Show a, Serializable a)
755 -> (ProcessId -> a -> Process b)
757catchExit act exitHandler = Catch.catch act handleExit
759 handleExit ex@(ProcessExitException from msg) =
760 if messageFingerprint msg == fingerprint (undefined :: a)
761 then exitHandler from decoded
765 -- Make sure the value is fully decoded so that we don't hang to
766 -- bytestrings if the caller doesn't use the value immediately
767 !decoded = decode (messageEncoding msg)
769-- | Lift 'Control.Exception.catches' (almost).
771-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
772-- message, a handler must accept inputs of the expected type. In order to
773-- handle a list of potentially different handlers (and therefore input types),
774-- a handler passed to 'catchesExit' must accept 'Message' and return
775-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
777-- See 'maybeHandleMessage' and 'Message' for more details.
778catchesExit :: Process b
779 -> [(ProcessId -> Message -> (Process (Maybe b)))]
781catchesExit act handlers = Catch.catch act ((flip handleExit) handlers)
783 handleExit :: ProcessExitException
784 -> [(ProcessId -> Message -> Process (Maybe b))]
786 handleExit ex [] = throwM ex
787 handleExit ex@(ProcessExitException from msg) (h:hs) = do
790 Nothing -> handleExit ex hs
793-- | Our own process ID
794getSelfPid :: Process ProcessId
795getSelfPid = processId <$> ask
797-- | Get the node ID of our local node
798getSelfNode :: Process NodeId
799getSelfNode = localNodeId . processNode <$> ask
802-- | Get statistics about the specified node
803getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
805 selfNode <- getSelfNode
807 then Right `fmap` getLocalNodeStats -- optimisation
808 else getNodeStatsRemote selfNode
810 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
811 getNodeStatsRemote selfNode = do
812 sendCtrlMsg (Just nid) $ GetNodeStats selfNode
813 bracket (monitorNode nid) unmonitor $ \mRef ->
814 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
815 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
816 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
819-- | Get statistics about our local node
820getLocalNodeStats :: Process NodeStats
821getLocalNodeStats = do
823 sendCtrlMsg Nothing $ GetNodeStats self
824 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
826-- | Get information about the specified process
827getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
829 let them = processNodeId pid in do
831 dest <- mkNode them us
832 sendCtrlMsg dest $ GetInfo pid
834 match (\(p :: ProcessInfo) -> return $ Just p)
835 , match (\(_ :: ProcessInfoNone) -> return Nothing)
837 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
838 mkNode them us = case them == us of
839 True -> return Nothing
840 _ -> return $ Just them
842--------------------------------------------------------------------------------
843-- Monitoring and linking --
844--------------------------------------------------------------------------------
846-- | Link to a remote process (asynchronous)
848-- When process A links to process B (that is, process A calls
849-- @link pidB@) then an asynchronous exception will be thrown to process A
850-- when process B terminates (normally or abnormally), or when process A gets
851-- disconnected from process B. Although it is /technically/ possible to catch
852-- these exceptions, chances are if you find yourself trying to do so you should
853-- probably be using 'monitor' rather than 'link'. In particular, code such as
855-- > link pidB -- Link to process B
856-- > expect -- Wait for a message from process B
857-- > unlink pidB -- Unlink again
859-- doesn't quite do what one might expect: if process B sends a message to
860-- process A, and /subsequently terminates/, then process A might or might not
861-- be terminated too, depending on whether the exception is thrown before or
862-- after the 'unlink' (i.e., this code has a race condition).
864-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
865-- call to 'link' has no effect.
867-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
868-- Linking makes no distinction between normal and abnormal termination of
869-- the remote process.
870link :: ProcessId -> Process ()
871link = sendCtrlMsg Nothing . Link . ProcessIdentifier
873-- | Monitor another process (asynchronous)
875-- When process A monitors process B (that is, process A calls
876-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
877-- when process B terminates (normally or abnormally), or when process A gets
878-- disconnected from process B. You receive this message like any other (using
879-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
880-- 'DiedDisconnect', etc.).
882-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
883-- multiple monitors are set up, multiple notifications will be delivered
884-- and monitors can be disabled individually using 'unmonitor'.
885monitor :: ProcessId -> Process MonitorRef
886monitor = monitor' . ProcessIdentifier
888-- | Establishes temporary monitoring of another process.
890-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
891-- of @code@. Note: although monitoring is no longer active when
892-- @withMonitor@ returns, there might still be unreceived monitor
893-- messages in the queue.
895withMonitor :: ProcessId -> (MonitorRef -> Process a) -> Process a
896withMonitor pid = bracket (monitor pid) unmonitor
898-- | Establishes temporary monitoring of another process.
900-- @withMonitor_ pid code@ sets up monitoring of @pid@ for the duration
901-- of @code@. Note: although monitoring is no longer active when
902-- @withMonitor_@ returns, there might still be unreceived monitor
903-- messages in the queue.
906withMonitor_ :: ProcessId -> Process a -> Process a
907withMonitor_ p = withMonitor p . const
911-- This is synchronous in the sense that once it returns you are guaranteed
912-- that no exception will be raised if the remote process dies. However, it is
913-- asynchronous in the sense that we do not wait for a response from the remote
915unlink :: ProcessId -> Process ()
918 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
922-- | Remove a node link
924-- This has the same synchronous/asynchronous nature as 'unlink'.
925unlinkNode :: NodeId -> Process ()
928 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
932-- | Remove a channel (send port) link
934-- This has the same synchronous/asynchronous nature as 'unlink'.
935unlinkPort :: SendPort a -> Process ()
937 unlinkPortAsync sport
938 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
944-- This has the same synchronous/asynchronous nature as 'unlink'.
946-- ProcessMonitorNotification messages for the given MonitorRef are removed from
948unmonitor :: MonitorRef -> Process ()
951 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
954 -- Discard the notification if any. With the current NC implementation at most
955 -- one notification is in the mailbox for any given ref.
956 void $ receiveTimeout 0
957 [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref' == ref)
961--------------------------------------------------------------------------------
962-- Exception handling --
963--------------------------------------------------------------------------------
965-- | Lift 'Control.Exception.catch'
966catch :: Exception e => Process a -> (e -> Process a) -> Process a
968{-# DEPRECATED catch "Use Control.Monad.Catch.catch instead" #-}
970-- | Lift 'Control.Exception.try'
971try :: Exception e => Process a -> Process (Either e a)
973{-# DEPRECATED try "Use Control.Monad.Catch.try instead" #-}
975-- | Lift 'Control.Exception.mask'
976mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
978{-# DEPRECATED mask "Use Control.Monad.Catch.mask_ instead" #-}
980-- | Lift 'Control.Exception.mask_'
981mask_ :: Process a -> Process a
983{-# DEPRECATED mask_ "Use Control.Monad.Catch.mask_ instead" #-}
985-- | Lift 'Control.Exception.onException'
986onException :: Process a -> Process b -> Process a
987onException = Catch.onException
988{-# DEPRECATED onException "Use Control.Monad.Catch.onException instead" #-}
990-- | Lift 'Control.Exception.bracket'
991bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
992bracket = Catch.bracket
993{-# DEPRECATED bracket "Use Control.Monad.Catch.bracket instead" #-}
995-- | Lift 'Control.Exception.bracket_'
996bracket_ :: Process a -> Process b -> Process c -> Process c
997bracket_ = Catch.bracket_
998{-# DEPRECATED bracket_ "Use Control.Monad.Catch.bracket_ instead" #-}
1000-- | Lift 'Control.Exception.finally'
1001finally :: Process a -> Process b -> Process a
1002finally = Catch.finally
1003{-# DEPRECATED finally "Use Control.Monad.Catch.finally instead" #-}
1005-- | You need this when using 'catches'
1006data Handler a = forall e . Exception e => Handler (e -> Process a)
1008instance Functor Handler where
1009 fmap f (Handler h) = Handler (fmap f . h)
1011-- | Lift 'Control.Exception.catches'
1012catches :: Process a -> [Handler a] -> Process a
1013catches proc handlers = proc `Catch.catch` catchesHandler handlers
1015catchesHandler :: [Handler a] -> SomeException -> Process a
1016catchesHandler handlers e = foldr tryHandler (throwM e) handlers
1017 where tryHandler (Handler handler) res
1018 = case fromException e of
1019 Just e' -> handler e'
1022--------------------------------------------------------------------------------
1024--------------------------------------------------------------------------------
1026-- | Like 'expect' but with a timeout
1027expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
1028expectTimeout n = receiveTimeout n [match return]
1030-- | Asynchronous version of 'spawn'
1032-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
1033spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
1034spawnAsync nid proc = do
1035 spawnRef <- getSpawnRef
1038 then sendCtrlMsg Nothing $ Spawn proc spawnRef
1039 else sendCtrlMsg (Just nid) $ Spawn proc spawnRef
1042-- | Monitor a node (asynchronous)
1043monitorNode :: NodeId -> Process MonitorRef
1045 monitor' . NodeIdentifier
1047-- | Monitor a typed channel (asynchronous)
1048monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
1049monitorPort (SendPort cid) =
1050 monitor' (SendPortIdentifier cid)
1052-- | Remove a monitor (asynchronous)
1053unmonitorAsync :: MonitorRef -> Process ()
1055 sendCtrlMsg Nothing . Unmonitor
1057-- | Link to a node (asynchronous)
1058linkNode :: NodeId -> Process ()
1059linkNode = link' . NodeIdentifier
1061-- | Link to a channel (asynchronous)
1062linkPort :: SendPort a -> Process ()
1063linkPort (SendPort cid) =
1064 link' (SendPortIdentifier cid)
1066-- | Remove a process link (asynchronous)
1067unlinkAsync :: ProcessId -> Process ()
1069 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
1071-- | Remove a node link (asynchronous)
1072unlinkNodeAsync :: NodeId -> Process ()
1074 sendCtrlMsg Nothing . Unlink . NodeIdentifier
1076-- | Remove a channel (send port) link (asynchronous)
1077unlinkPortAsync :: SendPort a -> Process ()
1078unlinkPortAsync (SendPort cid) =
1079 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1081--------------------------------------------------------------------------------
1083--------------------------------------------------------------------------------
1085data SayMessage = SayMessage { sayTime :: UTCTime
1086 , sayProcess :: ProcessId
1087 , sayMessage :: String }
1090-- There is sadly no Show UTCTime instance
1091instance Show SayMessage where
1094 $ showString "SayMessage "
1095 . showString (formatTime defaultTimeLocale "%c" (sayTime msg))
1097 . showsPrec 11 (sayProcess msg) . showChar ' '
1098 . showsPrec 11 (sayMessage msg) . showChar ' '
1100instance Binary SayMessage where
1102 putUTCTime (sayTime s)
1105 get = SayMessage <$> getUTCTime <*> get <*> get
1107-- Sadly there is no Binary UTCTime instance
1108putUTCTime :: UTCTime -> Put
1109putUTCTime (UTCTime (ModifiedJulianDay day) tod) = do
1111 put (toRational tod)
1113getUTCTime :: Get UTCTime
1117 return $! UTCTime (ModifiedJulianDay day)
1122-- @say message@ sends a message of type 'SayMessage' with the current time and
1123-- 'ProcessId' of the current process to the process registered as @logger@. By
1124-- default, this process simply sends the string to @stderr@. Individual Cloud
1125-- Haskell backends might replace this with a different logger process, however.
1126say :: String -> Process ()
1128 now <- liftIO getCurrentTime
1130 nsend "logger" (SayMessage now us string)
1132--------------------------------------------------------------------------------
1134--------------------------------------------------------------------------------
1136-- | Register a process with the local registry (synchronous). The name must not
1137-- already be registered. The process need not be on this node. A bad
1138-- registration will result in a 'ProcessRegistrationException'
1140-- The process to be registered does not have to be local itself.
1141register :: String -> ProcessId -> Process ()
1142register = registerImpl False
1144-- | Like 'register', but will replace an existing registration.
1145-- The name must already be registered.
1146reregister :: String -> ProcessId -> Process ()
1147reregister = registerImpl True
1149registerImpl :: Bool -> String -> ProcessId -> Process ()
1150registerImpl force label pid = do
1151 mynid <- getSelfNode
1152 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1154 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1155 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1158-- | Register a process with a remote registry (asynchronous).
1160-- The process to be registered does not have to live on the same remote node.
1161-- Reply wil come in the form of a 'RegisterReply' message
1163-- See comments in 'whereisRemoteAsync'
1164registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1165registerRemoteAsync nid label pid = do
1167 sendCtrlMsg (if nid == here then Nothing else Just nid)
1168 (Register label nid (Just pid) False)
1170reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1171reregisterRemoteAsync nid label pid =
1172 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1174-- | Remove a process from the local registry (asynchronous).
1175-- This version will wait until a response is gotten from the
1176-- management process. The name must already be registered.
1177unregister :: String -> Process ()
1178unregister label = do
1179 mynid <- getSelfNode
1180 sendCtrlMsg Nothing (Register label mynid Nothing False)
1182 [ matchIf (\(RegisterReply label' _ _) -> label == label')
1183 (\(RegisterReply _ ok owner) -> handleRegistrationReply label ok owner)
1186-- | Deal with the result from an attempted registration or unregistration
1187-- by throwing an exception if necessary
1188handleRegistrationReply :: String -> Bool -> Maybe ProcessId -> Process ()
1189handleRegistrationReply label ok owner =
1191 throwM $ ProcessRegistrationException label owner
1193-- | Remove a process from a remote registry (asynchronous).
1195-- Reply wil come in the form of a 'RegisterReply' message
1197-- See comments in 'whereisRemoteAsync'
1198unregisterRemoteAsync :: NodeId -> String -> Process ()
1199unregisterRemoteAsync nid label =
1200 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1202-- | Query the local process registry
1203whereis :: String -> Process (Maybe ProcessId)
1205 sendCtrlMsg Nothing (WhereIs label)
1206 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1207 (\(WhereIsReply _ mPid) -> return mPid)
1210-- | Query a remote process registry (asynchronous)
1212-- Reply will come in the form of a 'WhereIsReply' message.
1214-- There is currently no synchronous version of 'whereisRemoteAsync': if
1215-- you implement one yourself, be sure to take into account that the remote
1216-- node might die or get disconnect before it can respond (i.e. you should
1217-- use 'monitorNode' and take appropriate action when you receive a
1218-- 'NodeMonitorNotification').
1219whereisRemoteAsync :: NodeId -> String -> Process ()
1220whereisRemoteAsync nid label = do
1222 sendCtrlMsg (if nid == here then Nothing else Just nid) (WhereIs label)
1224-- | Named send to a process in the local registry (asynchronous)
1225nsend :: Serializable a => String -> a -> Process ()
1227 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1229-- | Named send to a process in the local registry (asynchronous).
1230-- This function makes /no/ attempt to serialize and (in the case when the
1231-- destination process resides on the same local node) therefore ensure that
1232-- the payload is fully evaluated before it is delivered.
1233unsafeNSend :: Serializable a => String -> a -> Process ()
1234unsafeNSend = Unsafe.nsend
1236-- | Named send to a process in a remote registry (asynchronous)
1237nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1238nsendRemote nid label msg = do
1240 if here == nid then nsend label msg
1241 else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1243-- | Named send to a process in a remote registry (asynchronous)
1244-- This function makes /no/ attempt to serialize and (in the case when the
1245-- destination process resides on the same local node) therefore ensure that
1246-- the payload is fully evaluated before it is delivered.
1247unsafeNSendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1248unsafeNSendRemote = Unsafe.nsendRemote
1250--------------------------------------------------------------------------------
1252--------------------------------------------------------------------------------
1254-- | Resolve a static value
1255unStatic :: Typeable a => Static a -> Process a
1257 rtable <- remoteTable . processNode <$> ask
1258 case Static.unstatic rtable static of
1259 Left err -> fail $ "Could not resolve static value: " ++ err
1262-- | Resolve a closure
1263unClosure :: Typeable a => Closure a -> Process a
1264unClosure closure = do
1265 rtable <- remoteTable . processNode <$> ask
1266 case Static.unclosure rtable closure of
1267 Left err -> fail $ "Could not resolve closure: " ++ err
1270--------------------------------------------------------------------------------
1272--------------------------------------------------------------------------------
1274-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1275-- message passing. However, when network connections get disrupted this
1276-- illusion cannot always be maintained. Once a network connection breaks (even
1277-- temporarily) no further communication on that connection will be possible.
1278-- For example, if process A sends a message to process B, and A is then
1279-- notified (by monitor notification) that it got disconnected from B, A will
1280-- not be able to send any further messages to B, /unless/ A explicitly
1281-- indicates that it is acceptable to attempt to reconnect to B using the
1282-- Cloud Haskell 'reconnect' primitive.
1284-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1285-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1286-- then receives a monitor notification that its connection to B has been lost,
1287-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1288-- and m3 but not m2.
1290-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1291-- to attempt to reconnect on the next send/. In particular, if no further
1292-- communication attempts are made to B then A can use reconnect to clean up
1293-- its connection to B.
1294reconnect :: ProcessId -> Process ()
1297 node <- processNode <$> ask
1298 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1300-- | Reconnect to a sendport. See 'reconnect' for more information.
1301reconnectPort :: SendPort a -> Process ()
1302reconnectPort them = do
1304 node <- processNode <$> ask
1305 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1307--------------------------------------------------------------------------------
1308-- Auxiliary functions --
1309--------------------------------------------------------------------------------
1311sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1313 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1315sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1316sendChanLocal spId msg =
1317 -- we *must* fully serialize/encode the message here, because
1318 -- attempting to use `unsafeCoerce' in the node controller
1319 -- won't work since we know nothing about the required type
1320 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1322getMonitorRefFor :: Identifier -> Process MonitorRef
1323getMonitorRefFor ident = do
1325 liftIO $ modifyMVar (processState proc) $ \st -> do
1326 let counter = st ^. monitorCounter
1327 return ( monitorCounter ^: (+ 1) $ st
1328 , MonitorRef ident counter
1331getSpawnRef :: Process SpawnRef
1334 liftIO $ modifyMVar (processState proc) $ \st -> do
1335 let counter = st ^. spawnCounter
1336 return ( spawnCounter ^: (+ 1) $ st
1340-- | Monitor a process/node/channel
1341monitor' :: Identifier -> Process MonitorRef
1343 monitorRef <- getMonitorRefFor ident
1344 sendCtrlMsg Nothing $ Monitor monitorRef
1347-- | Link to a process/node/channel
1348link' :: Identifier -> Process ()
1349link' = 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