1{-# LANGUAGE BangPatterns #-}
3-- | Cloud Haskell primitives
5-- We define these in a separate module so that we don't have to rely on
6-- the closure combinators
7module Control.Distributed.Process.Internal.Primitives
17 -- * Unsafe messaging variants
21 -- * Advanced messaging
45 -- * Process management
47 , ProcessTerminationException(..)
53 -- keep the exception constructor hidden, so that handling exit
54 -- reasons /must/ take place via the 'catchExit' family of primitives
55 , ProcessExitException()
60 -- * Monitoring and linking
75 , reregisterRemoteAsync
76 , unregisterRemoteAsync
82 -- * Exception handling
105 -- * Internal Exports
109#if ! MIN_VERSION_base(4,6,0)
110import Prelude hiding (catch)
113import Data.Binary (decode)
114import Data.Time.Clock (getCurrentTime)
115import Data.Time.Format (formatTime)
116import System.Locale (defaultTimeLocale)
117import System.Timeout (timeout)
118import Control.Monad (when)
119import Control.Monad.Reader (ask)
120import Control.Monad.IO.Class (MonadIO, liftIO)
121import Control.Applicative ((<$>))
122import Control.Exception (Exception(..), throw, throwIO, SomeException)
123import qualified Control.Exception as Ex (catch, mask, try)
124import Control.Distributed.Process.Internal.StrictMVar
129import Control.Concurrent.STM
138import Control.Distributed.Process.Internal.CQueue
143import Control.Distributed.Process.Serializable (Serializable, fingerprint)
144import Data.Accessor ((^.), (^:), (^=))
145import Control.Distributed.Static
149import Data.Rank1Typeable (Typeable)
150import qualified Control.Distributed.Static as Static (unstatic, unclosure)
151import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
152import Control.Distributed.Process.Internal.Types
171 , ProcessExitException(..)
173 , DidUnlinkProcess(..)
178 , ProcessRegistrationException(..)
180 , ProcessInfoNone(..)
183 , createUnencodedMessage
185 , ImplicitReconnect( NoImplicitReconnect)
190import Control.Distributed.Process.Internal.Messaging
197import Control.Distributed.Process.Management.Types
200import Control.Distributed.Process.Management.Trace.Types
203import Control.Distributed.Process.Internal.WeakTQueue
211--------------------------------------------------------------------------------
213--------------------------------------------------------------------------------
216send :: Serializable a => ProcessId -> a -> Process ()
217-- This requires a lookup on every send. If we want to avoid that we need to
218-- modify serializable to allow for stateful (IO) deserialization.
221 let us = processId proc
222 node = processNode proc
223 nodeId = localNodeId node
224 destNode = (processNodeId them) in do
225 case destNode == nodeId of
226 True -> sendLocal them msg
227 False -> liftIO $ sendMessage (processNode proc)
228 (ProcessIdentifier (processId proc))
229 (ProcessIdentifier them)
232 -- We do not fire the trace event until after the sending is complete;
233 -- In the remote case, 'sendMessage' can block in the networking stack.
234 liftIO $ traceEvent (localEventBus node)
235 (MxSent them us (createUnencodedMessage msg))
237-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
238-- and (in the case when the destination process resides on the same local
239-- node) therefore ensure that the payload is fully evaluated before it is
241unsafeSend :: Serializable a => ProcessId -> a -> Process ()
242unsafeSend = Unsafe.send
244-- | Wait for a message of a specific type
245expect :: forall a. Serializable a => Process a
246expect = receiveWait [match return]
248--------------------------------------------------------------------------------
250--------------------------------------------------------------------------------
252-- | Create a new typed channel
253newChan :: Serializable a => Process (SendPort a, ReceivePort a)
256 liftIO . modifyMVar (processState proc) $ \st -> do
257 let lcid = st ^. channelCounter
258 let cid = SendPortId { sendPortProcessId = processId proc
259 , sendPortLocalId = lcid
261 let sport = SendPort cid
262 chan <- liftIO newTQueueIO
263 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
264 let rport = ReceivePort $ readTQueue chan
265 let tch = TypedChannel chan'
266 return ( (channelCounter ^: (+ 1))
267 . (typedChannelWithId lcid ^= Just tch)
272 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
273 finalizer st lcid = modifyMVar_ st $
274 return . (typedChannelWithId lcid ^= Nothing)
276-- | Send a message on a typed channel
277sendChan :: Serializable a => SendPort a -> a -> Process ()
278sendChan (SendPort cid) msg = do
280 let node = localNodeId (processNode proc)
281 destNode = processNodeId (sendPortProcessId cid) in do
282 case destNode == node of
283 True -> sendChanLocal cid msg
285 liftIO $ sendBinary (processNode proc)
286 (ProcessIdentifier (processId proc))
287 (SendPortIdentifier cid)
291-- | Send a message on a typed channel. This function makes /no/ attempt to
292-- serialize and (in the case when the @ReceivePort@ resides on the same local
293-- node) therefore ensure that the payload is fully evaluated before it is
295unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
296unsafeSendChan = Unsafe.sendChan
298-- | Wait for a message on a typed channel
299receiveChan :: Serializable a => ReceivePort a -> Process a
300receiveChan = liftIO . atomically . receiveSTM
302-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
303-- non-blocking check for a message.
304receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
305receiveChanTimeout 0 ch = liftIO . atomically $
306 (Just <$> receiveSTM ch) `orElse` return Nothing
307receiveChanTimeout n ch = liftIO . timeout n . atomically $
310-- | Merge a list of typed channels.
312-- The result port is left-biased: if there are messages available on more
313-- than one port, the first available message is returned.
314mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
315mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
317-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
319mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
320mergePortsRR = \ps -> do
321 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
322 return $ ReceivePort (rr psVar)
326 rotate (x:xs) = xs ++ [x]
328 rr :: TVar [STM a] -> STM a
331 a <- foldr1 orElse ps
332 writeTVar psVar (rotate ps)
335--------------------------------------------------------------------------------
336-- Advanced messaging --
337--------------------------------------------------------------------------------
339-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
340newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
342-- | Test the matches in order against each message in the queue
343receiveWait :: [Match b] -> Process b
345 queue <- processQueue <$> ask
346 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
349-- | Like 'receiveWait' but with a timeout.
351-- If the timeout is zero do a non-blocking check for matching messages. A
352-- non-zero timeout is applied only when waiting for incoming messages (that is,
353-- /after/ we have checked the messages that are already in the mailbox).
354receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
355receiveTimeout t ms = do
356 queue <- processQueue <$> ask
357 let blockSpec = if t == 0 then NonBlocking else Timeout t
358 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
360 Nothing -> return Nothing
361 Just proc -> Just <$> proc
363matchChan :: ReceivePort a -> (a -> Process b) -> Match b
364matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
366-- | Match against any message of the right type
367match :: forall a b. Serializable a => (a -> Process b) -> Match b
368match = matchIf (const True)
370-- | Match against any message of the right type that satisfies a predicate
371matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
372matchIf c p = Match $ MatchMsg $ \msg ->
373 case messageFingerprint msg == fingerprint (undefined :: a) of
376 (UnencodedMessage _ m) ->
377 let m' = unsafeCoerce m :: a in
381 (EncodedMessage _ _) ->
382 if (c decoded) then Just (p decoded) else Nothing
385 -- Make sure the value is fully decoded so that we don't hang to
386 -- bytestrings when the process calling 'matchIf' doesn't process
387 -- the values immediately
388 !decoded = decode (messageEncoding msg)
390-- | Match against any message, regardless of the underlying (contained) type
391matchMessage :: (Message -> Process Message) -> Match Message
392matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
394-- | Match against any message (regardless of underlying type) that satisfies a predicate
395matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
396matchMessageIf c p = Match $ MatchMsg $ \msg ->
401-- | Forward a raw 'Message' to the given 'ProcessId'.
402forward :: Message -> ProcessId -> Process ()
405 let node = processNode proc
407 nid = localNodeId node
408 destNode = (processNodeId them) in do
409 case destNode == nid of
410 True -> sendCtrlMsg Nothing (LocalSend them msg)
411 False -> liftIO $ sendPayload (processNode proc)
412 (ProcessIdentifier (processId proc))
413 (ProcessIdentifier them)
415 (messageToPayload msg)
416 -- We do not fire the trace event until after the sending is complete;
417 -- In the remote case, 'sendMessage' can block in the networking stack.
418 liftIO $ traceEvent (localEventBus node)
422-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
423-- 'Serializable' - like the datum they contain - but also note, deserialising
424-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
425-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
430-- send self (wrapMessage "blah")
431-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
432-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
433-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
436wrapMessage :: Serializable a => a -> Message
437wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
439-- | This is the /unsafe/ variant of 'wrapMessage'. See
440-- "Control.Distributed.Process.UnsafePrimitives" for details.
441unsafeWrapMessage :: Serializable a => a -> Message
442unsafeWrapMessage = Unsafe.wrapMessage
444-- [note Serializable UnencodedMessage]
445-- if we attempt to serialise an UnencodedMessage, it will be converted to an
446-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
447-- Primitives.hs for the gory details
449-- | Attempt to unwrap a raw 'Message'.
450-- If the type of the decoded message payload matches the expected type, the
451-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
454-- This expression, for example, will evaluate to @Nothing@
455-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
457-- Whereas this expression, will yield @Just "foo"@
458-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
460unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
462 case messageFingerprint msg == fingerprint (undefined :: a) of
463 False -> return Nothing :: m (Maybe a)
465 (UnencodedMessage _ ms) ->
466 let ms' = unsafeCoerce ms :: a
468 (EncodedMessage _ _) ->
469 return (Just (decoded))
471 decoded :: a -- note [decoding]
472 !decoded = decode (messageEncoding msg)
474-- | Attempt to handle a raw 'Message'.
475-- If the type of the message matches the type of the first argument to
476-- the supplied expression, then the message will be decoded and the expression
477-- evaluated against its value. If this runtime type checking fails however,
478-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
479-- evaluation proceeds, the resulting value with be wrapped with @Just@.
481-- Intended for use in `catchesExit` and `matchAny` primitives.
483handleMessage :: forall m a b. (Monad m, Serializable a)
484 => Message -> (a -> m b) -> m (Maybe b)
485handleMessage msg proc = handleMessageIf msg (const True) proc
487-- | Conditionally handle a raw 'Message'.
488-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
489-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
490-- for further information about runtime type checking.
492handleMessageIf :: forall m a b . (Monad m, Serializable a)
497handleMessageIf msg c proc = do
498 case messageFingerprint msg == fingerprint (undefined :: a) of
499 False -> return Nothing :: m (Maybe b)
501 (UnencodedMessage _ ms) ->
502 let ms' = unsafeCoerce ms :: a in
504 True -> do { r <- proc ms'; return (Just r) }
505 False -> return Nothing :: m (Maybe b)
506 (EncodedMessage _ _) ->
508 True -> do { r <- proc (decoded :: a); return (Just r) }
509 False -> return Nothing :: m (Maybe b)
511 decoded :: a -- note [decoding]
512 !decoded = decode (messageEncoding msg)
514-- | As 'handleMessage' but ignores result, which is useful if you don't
515-- care whether or not the handler succeeded.
517handleMessage_ :: forall m a . (Monad m, Serializable a)
518 => Message -> (a -> m ()) -> m ()
519handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
521-- | Conditional version of 'handleMessage_'.
523handleMessageIf_ :: forall m a . (Monad m, Serializable a)
528handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
530-- | Match against an arbitrary message. 'matchAny' removes the first available
531-- message from the process mailbox. To handle arbitrary /raw/ messages once
532-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
534matchAny :: forall b. (Message -> Process b) -> Match b
535matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
537-- | Match against an arbitrary message. Intended for use with 'handleMessage'
538-- and 'unwrapMessage', this function /only/ removes a message from the process
539-- mailbox, /if/ the supplied condition matches. The success (or failure) of
540-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
541-- here, i.e., if the condition evaluates to @True@ then the message will
542-- be removed from the process mailbox and decoded, but that does /not/
543-- guarantee that an expression passed to @handleMessage@ will pass the
544-- runtime type checks and therefore be evaluated.
546matchAnyIf :: forall a b. (Serializable a)
548 -> (Message -> Process b)
550matchAnyIf c p = Match $ MatchMsg $ \msg ->
551 case messageFingerprint msg == fingerprint (undefined :: a) of
552 True | check -> Just (p msg)
557 (EncodedMessage _ _) -> c decoded
558 (UnencodedMessage _ m') -> c (unsafeCoerce m')
560 decoded :: a -- note [decoding]
561 !decoded = decode (messageEncoding msg)
565For an EncodedMessage, we need to ensure the value is fully decoded so that
566we don't hang to bytestrings if the calling process doesn't evaluate
567immediately. For UnencodedMessage we know (because the fingerprint comparison
568succeeds) that unsafeCoerce will not fail.
571-- | Remove any message from the queue
572matchUnknown :: Process b -> Match b
573matchUnknown p = Match $ MatchMsg (const (Just p))
575-- | Receives messages and forwards them to @pid@ if @p msg == True@.
576delegate :: ProcessId -> (Message -> Bool) -> Process ()
579 matchAny (\m -> case (p m) of
580 True -> forward m pid
585-- | A straight relay that forwards all messages to the supplied pid.
586relay :: ProcessId -> Process ()
587relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
589-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
590-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
591-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
592-- fails, no action is taken and the proxy process will continue running.
593proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
597 next <- handleMessage m proc
599 Just True -> forward m pid
600 Just False -> return () -- explicitly ignored
601 Nothing -> return ()) -- un-routable
605--------------------------------------------------------------------------------
606-- Process management --
607--------------------------------------------------------------------------------
609-- | Thrown by 'terminate'
610data ProcessTerminationException = ProcessTerminationException
611 deriving (Show, Typeable)
613instance Exception ProcessTerminationException
615-- | Terminate immediately (throws a ProcessTerminationException)
616terminate :: Process a
617terminate = liftIO $ throwIO ProcessTerminationException
620-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
621die :: Serializable a => a -> Process b
623-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
624-- could be decoded by a handler passed to 'catchExit', re-thrown or even
625-- passed to another process via the tracing mechanism.
627 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
629-- | Forceful request to kill a process. Where 'exit' provides an exception
630-- that can be caught and handled, 'kill' throws an unexposed exception type
631-- which cannot be handled explicitly (by type).
632kill :: ProcessId -> String -> Process ()
633-- NOTE: We send the message to our local node controller, which will then
634-- forward it to a remote node controller (if applicable). Sending it directly
635-- to a remote node controller means that that the message may overtake a
636-- 'monitor' or 'link' request.
637kill them reason = sendCtrlMsg Nothing (Kill them reason)
639-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
640-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
641-- manner can be handled using the 'catchExit' family of functions.
642exit :: Serializable a => ProcessId -> a -> Process ()
643-- NOTE: We send the message to our local node controller, which will then
644-- forward it to a remote node controller (if applicable). Sending it directly
645-- to a remote node controller means that that the message may overtake a
646-- 'monitor' or 'link' request.
647exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
649-- | Catches 'ProcessExitException'. The handler will not be applied unless its
650-- type matches the encoded data stored in the exception (see the /reason/
651-- argument given to the 'exit' primitive). If the handler cannot be applied,
652-- the exception will be re-thrown.
654-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
655-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
656catchExit :: forall a b . (Show a, Serializable a)
658 -> (ProcessId -> a -> Process b)
660catchExit act exitHandler = catch act handleExit
662 handleExit ex@(ProcessExitException from msg) =
663 if messageFingerprint msg == fingerprint (undefined :: a)
664 then exitHandler from decoded
665 else liftIO $ throwIO ex
668 -- Make sure the value is fully decoded so that we don't hang to
669 -- bytestrings if the caller doesn't use the value immediately
670 !decoded = decode (messageEncoding msg)
672-- | Lift 'Control.Exception.catches' (almost).
674-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
675-- message, a handler must accept inputs of the expected type. In order to
676-- handle a list of potentially different handlers (and therefore input types),
677-- a handler passed to 'catchesExit' must accept 'Message' and return
678-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
680-- See 'maybeHandleMessage' and 'Message' for more details.
681catchesExit :: Process b
682 -> [(ProcessId -> Message -> (Process (Maybe b)))]
684catchesExit act handlers = catch act ((flip handleExit) handlers)
686 handleExit :: ProcessExitException
687 -> [(ProcessId -> Message -> Process (Maybe b))]
689 handleExit ex [] = liftIO $ throwIO ex
690 handleExit ex@(ProcessExitException from msg) (h:hs) = do
693 Nothing -> handleExit ex hs
696-- | Our own process ID
697getSelfPid :: Process ProcessId
698getSelfPid = processId <$> ask
700-- | Get the node ID of our local node
701getSelfNode :: Process NodeId
702getSelfNode = localNodeId . processNode <$> ask
704-- | Get information about the specified process
705getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
707 let them = processNodeId pid in do
709 dest <- mkNode them us
710 sendCtrlMsg dest $ GetInfo pid
712 match (\(p :: ProcessInfo) -> return $ Just p)
713 , match (\(_ :: ProcessInfoNone) -> return Nothing)
715 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
716 mkNode them us = case them == us of
717 True -> return Nothing
718 _ -> return $ Just them
720--------------------------------------------------------------------------------
721-- Monitoring and linking --
722--------------------------------------------------------------------------------
724-- | Link to a remote process (asynchronous)
726-- When process A links to process B (that is, process A calls
727-- @link pidB@) then an asynchronous exception will be thrown to process A
728-- when process B terminates (normally or abnormally), or when process A gets
729-- disconnected from process B. Although it is /technically/ possible to catch
730-- these exceptions, chances are if you find yourself trying to do so you should
731-- probably be using 'monitor' rather than 'link'. In particular, code such as
733-- > link pidB -- Link to process B
734-- > expect -- Wait for a message from process B
735-- > unlink pidB -- Unlink again
737-- doesn't quite do what one might expect: if process B sends a message to
738-- process A, and /subsequently terminates/, then process A might or might not
739-- be terminated too, depending on whether the exception is thrown before or
740-- after the 'unlink' (i.e., this code has a race condition).
742-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
743-- call to 'link' has no effect.
745-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
746-- Linking makes no distinction between normal and abnormal termination of
747-- the remote process.
748link :: ProcessId -> Process ()
749link = sendCtrlMsg Nothing . Link . ProcessIdentifier
751-- | Monitor another process (asynchronous)
753-- When process A monitors process B (that is, process A calls
754-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
755-- when process B terminates (normally or abnormally), or when process A gets
756-- disconnected from process B. You receive this message like any other (using
757-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
758-- 'DiedDisconnect', etc.).
760-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
761-- multiple monitors are set up, multiple notifications will be delivered
762-- and monitors can be disabled individually using 'unmonitor'.
763monitor :: ProcessId -> Process MonitorRef
764monitor = monitor' . ProcessIdentifier
766-- | Establishes temporary monitoring of another process.
768-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
769-- of @code@. Note: although monitoring is no longer active when
770-- @withMonitor@ returns, there might still be unreceived monitor
771-- messages in the queue.
773withMonitor :: ProcessId -> Process a -> Process a
774withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
775 -- unmonitor blocks waiting for the response, so there's a possibility
776 -- that an exception might interrupt withMonitor before the unmonitor
777 -- has completed. I think that's better than making the unmonitor
782-- This is synchronous in the sense that once it returns you are guaranteed
783-- that no exception will be raised if the remote process dies. However, it is
784-- asynchronous in the sense that we do not wait for a response from the remote
786unlink :: ProcessId -> Process ()
789 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
793-- | Remove a node link
795-- This has the same synchronous/asynchronous nature as 'unlink'.
796unlinkNode :: NodeId -> Process ()
799 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
803-- | Remove a channel (send port) link
805-- This has the same synchronous/asynchronous nature as 'unlink'.
806unlinkPort :: SendPort a -> Process ()
808 unlinkPortAsync sport
809 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
815-- This has the same synchronous/asynchronous nature as 'unlink'.
816unmonitor :: MonitorRef -> Process ()
819 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
823--------------------------------------------------------------------------------
824-- Exception handling --
825--------------------------------------------------------------------------------
827-- | Lift 'Control.Exception.catch'
828catch :: Exception e => Process a -> (e -> Process a) -> Process a
831 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
833-- | Lift 'Control.Exception.try'
834try :: Exception e => Process a -> Process (Either e a)
837 liftIO $ Ex.try (runLocalProcess lproc p)
839-- | Lift 'Control.Exception.mask'
840mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
843 liftIO $ Ex.mask $ \restore ->
844 runLocalProcess lproc (p (liftRestore restore))
846 liftRestore :: (forall a. IO a -> IO a)
847 -> (forall a. Process a -> Process a)
848 liftRestore restoreIO = \p2 -> do
850 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
852-- | Lift 'Control.Exception.onException'
853onException :: Process a -> Process b -> Process a
854onException p what = p `catch` \e -> do _ <- what
855 liftIO $ throwIO (e :: SomeException)
857-- | Lift 'Control.Exception.bracket'
858bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
859bracket before after thing =
860 mask $ \restore -> do
862 r <- restore (thing a) `onException` after a
866-- | Lift 'Control.Exception.bracket_'
867bracket_ :: Process a -> Process b -> Process c -> Process c
868bracket_ before after thing = bracket before (const after) (const thing)
870-- | Lift 'Control.Exception.finally'
871finally :: Process a -> Process b -> Process a
872finally a sequel = bracket_ (return ()) sequel a
874-- | You need this when using 'catches'
875data Handler a = forall e . Exception e => Handler (e -> Process a)
877instance Functor Handler where
878 fmap f (Handler h) = Handler (fmap f . h)
880-- | Lift 'Control.Exception.catches'
881catches :: Process a -> [Handler a] -> Process a
882catches proc handlers = proc `catch` catchesHandler handlers
884catchesHandler :: [Handler a] -> SomeException -> Process a
885catchesHandler handlers e = foldr tryHandler (throw e) handlers
886 where tryHandler (Handler handler) res
887 = case fromException e of
888 Just e' -> handler e'
891--------------------------------------------------------------------------------
893--------------------------------------------------------------------------------
895-- | Like 'expect' but with a timeout
896expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
897expectTimeout n = receiveTimeout n [match return]
899-- | Asynchronous version of 'spawn'
901-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
902spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
903spawnAsync nid proc = do
904 spawnRef <- getSpawnRef
905 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
908-- | Monitor a node (asynchronous)
909monitorNode :: NodeId -> Process MonitorRef
911 monitor' . NodeIdentifier
913-- | Monitor a typed channel (asynchronous)
914monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
915monitorPort (SendPort cid) =
916 monitor' (SendPortIdentifier cid)
918-- | Remove a monitor (asynchronous)
919unmonitorAsync :: MonitorRef -> Process ()
921 sendCtrlMsg Nothing . Unmonitor
923-- | Link to a node (asynchronous)
924linkNode :: NodeId -> Process ()
925linkNode = link' . NodeIdentifier
927-- | Link to a channel (asynchronous)
928linkPort :: SendPort a -> Process ()
929linkPort (SendPort cid) =
930 link' (SendPortIdentifier cid)
932-- | Remove a process link (asynchronous)
933unlinkAsync :: ProcessId -> Process ()
935 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
937-- | Remove a node link (asynchronous)
938unlinkNodeAsync :: NodeId -> Process ()
940 sendCtrlMsg Nothing . Unlink . NodeIdentifier
942-- | Remove a channel (send port) link (asynchronous)
943unlinkPortAsync :: SendPort a -> Process ()
944unlinkPortAsync (SendPort cid) =
945 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
947--------------------------------------------------------------------------------
949--------------------------------------------------------------------------------
953-- @say message@ sends a message (time, pid of the current process, message)
954-- to the process registered as 'logger'. By default, this process simply
955-- sends the string to 'stderr'. Individual Cloud Haskell backends might
956-- replace this with a different logger process, however.
957say :: String -> Process ()
959 now <- liftIO getCurrentTime
961 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
963--------------------------------------------------------------------------------
965--------------------------------------------------------------------------------
967-- | Register a process with the local registry (asynchronous).
968-- This version will wait until a response is gotten from the
969-- management process. The name must not already be registered.
970-- The process need not be on this node.
971-- A bad registration will result in a 'ProcessRegistrationException'
973-- The process to be registered does not have to be local itself.
974register :: String -> ProcessId -> Process ()
975register = registerImpl False
977-- | Like 'register', but will replace an existing registration.
978-- The name must already be registered.
979reregister :: String -> ProcessId -> Process ()
980reregister = registerImpl True
982registerImpl :: Bool -> String -> ProcessId -> Process ()
983registerImpl force label pid = do
985 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
986 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
987 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
990-- | Register a process with a remote registry (asynchronous).
992-- The process to be registered does not have to live on the same remote node.
993-- Reply wil come in the form of a 'RegisterReply' message
995-- See comments in 'whereisRemoteAsync'
996registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
997registerRemoteAsync nid label pid =
998 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
1000reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1001reregisterRemoteAsync nid label pid =
1002 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1004-- | Remove a process from the local registry (asynchronous).
1005-- This version will wait until a response is gotten from the
1006-- management process. The name must already be registered.
1007unregister :: String -> Process ()
1008unregister label = do
1009 mynid <- getSelfNode
1010 sendCtrlMsg Nothing (Register label mynid Nothing False)
1011 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
1012 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
1015-- | Deal with the result from an attempted registration or unregistration
1016-- by throwing an exception if necessary
1017handleRegistrationReply :: String -> Bool -> Process ()
1018handleRegistrationReply label ok =
1020 liftIO $ throwIO $ ProcessRegistrationException label
1022-- | Remove a process from a remote registry (asynchronous).
1024-- Reply wil come in the form of a 'RegisterReply' message
1026-- See comments in 'whereisRemoteAsync'
1027unregisterRemoteAsync :: NodeId -> String -> Process ()
1028unregisterRemoteAsync nid label =
1029 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1031-- | Query the local process registry
1032whereis :: String -> Process (Maybe ProcessId)
1034 sendCtrlMsg Nothing (WhereIs label)
1035 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1036 (\(WhereIsReply _ mPid) -> return mPid)
1039-- | Query a remote process registry (asynchronous)
1041-- Reply will come in the form of a 'WhereIsReply' message.
1043-- There is currently no synchronous version of 'whereisRemoteAsync': if
1044-- you implement one yourself, be sure to take into account that the remote
1045-- node might die or get disconnect before it can respond (i.e. you should
1046-- use 'monitorNode' and take appropriate action when you receive a
1047-- 'NodeMonitorNotification').
1048whereisRemoteAsync :: NodeId -> String -> Process ()
1049whereisRemoteAsync nid label =
1050 sendCtrlMsg (Just nid) (WhereIs label)
1052-- | Named send to a process in the local registry (asynchronous)
1053nsend :: Serializable a => String -> a -> Process ()
1055 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1057-- | Named send to a process in the local registry (asynchronous).
1058-- This function makes /no/ attempt to serialize and (in the case when the
1059-- destination process resides on the same local node) therefore ensure that
1060-- the payload is fully evaluated before it is delivered.
1061unsafeNSend :: Serializable a => String -> a -> Process ()
1062unsafeNSend = Unsafe.nsend
1064-- | Named send to a process in a remote registry (asynchronous)
1065nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1066nsendRemote nid label msg =
1067 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1069--------------------------------------------------------------------------------
1071--------------------------------------------------------------------------------
1073-- | Resolve a static value
1074unStatic :: Typeable a => Static a -> Process a
1076 rtable <- remoteTable . processNode <$> ask
1077 case Static.unstatic rtable static of
1078 Left err -> fail $ "Could not resolve static value: " ++ err
1081-- | Resolve a closure
1082unClosure :: Typeable a => Closure a -> Process a
1083unClosure closure = do
1084 rtable <- remoteTable . processNode <$> ask
1085 case Static.unclosure rtable closure of
1086 Left err -> fail $ "Could not resolve closure: " ++ err
1089--------------------------------------------------------------------------------
1091--------------------------------------------------------------------------------
1093-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1094-- message passing. However, when network connections get disrupted this
1095-- illusion cannot always be maintained. Once a network connection breaks (even
1096-- temporarily) no further communication on that connection will be possible.
1097-- For example, if process A sends a message to process B, and A is then
1098-- notified (by monitor notification) that it got disconnected from B, A will
1099-- not be able to send any further messages to B, /unless/ A explicitly
1100-- indicates that it is acceptable to attempt to reconnect to B using the
1101-- Cloud Haskell 'reconnect' primitive.
1103-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1104-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1105-- then receives a monitor notification that its connection to B has been lost,
1106-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1107-- and m3 but not m2.
1109-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1110-- to attempt to reconnect on the next send/. In particular, if no further
1111-- communication attempts are made to B then A can use reconnect to clean up
1112-- its connection to B.
1113reconnect :: ProcessId -> Process ()
1116 node <- processNode <$> ask
1117 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1119-- | Reconnect to a sendport. See 'reconnect' for more information.
1120reconnectPort :: SendPort a -> Process ()
1121reconnectPort them = do
1123 node <- processNode <$> ask
1124 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1126--------------------------------------------------------------------------------
1127-- Auxiliary functions --
1128--------------------------------------------------------------------------------
1130sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1132 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1134sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1135sendChanLocal spId msg =
1136 -- we *must* fully serialize/encode the message here, because
1137 -- attempting to use `unsafeCoerce' in the node controller
1138 -- won't work since we know nothing about the required type
1139 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1141getMonitorRefFor :: Identifier -> Process MonitorRef
1142getMonitorRefFor ident = do
1144 liftIO $ modifyMVar (processState proc) $ \st -> do
1145 let counter = st ^. monitorCounter
1146 return ( monitorCounter ^: (+ 1) $ st
1147 , MonitorRef ident counter
1150getSpawnRef :: Process SpawnRef
1153 liftIO $ modifyMVar (processState proc) $ \st -> do
1154 let counter = st ^. spawnCounter
1155 return ( spawnCounter ^: (+ 1) $ st
1159-- | Monitor a process/node/channel
1160monitor' :: Identifier -> Process MonitorRef
1162 monitorRef <- getMonitorRefFor ident
1163 sendCtrlMsg Nothing $ Monitor monitorRef
1166-- | Link to a process/node/channel
1167link' :: Identifier -> Process ()
1168link' = 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