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
42 -- * Process management
44 , ProcessTerminationException(..)
50 -- keep the exception constructor hidden, so that handling exit
51 -- reasons /must/ take place via the 'catchExit' family of primitives
52 , ProcessExitException()
57 -- * Monitoring and linking
72 , reregisterRemoteAsync
73 , unregisterRemoteAsync
79 -- * Exception handling
102 -- * Internal Exports
106#if ! MIN_VERSION_base(4,6,0)
107import Prelude hiding (catch)
110import Data.Binary (decode)
111import Data.Time.Clock (getCurrentTime)
112import Data.Time.Format (formatTime)
113import System.Locale (defaultTimeLocale)
114import System.Timeout (timeout)
115import Control.Monad (when)
116import Control.Monad.Reader (ask)
117import Control.Monad.IO.Class (MonadIO, liftIO)
118import Control.Applicative ((<$>))
119import Control.Exception (Exception(..), throw, throwIO, SomeException)
120import qualified Control.Exception as Ex (catch, mask, try)
121import Control.Distributed.Process.Internal.StrictMVar
126import Control.Concurrent.STM
135import Control.Distributed.Process.Internal.CQueue
140import Control.Distributed.Process.Serializable (Serializable, fingerprint)
141import Data.Accessor ((^.), (^:), (^=))
142import Control.Distributed.Static
146import Data.Rank1Typeable (Typeable)
147import qualified Control.Distributed.Static as Static (unstatic, unclosure)
148import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
149import Control.Distributed.Process.Internal.Types
168 , ProcessExitException(..)
170 , DidUnlinkProcess(..)
175 , ProcessRegistrationException(..)
177 , ProcessInfoNone(..)
180 , createUnencodedMessage
182 , ImplicitReconnect( NoImplicitReconnect)
187import Control.Distributed.Process.Internal.Messaging
194import Control.Distributed.Process.Management.Types
197import Control.Distributed.Process.Management.Trace.Types
200import Control.Distributed.Process.Internal.WeakTQueue
208--------------------------------------------------------------------------------
210--------------------------------------------------------------------------------
213send :: Serializable a => ProcessId -> a -> Process ()
214-- This requires a lookup on every send. If we want to avoid that we need to
215-- modify serializable to allow for stateful (IO) deserialization.
218 let us = processId proc
219 node = processNode proc
220 nodeId = localNodeId node
221 destNode = (processNodeId them) in do
222 case destNode == nodeId of
223 True -> sendLocal them msg
224 False -> liftIO $ sendMessage (processNode proc)
225 (ProcessIdentifier (processId proc))
226 (ProcessIdentifier them)
229 -- We do not fire the trace event until after the sending is complete;
230 -- In the remote case, 'sendMessage' can block in the networking stack.
231 liftIO $ traceEvent (localEventBus node)
232 (MxSent them us (createUnencodedMessage msg))
234-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
235-- and (in the case when the destination process resides on the same local
236-- node) therefore ensure that the payload is fully evaluated before it is
238unsafeSend :: Serializable a => ProcessId -> a -> Process ()
239unsafeSend = Unsafe.send
241-- | Wait for a message of a specific type
242expect :: forall a. Serializable a => Process a
243expect = receiveWait [match return]
245--------------------------------------------------------------------------------
247--------------------------------------------------------------------------------
249-- | Create a new typed channel
250newChan :: Serializable a => Process (SendPort a, ReceivePort a)
253 liftIO . modifyMVar (processState proc) $ \st -> do
254 let lcid = st ^. channelCounter
255 let cid = SendPortId { sendPortProcessId = processId proc
256 , sendPortLocalId = lcid
258 let sport = SendPort cid
259 chan <- liftIO newTQueueIO
260 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
261 let rport = ReceivePort $ readTQueue chan
262 let tch = TypedChannel chan'
263 return ( (channelCounter ^: (+ 1))
264 . (typedChannelWithId lcid ^= Just tch)
269 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
270 finalizer st lcid = modifyMVar_ st $
271 return . (typedChannelWithId lcid ^= Nothing)
273-- | Send a message on a typed channel
274sendChan :: Serializable a => SendPort a -> a -> Process ()
275sendChan (SendPort cid) msg = do
277 let node = localNodeId (processNode proc)
278 destNode = processNodeId (sendPortProcessId cid) in do
279 case destNode == node of
280 True -> sendChanLocal cid msg
282 liftIO $ sendBinary (processNode proc)
283 (ProcessIdentifier (processId proc))
284 (SendPortIdentifier cid)
288-- | Send a message on a typed channel. This function makes /no/ attempt to
289-- serialize and (in the case when the @ReceivePort@ resides on the same local
290-- node) therefore ensure that the payload is fully evaluated before it is
292unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
293unsafeSendChan = Unsafe.sendChan
295-- | Wait for a message on a typed channel
296receiveChan :: Serializable a => ReceivePort a -> Process a
297receiveChan = liftIO . atomically . receiveSTM
299-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
300-- non-blocking check for a message.
301receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
302receiveChanTimeout 0 ch = liftIO . atomically $
303 (Just <$> receiveSTM ch) `orElse` return Nothing
304receiveChanTimeout n ch = liftIO . timeout n . atomically $
307-- | Merge a list of typed channels.
309-- The result port is left-biased: if there are messages available on more
310-- than one port, the first available message is returned.
311mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
312mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
314-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
316mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
317mergePortsRR = \ps -> do
318 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
319 return $ ReceivePort (rr psVar)
323 rotate (x:xs) = xs ++ [x]
325 rr :: TVar [STM a] -> STM a
328 a <- foldr1 orElse ps
329 writeTVar psVar (rotate ps)
332--------------------------------------------------------------------------------
333-- Advanced messaging --
334--------------------------------------------------------------------------------
336-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
337newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
339-- | Test the matches in order against each message in the queue
340receiveWait :: [Match b] -> Process b
342 queue <- processQueue <$> ask
343 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
346-- | Like 'receiveWait' but with a timeout.
348-- If the timeout is zero do a non-blocking check for matching messages. A
349-- non-zero timeout is applied only when waiting for incoming messages (that is,
350-- /after/ we have checked the messages that are already in the mailbox).
351receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
352receiveTimeout t ms = do
353 queue <- processQueue <$> ask
354 let blockSpec = if t == 0 then NonBlocking else Timeout t
355 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
357 Nothing -> return Nothing
358 Just proc -> Just <$> proc
360matchChan :: ReceivePort a -> (a -> Process b) -> Match b
361matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
363-- | Match against any message of the right type
364match :: forall a b. Serializable a => (a -> Process b) -> Match b
365match = matchIf (const True)
367-- | Match against any message of the right type that satisfies a predicate
368matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
369matchIf c p = Match $ MatchMsg $ \msg ->
370 case messageFingerprint msg == fingerprint (undefined :: a) of
373 (UnencodedMessage _ m) ->
374 let m' = unsafeCoerce m :: a in
378 (EncodedMessage _ _) ->
379 if (c decoded) then Just (p decoded) else Nothing
382 -- Make sure the value is fully decoded so that we don't hang to
383 -- bytestrings when the process calling 'matchIf' doesn't process
384 -- the values immediately
385 !decoded = decode (messageEncoding msg)
387-- | Match against any message, regardless of the underlying (contained) type
388matchMessage :: (Message -> Process Message) -> Match Message
389matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
391-- | Match against any message (regardless of underlying type) that satisfies a predicate
392matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
393matchMessageIf c p = Match $ MatchMsg $ \msg ->
398-- | Forward a raw 'Message' to the given 'ProcessId'.
399forward :: Message -> ProcessId -> Process ()
402 let node = processNode proc
404 nid = localNodeId node
405 destNode = (processNodeId them) in do
406 case destNode == nid of
407 True -> sendCtrlMsg Nothing (LocalSend them msg)
408 False -> liftIO $ sendPayload (processNode proc)
409 (ProcessIdentifier (processId proc))
410 (ProcessIdentifier them)
412 (messageToPayload msg)
413 -- We do not fire the trace event until after the sending is complete;
414 -- In the remote case, 'sendMessage' can block in the networking stack.
415 liftIO $ traceEvent (localEventBus node)
419-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
420-- 'Serializable' - like the datum they contain - but also note, deserialising
421-- a 'Message' will yield a 'Message', not the type within it! To obtain the
422-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
427-- send self (wrapMessage "blah")
428-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
429-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
430-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
433wrapMessage :: Serializable a => a -> Message
434wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
436-- [note Serializable UnencodedMessage]
437-- if we attempt to serialise an UnencodedMessage, it will be converted to an
438-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
439-- Primitives.hs for the gory details
441-- | Attempt to unwrap a raw 'Message'.
442-- If the type of the decoded message payload matches the expected type, the
443-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
446-- This expression, for example, will evaluate to @Nothing@
447-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
449-- Whereas this expression, will yield @Just "foo"@
450-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
452unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
454 case messageFingerprint msg == fingerprint (undefined :: a) of
455 False -> return Nothing :: m (Maybe a)
457 (UnencodedMessage _ ms) ->
458 let ms' = unsafeCoerce ms :: a
460 (EncodedMessage _ _) ->
461 return (Just (decoded))
463 decoded :: a -- note [decoding]
464 !decoded = decode (messageEncoding msg)
466-- | Attempt to handle a raw 'Message'.
467-- If the type of the message matches the type of the first argument to
468-- the supplied expression, then the message will be decoded and the expression
469-- evaluated against its value. If this runtime type checking fails however,
470-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
471-- evaluation proceeds, the resulting value with be wrapped with @Just@.
473-- Intended for use in `catchesExit` and `matchAny` primitives.
475handleMessage :: forall m a b. (Monad m, Serializable a)
476 => Message -> (a -> m b) -> m (Maybe b)
477handleMessage msg proc = handleMessageIf msg (const True) proc
479handleMessageIf :: forall m a b . (Monad m, Serializable a)
484handleMessageIf msg c proc = do
485 case messageFingerprint msg == fingerprint (undefined :: a) of
486 False -> return Nothing :: m (Maybe b)
488 (UnencodedMessage _ ms) ->
489 let ms' = unsafeCoerce ms :: a in
491 True -> do { r <- proc ms'; return (Just r) }
492 False -> return Nothing :: m (Maybe b)
493 (EncodedMessage _ _) ->
495 True -> do { r <- proc (decoded :: a); return (Just r) }
496 False -> return Nothing :: m (Maybe b)
498 decoded :: a -- note [decoding]
499 !decoded = decode (messageEncoding msg)
501-- | Match against an arbitrary message. 'matchAny' removes the first available
502-- message from the process mailbox. To handle arbitrary /raw/ messages once
503-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
505matchAny :: forall b. (Message -> Process b) -> Match b
506matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
508-- | Match against an arbitrary message. Intended for use with 'handleMessage'
509-- and 'unwrapMessage', this function /only/ removes a message from the process
510-- mailbox, /if/ the supplied condition matches. The success (or failure) of
511-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
512-- here, i.e., if the condition evaluates to @True@ then the message will
513-- be removed from the process mailbox and decoded, but that does /not/
514-- guarantee that an expression passed to @handleMessage@ will pass the
515-- runtime type checks and therefore be evaluated.
517matchAnyIf :: forall a b. (Serializable a)
519 -> (Message -> Process b)
521matchAnyIf c p = Match $ MatchMsg $ \msg ->
522 case messageFingerprint msg == fingerprint (undefined :: a) of
523 True | check -> Just (p msg)
528 (EncodedMessage _ _) -> c decoded
529 (UnencodedMessage _ m') -> c (unsafeCoerce m')
531 decoded :: a -- note [decoding]
532 !decoded = decode (messageEncoding msg)
536For an EncodedMessage, we need to ensure the value is fully decoded so that
537we don't hang to bytestrings if the calling process doesn't evaluate
538immediately. For UnencodedMessage we know (because the fingerprint comparison
539succeeds) that unsafeCoerce will not fail.
542-- | Remove any message from the queue
543matchUnknown :: Process b -> Match b
544matchUnknown p = Match $ MatchMsg (const (Just p))
546-- | Receives messages and forwards them to @pid@ if @p msg == True@.
547delegate :: ProcessId -> (Message -> Bool) -> Process ()
550 matchAny (\m -> case (p m) of
551 True -> forward m pid
556-- | A straight relay that forwards all messages to the supplied pid.
557relay :: ProcessId -> Process ()
558relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
560-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
561-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
562-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
563-- fails, no action is taken and the proxy process will continue running.
564proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
568 next <- handleMessage m proc
570 Just True -> forward m pid
571 Just False -> return () -- explicitly ignored
572 Nothing -> return ()) -- un-routable
576--------------------------------------------------------------------------------
577-- Process management --
578--------------------------------------------------------------------------------
580-- | Thrown by 'terminate'
581data ProcessTerminationException = ProcessTerminationException
582 deriving (Show, Typeable)
584instance Exception ProcessTerminationException
586-- | Terminate immediately (throws a ProcessTerminationException)
587terminate :: Process a
588terminate = liftIO $ throwIO ProcessTerminationException
591-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
592die :: Serializable a => a -> Process b
594-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
595-- could be decoded by a handler passed to 'catchExit', re-thrown or even
596-- passed to another process via the tracing mechanism.
598 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
600-- | Forceful request to kill a process. Where 'exit' provides an exception
601-- that can be caught and handled, 'kill' throws an unexposed exception type
602-- which cannot be handled explicitly (by type).
603kill :: ProcessId -> String -> Process ()
604-- NOTE: We send the message to our local node controller, which will then
605-- forward it to a remote node controller (if applicable). Sending it directly
606-- to a remote node controller means that that the message may overtake a
607-- 'monitor' or 'link' request.
608kill them reason = sendCtrlMsg Nothing (Kill them reason)
610-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
611-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
612-- manner can be handled using the 'catchExit' family of functions.
613exit :: Serializable a => ProcessId -> a -> Process ()
614-- NOTE: We send the message to our local node controller, which will then
615-- forward it to a remote node controller (if applicable). Sending it directly
616-- to a remote node controller means that that the message may overtake a
617-- 'monitor' or 'link' request.
618exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
620-- | Catches 'ProcessExitException'. The handler will not be applied unless its
621-- type matches the encoded data stored in the exception (see the /reason/
622-- argument given to the 'exit' primitive). If the handler cannot be applied,
623-- the exception will be re-thrown.
625-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
626-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
627catchExit :: forall a b . (Show a, Serializable a)
629 -> (ProcessId -> a -> Process b)
631catchExit act exitHandler = catch act handleExit
633 handleExit ex@(ProcessExitException from msg) =
634 if messageFingerprint msg == fingerprint (undefined :: a)
635 then exitHandler from decoded
636 else liftIO $ throwIO ex
639 -- Make sure the value is fully decoded so that we don't hang to
640 -- bytestrings if the caller doesn't use the value immediately
641 !decoded = decode (messageEncoding msg)
643-- | Lift 'Control.Exception.catches' (almost).
645-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
646-- message, a handler must accept inputs of the expected type. In order to
647-- handle a list of potentially different handlers (and therefore input types),
648-- a handler passed to 'catchesExit' must accept 'Message' and return
649-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
651-- See 'maybeHandleMessage' and 'Message' for more details.
652catchesExit :: Process b
653 -> [(ProcessId -> Message -> (Process (Maybe b)))]
655catchesExit act handlers = catch act ((flip handleExit) handlers)
657 handleExit :: ProcessExitException
658 -> [(ProcessId -> Message -> Process (Maybe b))]
660 handleExit ex [] = liftIO $ throwIO ex
661 handleExit ex@(ProcessExitException from msg) (h:hs) = do
664 Nothing -> handleExit ex hs
667-- | Our own process ID
668getSelfPid :: Process ProcessId
669getSelfPid = processId <$> ask
671-- | Get the node ID of our local node
672getSelfNode :: Process NodeId
673getSelfNode = localNodeId . processNode <$> ask
675-- | Get information about the specified process
676getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
678 let them = processNodeId pid in do
680 dest <- mkNode them us
681 sendCtrlMsg dest $ GetInfo pid
683 match (\(p :: ProcessInfo) -> return $ Just p)
684 , match (\(_ :: ProcessInfoNone) -> return Nothing)
686 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
687 mkNode them us = case them == us of
688 True -> return Nothing
689 _ -> return $ Just them
691--------------------------------------------------------------------------------
692-- Monitoring and linking --
693--------------------------------------------------------------------------------
695-- | Link to a remote process (asynchronous)
697-- When process A links to process B (that is, process A calls
698-- @link pidB@) then an asynchronous exception will be thrown to process A
699-- when process B terminates (normally or abnormally), or when process A gets
700-- disconnected from process B. Although it is /technically/ possible to catch
701-- these exceptions, chances are if you find yourself trying to do so you should
702-- probably be using 'monitor' rather than 'link'. In particular, code such as
704-- > link pidB -- Link to process B
705-- > expect -- Wait for a message from process B
706-- > unlink pidB -- Unlink again
708-- doesn't quite do what one might expect: if process B sends a message to
709-- process A, and /subsequently terminates/, then process A might or might not
710-- be terminated too, depending on whether the exception is thrown before or
711-- after the 'unlink' (i.e., this code has a race condition).
713-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
714-- call to 'link' has no effect.
716-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
717-- Linking makes no distinction between normal and abnormal termination of
718-- the remote process.
719link :: ProcessId -> Process ()
720link = sendCtrlMsg Nothing . Link . ProcessIdentifier
722-- | Monitor another process (asynchronous)
724-- When process A monitors process B (that is, process A calls
725-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
726-- when process B terminates (normally or abnormally), or when process A gets
727-- disconnected from process B. You receive this message like any other (using
728-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
729-- 'DiedDisconnect', etc.).
731-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
732-- multiple monitors are set up, multiple notifications will be delivered
733-- and monitors can be disabled individually using 'unmonitor'.
734monitor :: ProcessId -> Process MonitorRef
735monitor = monitor' . ProcessIdentifier
737-- | Establishes temporary monitoring of another process.
739-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
740-- of @code@. Note: although monitoring is no longer active when
741-- @withMonitor@ returns, there might still be unreceived monitor
742-- messages in the queue.
744withMonitor :: ProcessId -> Process a -> Process a
745withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
746 -- unmonitor blocks waiting for the response, so there's a possibility
747 -- that an exception might interrupt withMonitor before the unmonitor
748 -- has completed. I think that's better than making the unmonitor
753-- This is synchronous in the sense that once it returns you are guaranteed
754-- that no exception will be raised if the remote process dies. However, it is
755-- asynchronous in the sense that we do not wait for a response from the remote
757unlink :: ProcessId -> Process ()
760 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
764-- | Remove a node link
766-- This has the same synchronous/asynchronous nature as 'unlink'.
767unlinkNode :: NodeId -> Process ()
770 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
774-- | Remove a channel (send port) link
776-- This has the same synchronous/asynchronous nature as 'unlink'.
777unlinkPort :: SendPort a -> Process ()
779 unlinkPortAsync sport
780 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
786-- This has the same synchronous/asynchronous nature as 'unlink'.
787unmonitor :: MonitorRef -> Process ()
790 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
794--------------------------------------------------------------------------------
795-- Exception handling --
796--------------------------------------------------------------------------------
798-- | Lift 'Control.Exception.catch'
799catch :: Exception e => Process a -> (e -> Process a) -> Process a
802 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
804-- | Lift 'Control.Exception.try'
805try :: Exception e => Process a -> Process (Either e a)
808 liftIO $ Ex.try (runLocalProcess lproc p)
810-- | Lift 'Control.Exception.mask'
811mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
814 liftIO $ Ex.mask $ \restore ->
815 runLocalProcess lproc (p (liftRestore restore))
817 liftRestore :: (forall a. IO a -> IO a)
818 -> (forall a. Process a -> Process a)
819 liftRestore restoreIO = \p2 -> do
821 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
823-- | Lift 'Control.Exception.onException'
824onException :: Process a -> Process b -> Process a
825onException p what = p `catch` \e -> do _ <- what
826 liftIO $ throwIO (e :: SomeException)
828-- | Lift 'Control.Exception.bracket'
829bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
830bracket before after thing =
831 mask $ \restore -> do
833 r <- restore (thing a) `onException` after a
837-- | Lift 'Control.Exception.bracket_'
838bracket_ :: Process a -> Process b -> Process c -> Process c
839bracket_ before after thing = bracket before (const after) (const thing)
841-- | Lift 'Control.Exception.finally'
842finally :: Process a -> Process b -> Process a
843finally a sequel = bracket_ (return ()) sequel a
845-- | You need this when using 'catches'
846data Handler a = forall e . Exception e => Handler (e -> Process a)
848instance Functor Handler where
849 fmap f (Handler h) = Handler (fmap f . h)
851-- | Lift 'Control.Exception.catches'
852catches :: Process a -> [Handler a] -> Process a
853catches proc handlers = proc `catch` catchesHandler handlers
855catchesHandler :: [Handler a] -> SomeException -> Process a
856catchesHandler handlers e = foldr tryHandler (throw e) handlers
857 where tryHandler (Handler handler) res
858 = case fromException e of
859 Just e' -> handler e'
862--------------------------------------------------------------------------------
864--------------------------------------------------------------------------------
866-- | Like 'expect' but with a timeout
867expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
868expectTimeout n = receiveTimeout n [match return]
870-- | Asynchronous version of 'spawn'
872-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
873spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
874spawnAsync nid proc = do
875 spawnRef <- getSpawnRef
876 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
879-- | Monitor a node (asynchronous)
880monitorNode :: NodeId -> Process MonitorRef
882 monitor' . NodeIdentifier
884-- | Monitor a typed channel (asynchronous)
885monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
886monitorPort (SendPort cid) =
887 monitor' (SendPortIdentifier cid)
889-- | Remove a monitor (asynchronous)
890unmonitorAsync :: MonitorRef -> Process ()
892 sendCtrlMsg Nothing . Unmonitor
894-- | Link to a node (asynchronous)
895linkNode :: NodeId -> Process ()
896linkNode = link' . NodeIdentifier
898-- | Link to a channel (asynchronous)
899linkPort :: SendPort a -> Process ()
900linkPort (SendPort cid) =
901 link' (SendPortIdentifier cid)
903-- | Remove a process link (asynchronous)
904unlinkAsync :: ProcessId -> Process ()
906 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
908-- | Remove a node link (asynchronous)
909unlinkNodeAsync :: NodeId -> Process ()
911 sendCtrlMsg Nothing . Unlink . NodeIdentifier
913-- | Remove a channel (send port) link (asynchronous)
914unlinkPortAsync :: SendPort a -> Process ()
915unlinkPortAsync (SendPort cid) =
916 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
918--------------------------------------------------------------------------------
920--------------------------------------------------------------------------------
924-- @say message@ sends a message (time, pid of the current process, message)
925-- to the process registered as 'logger'. By default, this process simply
926-- sends the string to 'stderr'. Individual Cloud Haskell backends might
927-- replace this with a different logger process, however.
928say :: String -> Process ()
930 now <- liftIO getCurrentTime
932 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
934--------------------------------------------------------------------------------
936--------------------------------------------------------------------------------
938-- | Register a process with the local registry (asynchronous).
939-- This version will wait until a response is gotten from the
940-- management process. The name must not already be registered.
941-- The process need not be on this node.
942-- A bad registration will result in a 'ProcessRegistrationException'
944-- The process to be registered does not have to be local itself.
945register :: String -> ProcessId -> Process ()
946register = registerImpl False
948-- | Like 'register', but will replace an existing registration.
949-- The name must already be registered.
950reregister :: String -> ProcessId -> Process ()
951reregister = registerImpl True
953registerImpl :: Bool -> String -> ProcessId -> Process ()
954registerImpl force label pid = do
956 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
957 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
958 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
961-- | Register a process with a remote registry (asynchronous).
963-- The process to be registered does not have to live on the same remote node.
964-- Reply wil come in the form of a 'RegisterReply' message
966-- See comments in 'whereisRemoteAsync'
967registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
968registerRemoteAsync nid label pid =
969 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
971reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
972reregisterRemoteAsync nid label pid =
973 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
975-- | Remove a process from the local registry (asynchronous).
976-- This version will wait until a response is gotten from the
977-- management process. The name must already be registered.
978unregister :: String -> Process ()
981 sendCtrlMsg Nothing (Register label mynid Nothing False)
982 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
983 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
986-- | Deal with the result from an attempted registration or unregistration
987-- by throwing an exception if necessary
988handleRegistrationReply :: String -> Bool -> Process ()
989handleRegistrationReply label ok =
991 liftIO $ throwIO $ ProcessRegistrationException label
993-- | Remove a process from a remote registry (asynchronous).
995-- Reply wil come in the form of a 'RegisterReply' message
997-- See comments in 'whereisRemoteAsync'
998unregisterRemoteAsync :: NodeId -> String -> Process ()
999unregisterRemoteAsync nid label =
1000 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1002-- | Query the local process registry
1003whereis :: String -> Process (Maybe ProcessId)
1005 sendCtrlMsg Nothing (WhereIs label)
1006 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1007 (\(WhereIsReply _ mPid) -> return mPid)
1010-- | Query a remote process registry (asynchronous)
1012-- Reply will come in the form of a 'WhereIsReply' message.
1014-- There is currently no synchronous version of 'whereisRemoteAsync': if
1015-- you implement one yourself, be sure to take into account that the remote
1016-- node might die or get disconnect before it can respond (i.e. you should
1017-- use 'monitorNode' and take appropriate action when you receive a
1018-- 'NodeMonitorNotification').
1019whereisRemoteAsync :: NodeId -> String -> Process ()
1020whereisRemoteAsync nid label =
1021 sendCtrlMsg (Just nid) (WhereIs label)
1023-- | Named send to a process in the local registry (asynchronous)
1024nsend :: Serializable a => String -> a -> Process ()
1026 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1028-- | Named send to a process in the local registry (asynchronous).
1029-- This function makes /no/ attempt to serialize and (in the case when the
1030-- destination process resides on the same local node) therefore ensure that
1031-- the payload is fully evaluated before it is delivered.
1032unsafeNSend :: Serializable a => String -> a -> Process ()
1033unsafeNSend = Unsafe.nsend
1035-- | Named send to a process in a remote registry (asynchronous)
1036nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1037nsendRemote nid label msg =
1038 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1040--------------------------------------------------------------------------------
1042--------------------------------------------------------------------------------
1044-- | Resolve a static value
1045unStatic :: Typeable a => Static a -> Process a
1047 rtable <- remoteTable . processNode <$> ask
1048 case Static.unstatic rtable static of
1049 Left err -> fail $ "Could not resolve static value: " ++ err
1052-- | Resolve a closure
1053unClosure :: Typeable a => Closure a -> Process a
1054unClosure closure = do
1055 rtable <- remoteTable . processNode <$> ask
1056 case Static.unclosure rtable closure of
1057 Left err -> fail $ "Could not resolve closure: " ++ err
1060--------------------------------------------------------------------------------
1062--------------------------------------------------------------------------------
1064-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1065-- message passing. However, when network connections get disrupted this
1066-- illusion cannot always be maintained. Once a network connection breaks (even
1067-- temporarily) no further communication on that connection will be possible.
1068-- For example, if process A sends a message to process B, and A is then
1069-- notified (by monitor notification) that it got disconnected from B, A will
1070-- not be able to send any further messages to B, /unless/ A explicitly
1071-- indicates that it is acceptable to attempt to reconnect to B using the
1072-- Cloud Haskell 'reconnect' primitive.
1074-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1075-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1076-- then receives a monitor notification that its connection to B has been lost,
1077-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1078-- and m3 but not m2.
1080-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1081-- to attempt to reconnect on the next send/. In particular, if no further
1082-- communication attempts are made to B then A can use reconnect to clean up
1083-- its connection to B.
1084reconnect :: ProcessId -> Process ()
1087 node <- processNode <$> ask
1088 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1090-- | Reconnect to a sendport. See 'reconnect' for more information.
1091reconnectPort :: SendPort a -> Process ()
1092reconnectPort them = do
1094 node <- processNode <$> ask
1095 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1097--------------------------------------------------------------------------------
1098-- Auxiliary functions --
1099--------------------------------------------------------------------------------
1101sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1103 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1105sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1106sendChanLocal spId msg =
1107 -- we *must* fully serialize/encode the message here, because
1108 -- attempting to use `unsafeCoerce' in the node controller
1109 -- won't work since we know nothing about the required type
1110 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1112getMonitorRefFor :: Identifier -> Process MonitorRef
1113getMonitorRefFor ident = do
1115 liftIO $ modifyMVar (processState proc) $ \st -> do
1116 let counter = st ^. monitorCounter
1117 return ( monitorCounter ^: (+ 1) $ st
1118 , MonitorRef ident counter
1121getSpawnRef :: Process SpawnRef
1124 liftIO $ modifyMVar (processState proc) $ \st -> do
1125 let counter = st ^. spawnCounter
1126 return ( spawnCounter ^: (+ 1) $ st
1130-- | Monitor a process/node/channel
1131monitor' :: Identifier -> Process MonitorRef
1133 monitorRef <- getMonitorRefFor ident
1134 sendCtrlMsg Nothing $ Monitor monitorRef
1137-- | Link to a process/node/channel
1138link' :: Identifier -> Process ()
1139link' = 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