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
46 -- * Process management
48 , ProcessTerminationException(..)
54 -- keep the exception constructor hidden, so that handling exit
55 -- reasons /must/ take place via the 'catchExit' family of primitives
56 , ProcessExitException()
61 -- * Monitoring and linking
76 , reregisterRemoteAsync
77 , unregisterRemoteAsync
83 -- * Exception handling
106 -- * Internal Exports
110#if ! MIN_VERSION_base(4,6,0)
111import Prelude hiding (catch)
114import Data.Binary (decode)
115import Data.Time.Clock (getCurrentTime)
116import Data.Time.Format (formatTime)
117import System.Locale (defaultTimeLocale)
118import System.Timeout (timeout)
119import Control.Monad (when)
120import Control.Monad.Reader (ask)
121import Control.Monad.IO.Class (MonadIO, liftIO)
122import Control.Applicative ((<$>))
123import Control.Exception (Exception(..), throw, throwIO, SomeException)
124import qualified Control.Exception as Ex (catch, mask, try)
125import Control.Distributed.Process.Internal.StrictMVar
130import Control.Concurrent.STM
139import Control.Distributed.Process.Internal.CQueue
144import Control.Distributed.Process.Serializable (Serializable, fingerprint)
145import Data.Accessor ((^.), (^:), (^=))
146import Control.Distributed.Static
150import Data.Rank1Typeable (Typeable)
151import qualified Control.Distributed.Static as Static (unstatic, unclosure)
152import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
153import Control.Distributed.Process.Internal.Types
172 , ProcessExitException(..)
174 , DidUnlinkProcess(..)
179 , ProcessRegistrationException(..)
181 , ProcessInfoNone(..)
184 , createUnencodedMessage
186 , ImplicitReconnect( NoImplicitReconnect)
191import Control.Distributed.Process.Internal.Messaging
198import Control.Distributed.Process.Management.Types
201import Control.Distributed.Process.Management.Trace.Types
204import Control.Distributed.Process.Internal.WeakTQueue
212--------------------------------------------------------------------------------
214--------------------------------------------------------------------------------
217send :: Serializable a => ProcessId -> a -> Process ()
218-- This requires a lookup on every send. If we want to avoid that we need to
219-- modify serializable to allow for stateful (IO) deserialization.
222 let us = processId proc
223 node = processNode proc
224 nodeId = localNodeId node
225 destNode = (processNodeId them) in do
226 case destNode == nodeId of
227 True -> sendLocal them msg
228 False -> liftIO $ sendMessage (processNode proc)
229 (ProcessIdentifier (processId proc))
230 (ProcessIdentifier them)
233 -- We do not fire the trace event until after the sending is complete;
234 -- In the remote case, 'sendMessage' can block in the networking stack.
235 liftIO $ traceEvent (localEventBus node)
236 (MxSent them us (createUnencodedMessage msg))
238-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
239-- and (in the case when the destination process resides on the same local
240-- node) therefore ensure that the payload is fully evaluated before it is
242unsafeSend :: Serializable a => ProcessId -> a -> Process ()
243unsafeSend = Unsafe.send
245-- | Wait for a message of a specific type
246expect :: forall a. Serializable a => Process a
247expect = receiveWait [match return]
249--------------------------------------------------------------------------------
251--------------------------------------------------------------------------------
253-- | Create a new typed channel
254newChan :: Serializable a => Process (SendPort a, ReceivePort a)
257 liftIO . modifyMVar (processState proc) $ \st -> do
258 let lcid = st ^. channelCounter
259 let cid = SendPortId { sendPortProcessId = processId proc
260 , sendPortLocalId = lcid
262 let sport = SendPort cid
263 chan <- liftIO newTQueueIO
264 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
265 let rport = ReceivePort $ readTQueue chan
266 let tch = TypedChannel chan'
267 return ( (channelCounter ^: (+ 1))
268 . (typedChannelWithId lcid ^= Just tch)
273 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
274 finalizer st lcid = modifyMVar_ st $
275 return . (typedChannelWithId lcid ^= Nothing)
277-- | Send a message on a typed channel
278sendChan :: Serializable a => SendPort a -> a -> Process ()
279sendChan (SendPort cid) msg = do
281 let node = localNodeId (processNode proc)
282 destNode = processNodeId (sendPortProcessId cid) in do
283 case destNode == node of
284 True -> sendChanLocal cid msg
286 liftIO $ sendBinary (processNode proc)
287 (ProcessIdentifier (processId proc))
288 (SendPortIdentifier cid)
292-- | Send a message on a typed channel. This function makes /no/ attempt to
293-- serialize and (in the case when the @ReceivePort@ resides on the same local
294-- node) therefore ensure that the payload is fully evaluated before it is
296unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
297unsafeSendChan = Unsafe.sendChan
299-- | Wait for a message on a typed channel
300receiveChan :: Serializable a => ReceivePort a -> Process a
301receiveChan = liftIO . atomically . receiveSTM
303-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
304-- non-blocking check for a message.
305receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
306receiveChanTimeout 0 ch = liftIO . atomically $
307 (Just <$> receiveSTM ch) `orElse` return Nothing
308receiveChanTimeout n ch = liftIO . timeout n . atomically $
311-- | Merge a list of typed channels.
313-- The result port is left-biased: if there are messages available on more
314-- than one port, the first available message is returned.
315mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
316mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
318-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
320mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
321mergePortsRR = \ps -> do
322 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
323 return $ ReceivePort (rr psVar)
327 rotate (x:xs) = xs ++ [x]
329 rr :: TVar [STM a] -> STM a
332 a <- foldr1 orElse ps
333 writeTVar psVar (rotate ps)
336--------------------------------------------------------------------------------
337-- Advanced messaging --
338--------------------------------------------------------------------------------
340-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
341newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
343-- | Test the matches in order against each message in the queue
344receiveWait :: [Match b] -> Process b
346 queue <- processQueue <$> ask
347 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
350-- | Like 'receiveWait' but with a timeout.
352-- If the timeout is zero do a non-blocking check for matching messages. A
353-- non-zero timeout is applied only when waiting for incoming messages (that is,
354-- /after/ we have checked the messages that are already in the mailbox).
355receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
356receiveTimeout t ms = do
357 queue <- processQueue <$> ask
358 let blockSpec = if t == 0 then NonBlocking else Timeout t
359 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
361 Nothing -> return Nothing
362 Just proc -> Just <$> proc
364-- | Match on a typed channel
365matchChan :: ReceivePort a -> (a -> Process b) -> Match b
366matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
368-- | Match on an arbitrary STM action.
370-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
371-- actions with checks against our process' mailbox and/or any typed channel
372-- @ReceivePort@s we may hold.
374-- This allows us to process multiple input streams /along with our mailbox/,
375-- in just the same way that 'matchChan' supports checking both the mailbox
376-- /and/ an arbitrary set of typed channels in one atomic transaction.
378-- Note there are no ordering guarnatees with respect to these disparate input
381matchSTM :: STM a -> (a -> Process b) -> Match b
382matchSTM stm fn = Match $ MatchChan (fmap fn stm)
384-- | Match against any message of the right type
385match :: forall a b. Serializable a => (a -> Process b) -> Match b
386match = matchIf (const True)
388-- | Match against any message of the right type that satisfies a predicate
389matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
390matchIf c p = Match $ MatchMsg $ \msg ->
391 case messageFingerprint msg == fingerprint (undefined :: a) of
394 (UnencodedMessage _ m) ->
395 let m' = unsafeCoerce m :: a in
399 (EncodedMessage _ _) ->
400 if (c decoded) then Just (p decoded) else Nothing
403 -- Make sure the value is fully decoded so that we don't hang to
404 -- bytestrings when the process calling 'matchIf' doesn't process
405 -- the values immediately
406 !decoded = decode (messageEncoding msg)
408-- | Match against any message, regardless of the underlying (contained) type
409matchMessage :: (Message -> Process Message) -> Match Message
410matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
412-- | Match against any message (regardless of underlying type) that satisfies a predicate
413matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
414matchMessageIf c p = Match $ MatchMsg $ \msg ->
419-- | Forward a raw 'Message' to the given 'ProcessId'.
420forward :: Message -> ProcessId -> Process ()
423 let node = processNode proc
425 nid = localNodeId node
426 destNode = (processNodeId them) in do
427 case destNode == nid of
428 True -> sendCtrlMsg Nothing (LocalSend them msg)
429 False -> liftIO $ sendPayload (processNode proc)
430 (ProcessIdentifier (processId proc))
431 (ProcessIdentifier them)
433 (messageToPayload msg)
434 -- We do not fire the trace event until after the sending is complete;
435 -- In the remote case, 'sendMessage' can block in the networking stack.
436 liftIO $ traceEvent (localEventBus node)
440-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
441-- 'Serializable' - like the datum they contain - but also note, deserialising
442-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
443-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
448-- send self (wrapMessage "blah")
449-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
450-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
451-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
454wrapMessage :: Serializable a => a -> Message
455wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
457-- | This is the /unsafe/ variant of 'wrapMessage'. See
458-- "Control.Distributed.Process.UnsafePrimitives" for details.
459unsafeWrapMessage :: Serializable a => a -> Message
460unsafeWrapMessage = Unsafe.wrapMessage
462-- [note Serializable UnencodedMessage]
463-- if we attempt to serialise an UnencodedMessage, it will be converted to an
464-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
465-- Primitives.hs for the gory details
467-- | Attempt to unwrap a raw 'Message'.
468-- If the type of the decoded message payload matches the expected type, the
469-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
472-- This expression, for example, will evaluate to @Nothing@
473-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
475-- Whereas this expression, will yield @Just "foo"@
476-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
478unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
480 case messageFingerprint msg == fingerprint (undefined :: a) of
481 False -> return Nothing :: m (Maybe a)
483 (UnencodedMessage _ ms) ->
484 let ms' = unsafeCoerce ms :: a
486 (EncodedMessage _ _) ->
487 return (Just (decoded))
489 decoded :: a -- note [decoding]
490 !decoded = decode (messageEncoding msg)
492-- | Attempt to handle a raw 'Message'.
493-- If the type of the message matches the type of the first argument to
494-- the supplied expression, then the message will be decoded and the expression
495-- evaluated against its value. If this runtime type checking fails however,
496-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
497-- evaluation proceeds, the resulting value with be wrapped with @Just@.
499-- Intended for use in `catchesExit` and `matchAny` primitives.
501handleMessage :: forall m a b. (Monad m, Serializable a)
502 => Message -> (a -> m b) -> m (Maybe b)
503handleMessage msg proc = handleMessageIf msg (const True) proc
505-- | Conditionally handle a raw 'Message'.
506-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
507-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
508-- for further information about runtime type checking.
510handleMessageIf :: forall m a b . (Monad m, Serializable a)
515handleMessageIf msg c proc = do
516 case messageFingerprint msg == fingerprint (undefined :: a) of
517 False -> return Nothing :: m (Maybe b)
519 (UnencodedMessage _ ms) ->
520 let ms' = unsafeCoerce ms :: a in
522 True -> do { r <- proc ms'; return (Just r) }
523 False -> return Nothing :: m (Maybe b)
524 (EncodedMessage _ _) ->
526 True -> do { r <- proc (decoded :: a); return (Just r) }
527 False -> return Nothing :: m (Maybe b)
529 decoded :: a -- note [decoding]
530 !decoded = decode (messageEncoding msg)
532-- | As 'handleMessage' but ignores result, which is useful if you don't
533-- care whether or not the handler succeeded.
535handleMessage_ :: forall m a . (Monad m, Serializable a)
536 => Message -> (a -> m ()) -> m ()
537handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
539-- | Conditional version of 'handleMessage_'.
541handleMessageIf_ :: forall m a . (Monad m, Serializable a)
546handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
548-- | Match against an arbitrary message. 'matchAny' removes the first available
549-- message from the process mailbox. To handle arbitrary /raw/ messages once
550-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
552matchAny :: forall b. (Message -> Process b) -> Match b
553matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
555-- | Match against an arbitrary message. Intended for use with 'handleMessage'
556-- and 'unwrapMessage', this function /only/ removes a message from the process
557-- mailbox, /if/ the supplied condition matches. The success (or failure) of
558-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
559-- here, i.e., if the condition evaluates to @True@ then the message will
560-- be removed from the process mailbox and decoded, but that does /not/
561-- guarantee that an expression passed to @handleMessage@ will pass the
562-- runtime type checks and therefore be evaluated.
564matchAnyIf :: forall a b. (Serializable a)
566 -> (Message -> Process b)
568matchAnyIf c p = Match $ MatchMsg $ \msg ->
569 case messageFingerprint msg == fingerprint (undefined :: a) of
570 True | check -> Just (p msg)
575 (EncodedMessage _ _) -> c decoded
576 (UnencodedMessage _ m') -> c (unsafeCoerce m')
578 decoded :: a -- note [decoding]
579 !decoded = decode (messageEncoding msg)
583For an EncodedMessage, we need to ensure the value is fully decoded so that
584we don't hang to bytestrings if the calling process doesn't evaluate
585immediately. For UnencodedMessage we know (because the fingerprint comparison
586succeeds) that unsafeCoerce will not fail.
589-- | Remove any message from the queue
590matchUnknown :: Process b -> Match b
591matchUnknown p = Match $ MatchMsg (const (Just p))
593-- | Receives messages and forwards them to @pid@ if @p msg == True@.
594delegate :: ProcessId -> (Message -> Bool) -> Process ()
597 matchAny (\m -> case (p m) of
598 True -> forward m pid
603-- | A straight relay that forwards all messages to the supplied pid.
604relay :: ProcessId -> Process ()
605relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
607-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
608-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
609-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
610-- fails, no action is taken and the proxy process will continue running.
611proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
615 next <- handleMessage m proc
617 Just True -> forward m pid
618 Just False -> return () -- explicitly ignored
619 Nothing -> return ()) -- un-routable
623--------------------------------------------------------------------------------
624-- Process management --
625--------------------------------------------------------------------------------
627-- | Thrown by 'terminate'
628data ProcessTerminationException = ProcessTerminationException
629 deriving (Show, Typeable)
631instance Exception ProcessTerminationException
633-- | Terminate immediately (throws a ProcessTerminationException)
634terminate :: Process a
635terminate = liftIO $ throwIO ProcessTerminationException
638-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
639die :: Serializable a => a -> Process b
641-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
642-- could be decoded by a handler passed to 'catchExit', re-thrown or even
643-- passed to another process via the tracing mechanism.
645 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
647-- | Forceful request to kill a process. Where 'exit' provides an exception
648-- that can be caught and handled, 'kill' throws an unexposed exception type
649-- which cannot be handled explicitly (by type).
650kill :: ProcessId -> String -> Process ()
651-- NOTE: We send the message to our local node controller, which will then
652-- forward it to a remote node controller (if applicable). Sending it directly
653-- to a remote node controller means that that the message may overtake a
654-- 'monitor' or 'link' request.
655kill them reason = sendCtrlMsg Nothing (Kill them reason)
657-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
658-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
659-- manner can be handled using the 'catchExit' family of functions.
660exit :: Serializable a => ProcessId -> a -> Process ()
661-- NOTE: We send the message to our local node controller, which will then
662-- forward it to a remote node controller (if applicable). Sending it directly
663-- to a remote node controller means that that the message may overtake a
664-- 'monitor' or 'link' request.
665exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
667-- | Catches 'ProcessExitException'. The handler will not be applied unless its
668-- type matches the encoded data stored in the exception (see the /reason/
669-- argument given to the 'exit' primitive). If the handler cannot be applied,
670-- the exception will be re-thrown.
672-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
673-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
674catchExit :: forall a b . (Show a, Serializable a)
676 -> (ProcessId -> a -> Process b)
678catchExit act exitHandler = catch act handleExit
680 handleExit ex@(ProcessExitException from msg) =
681 if messageFingerprint msg == fingerprint (undefined :: a)
682 then exitHandler from decoded
683 else liftIO $ throwIO ex
686 -- Make sure the value is fully decoded so that we don't hang to
687 -- bytestrings if the caller doesn't use the value immediately
688 !decoded = decode (messageEncoding msg)
690-- | Lift 'Control.Exception.catches' (almost).
692-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
693-- message, a handler must accept inputs of the expected type. In order to
694-- handle a list of potentially different handlers (and therefore input types),
695-- a handler passed to 'catchesExit' must accept 'Message' and return
696-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
698-- See 'maybeHandleMessage' and 'Message' for more details.
699catchesExit :: Process b
700 -> [(ProcessId -> Message -> (Process (Maybe b)))]
702catchesExit act handlers = catch act ((flip handleExit) handlers)
704 handleExit :: ProcessExitException
705 -> [(ProcessId -> Message -> Process (Maybe b))]
707 handleExit ex [] = liftIO $ throwIO ex
708 handleExit ex@(ProcessExitException from msg) (h:hs) = do
711 Nothing -> handleExit ex hs
714-- | Our own process ID
715getSelfPid :: Process ProcessId
716getSelfPid = processId <$> ask
718-- | Get the node ID of our local node
719getSelfNode :: Process NodeId
720getSelfNode = localNodeId . processNode <$> ask
722-- | Get information about the specified process
723getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
725 let them = processNodeId pid in do
727 dest <- mkNode them us
728 sendCtrlMsg dest $ GetInfo pid
730 match (\(p :: ProcessInfo) -> return $ Just p)
731 , match (\(_ :: ProcessInfoNone) -> return Nothing)
733 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
734 mkNode them us = case them == us of
735 True -> return Nothing
736 _ -> return $ Just them
738--------------------------------------------------------------------------------
739-- Monitoring and linking --
740--------------------------------------------------------------------------------
742-- | Link to a remote process (asynchronous)
744-- When process A links to process B (that is, process A calls
745-- @link pidB@) then an asynchronous exception will be thrown to process A
746-- when process B terminates (normally or abnormally), or when process A gets
747-- disconnected from process B. Although it is /technically/ possible to catch
748-- these exceptions, chances are if you find yourself trying to do so you should
749-- probably be using 'monitor' rather than 'link'. In particular, code such as
751-- > link pidB -- Link to process B
752-- > expect -- Wait for a message from process B
753-- > unlink pidB -- Unlink again
755-- doesn't quite do what one might expect: if process B sends a message to
756-- process A, and /subsequently terminates/, then process A might or might not
757-- be terminated too, depending on whether the exception is thrown before or
758-- after the 'unlink' (i.e., this code has a race condition).
760-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
761-- call to 'link' has no effect.
763-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
764-- Linking makes no distinction between normal and abnormal termination of
765-- the remote process.
766link :: ProcessId -> Process ()
767link = sendCtrlMsg Nothing . Link . ProcessIdentifier
769-- | Monitor another process (asynchronous)
771-- When process A monitors process B (that is, process A calls
772-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
773-- when process B terminates (normally or abnormally), or when process A gets
774-- disconnected from process B. You receive this message like any other (using
775-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
776-- 'DiedDisconnect', etc.).
778-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
779-- multiple monitors are set up, multiple notifications will be delivered
780-- and monitors can be disabled individually using 'unmonitor'.
781monitor :: ProcessId -> Process MonitorRef
782monitor = monitor' . ProcessIdentifier
784-- | Establishes temporary monitoring of another process.
786-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
787-- of @code@. Note: although monitoring is no longer active when
788-- @withMonitor@ returns, there might still be unreceived monitor
789-- messages in the queue.
791withMonitor :: ProcessId -> Process a -> Process a
792withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
793 -- unmonitor blocks waiting for the response, so there's a possibility
794 -- that an exception might interrupt withMonitor before the unmonitor
795 -- has completed. I think that's better than making the unmonitor
800-- This is synchronous in the sense that once it returns you are guaranteed
801-- that no exception will be raised if the remote process dies. However, it is
802-- asynchronous in the sense that we do not wait for a response from the remote
804unlink :: ProcessId -> Process ()
807 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
811-- | Remove a node link
813-- This has the same synchronous/asynchronous nature as 'unlink'.
814unlinkNode :: NodeId -> Process ()
817 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
821-- | Remove a channel (send port) link
823-- This has the same synchronous/asynchronous nature as 'unlink'.
824unlinkPort :: SendPort a -> Process ()
826 unlinkPortAsync sport
827 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
833-- This has the same synchronous/asynchronous nature as 'unlink'.
834unmonitor :: MonitorRef -> Process ()
837 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
841--------------------------------------------------------------------------------
842-- Exception handling --
843--------------------------------------------------------------------------------
845-- | Lift 'Control.Exception.catch'
846catch :: Exception e => Process a -> (e -> Process a) -> Process a
849 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
851-- | Lift 'Control.Exception.try'
852try :: Exception e => Process a -> Process (Either e a)
855 liftIO $ Ex.try (runLocalProcess lproc p)
857-- | Lift 'Control.Exception.mask'
858mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
861 liftIO $ Ex.mask $ \restore ->
862 runLocalProcess lproc (p (liftRestore restore))
864 liftRestore :: (forall a. IO a -> IO a)
865 -> (forall a. Process a -> Process a)
866 liftRestore restoreIO = \p2 -> do
868 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
870-- | Lift 'Control.Exception.onException'
871onException :: Process a -> Process b -> Process a
872onException p what = p `catch` \e -> do _ <- what
873 liftIO $ throwIO (e :: SomeException)
875-- | Lift 'Control.Exception.bracket'
876bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
877bracket before after thing =
878 mask $ \restore -> do
880 r <- restore (thing a) `onException` after a
884-- | Lift 'Control.Exception.bracket_'
885bracket_ :: Process a -> Process b -> Process c -> Process c
886bracket_ before after thing = bracket before (const after) (const thing)
888-- | Lift 'Control.Exception.finally'
889finally :: Process a -> Process b -> Process a
890finally a sequel = bracket_ (return ()) sequel a
892-- | You need this when using 'catches'
893data Handler a = forall e . Exception e => Handler (e -> Process a)
895instance Functor Handler where
896 fmap f (Handler h) = Handler (fmap f . h)
898-- | Lift 'Control.Exception.catches'
899catches :: Process a -> [Handler a] -> Process a
900catches proc handlers = proc `catch` catchesHandler handlers
902catchesHandler :: [Handler a] -> SomeException -> Process a
903catchesHandler handlers e = foldr tryHandler (throw e) handlers
904 where tryHandler (Handler handler) res
905 = case fromException e of
906 Just e' -> handler e'
909--------------------------------------------------------------------------------
911--------------------------------------------------------------------------------
913-- | Like 'expect' but with a timeout
914expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
915expectTimeout n = receiveTimeout n [match return]
917-- | Asynchronous version of 'spawn'
919-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
920spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
921spawnAsync nid proc = do
922 spawnRef <- getSpawnRef
923 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
926-- | Monitor a node (asynchronous)
927monitorNode :: NodeId -> Process MonitorRef
929 monitor' . NodeIdentifier
931-- | Monitor a typed channel (asynchronous)
932monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
933monitorPort (SendPort cid) =
934 monitor' (SendPortIdentifier cid)
936-- | Remove a monitor (asynchronous)
937unmonitorAsync :: MonitorRef -> Process ()
939 sendCtrlMsg Nothing . Unmonitor
941-- | Link to a node (asynchronous)
942linkNode :: NodeId -> Process ()
943linkNode = link' . NodeIdentifier
945-- | Link to a channel (asynchronous)
946linkPort :: SendPort a -> Process ()
947linkPort (SendPort cid) =
948 link' (SendPortIdentifier cid)
950-- | Remove a process link (asynchronous)
951unlinkAsync :: ProcessId -> Process ()
953 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
955-- | Remove a node link (asynchronous)
956unlinkNodeAsync :: NodeId -> Process ()
958 sendCtrlMsg Nothing . Unlink . NodeIdentifier
960-- | Remove a channel (send port) link (asynchronous)
961unlinkPortAsync :: SendPort a -> Process ()
962unlinkPortAsync (SendPort cid) =
963 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
965--------------------------------------------------------------------------------
967--------------------------------------------------------------------------------
971-- @say message@ sends a message (time, pid of the current process, message)
972-- to the process registered as 'logger'. By default, this process simply
973-- sends the string to 'stderr'. Individual Cloud Haskell backends might
974-- replace this with a different logger process, however.
975say :: String -> Process ()
977 now <- liftIO getCurrentTime
979 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
981--------------------------------------------------------------------------------
983--------------------------------------------------------------------------------
985-- | Register a process with the local registry (asynchronous).
986-- This version will wait until a response is gotten from the
987-- management process. The name must not already be registered.
988-- The process need not be on this node.
989-- A bad registration will result in a 'ProcessRegistrationException'
991-- The process to be registered does not have to be local itself.
992register :: String -> ProcessId -> Process ()
993register = registerImpl False
995-- | Like 'register', but will replace an existing registration.
996-- The name must already be registered.
997reregister :: String -> ProcessId -> Process ()
998reregister = registerImpl True
1000registerImpl :: Bool -> String -> ProcessId -> Process ()
1001registerImpl force label pid = do
1002 mynid <- getSelfNode
1003 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1004 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
1005 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
1008-- | Register a process with a remote registry (asynchronous).
1010-- The process to be registered does not have to live on the same remote node.
1011-- Reply wil come in the form of a 'RegisterReply' message
1013-- See comments in 'whereisRemoteAsync'
1014registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1015registerRemoteAsync nid label pid =
1016 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
1018reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1019reregisterRemoteAsync nid label pid =
1020 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1022-- | Remove a process from the local registry (asynchronous).
1023-- This version will wait until a response is gotten from the
1024-- management process. The name must already be registered.
1025unregister :: String -> Process ()
1026unregister label = do
1027 mynid <- getSelfNode
1028 sendCtrlMsg Nothing (Register label mynid Nothing False)
1029 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
1030 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
1033-- | Deal with the result from an attempted registration or unregistration
1034-- by throwing an exception if necessary
1035handleRegistrationReply :: String -> Bool -> Process ()
1036handleRegistrationReply label ok =
1038 liftIO $ throwIO $ ProcessRegistrationException label
1040-- | Remove a process from a remote registry (asynchronous).
1042-- Reply wil come in the form of a 'RegisterReply' message
1044-- See comments in 'whereisRemoteAsync'
1045unregisterRemoteAsync :: NodeId -> String -> Process ()
1046unregisterRemoteAsync nid label =
1047 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1049-- | Query the local process registry
1050whereis :: String -> Process (Maybe ProcessId)
1052 sendCtrlMsg Nothing (WhereIs label)
1053 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1054 (\(WhereIsReply _ mPid) -> return mPid)
1057-- | Query a remote process registry (asynchronous)
1059-- Reply will come in the form of a 'WhereIsReply' message.
1061-- There is currently no synchronous version of 'whereisRemoteAsync': if
1062-- you implement one yourself, be sure to take into account that the remote
1063-- node might die or get disconnect before it can respond (i.e. you should
1064-- use 'monitorNode' and take appropriate action when you receive a
1065-- 'NodeMonitorNotification').
1066whereisRemoteAsync :: NodeId -> String -> Process ()
1067whereisRemoteAsync nid label =
1068 sendCtrlMsg (Just nid) (WhereIs label)
1070-- | Named send to a process in the local registry (asynchronous)
1071nsend :: Serializable a => String -> a -> Process ()
1073 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1075-- | Named send to a process in the local registry (asynchronous).
1076-- This function makes /no/ attempt to serialize and (in the case when the
1077-- destination process resides on the same local node) therefore ensure that
1078-- the payload is fully evaluated before it is delivered.
1079unsafeNSend :: Serializable a => String -> a -> Process ()
1080unsafeNSend = Unsafe.nsend
1082-- | Named send to a process in a remote registry (asynchronous)
1083nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1084nsendRemote nid label msg =
1085 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1087--------------------------------------------------------------------------------
1089--------------------------------------------------------------------------------
1091-- | Resolve a static value
1092unStatic :: Typeable a => Static a -> Process a
1094 rtable <- remoteTable . processNode <$> ask
1095 case Static.unstatic rtable static of
1096 Left err -> fail $ "Could not resolve static value: " ++ err
1099-- | Resolve a closure
1100unClosure :: Typeable a => Closure a -> Process a
1101unClosure closure = do
1102 rtable <- remoteTable . processNode <$> ask
1103 case Static.unclosure rtable closure of
1104 Left err -> fail $ "Could not resolve closure: " ++ err
1107--------------------------------------------------------------------------------
1109--------------------------------------------------------------------------------
1111-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1112-- message passing. However, when network connections get disrupted this
1113-- illusion cannot always be maintained. Once a network connection breaks (even
1114-- temporarily) no further communication on that connection will be possible.
1115-- For example, if process A sends a message to process B, and A is then
1116-- notified (by monitor notification) that it got disconnected from B, A will
1117-- not be able to send any further messages to B, /unless/ A explicitly
1118-- indicates that it is acceptable to attempt to reconnect to B using the
1119-- Cloud Haskell 'reconnect' primitive.
1121-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1122-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1123-- then receives a monitor notification that its connection to B has been lost,
1124-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1125-- and m3 but not m2.
1127-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1128-- to attempt to reconnect on the next send/. In particular, if no further
1129-- communication attempts are made to B then A can use reconnect to clean up
1130-- its connection to B.
1131reconnect :: ProcessId -> Process ()
1134 node <- processNode <$> ask
1135 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1137-- | Reconnect to a sendport. See 'reconnect' for more information.
1138reconnectPort :: SendPort a -> Process ()
1139reconnectPort them = do
1141 node <- processNode <$> ask
1142 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1144--------------------------------------------------------------------------------
1145-- Auxiliary functions --
1146--------------------------------------------------------------------------------
1148sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1150 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1152sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1153sendChanLocal spId msg =
1154 -- we *must* fully serialize/encode the message here, because
1155 -- attempting to use `unsafeCoerce' in the node controller
1156 -- won't work since we know nothing about the required type
1157 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1159getMonitorRefFor :: Identifier -> Process MonitorRef
1160getMonitorRefFor ident = do
1162 liftIO $ modifyMVar (processState proc) $ \st -> do
1163 let counter = st ^. monitorCounter
1164 return ( monitorCounter ^: (+ 1) $ st
1165 , MonitorRef ident counter
1168getSpawnRef :: Process SpawnRef
1171 liftIO $ modifyMVar (processState proc) $ \st -> do
1172 let counter = st ^. spawnCounter
1173 return ( spawnCounter ^: (+ 1) $ st
1177-- | Monitor a process/node/channel
1178monitor' :: Identifier -> Process MonitorRef
1180 monitorRef <- getMonitorRefFor ident
1181 sendCtrlMsg Nothing $ Monitor monitorRef
1184-- | Link to a process/node/channel
1185link' :: Identifier -> Process ()
1186link' = 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