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
40 -- * Process management
42 , ProcessTerminationException(..)
48 -- keep the exception constructor hidden, so that handling exit
49 -- reasons /must/ take place via the 'catchExit' family of primitives
50 , ProcessExitException()
55 -- * Monitoring and linking
70 , reregisterRemoteAsync
71 , unregisterRemoteAsync
77 -- * Exception handling
100 -- * Internal Exports
104#if ! MIN_VERSION_base(4,6,0)
105import Prelude hiding (catch)
108import Data.Binary (decode)
109import Data.Time.Clock (getCurrentTime)
110import Data.Time.Format (formatTime)
111import System.Locale (defaultTimeLocale)
112import System.Timeout (timeout)
113import Control.Monad (when)
114import Control.Monad.Reader (ask)
115import Control.Monad.IO.Class (MonadIO, liftIO)
116import Control.Applicative ((<$>))
117import Control.Exception (Exception(..), throw, throwIO, SomeException)
118import qualified Control.Exception as Ex (catch, mask, try)
119import Control.Distributed.Process.Internal.StrictMVar
124import Control.Concurrent.STM
133import Control.Distributed.Process.Internal.CQueue
138import Control.Distributed.Process.Serializable (Serializable, fingerprint)
139import Data.Accessor ((^.), (^:), (^=))
140import Control.Distributed.Static
144import Data.Rank1Typeable (Typeable)
145import qualified Control.Distributed.Static as Static (unstatic, unclosure)
146import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
147import Control.Distributed.Process.Internal.Types
166 , ProcessExitException(..)
168 , DidUnlinkProcess(..)
173 , ProcessRegistrationException(..)
175 , ProcessInfoNone(..)
177 , createUnencodedMessage
179 , ImplicitReconnect(NoImplicitReconnect)
184import Control.Distributed.Process.Internal.Messaging
191import Control.Distributed.Process.Internal.Trace.Types
195import Control.Distributed.Process.Internal.WeakTQueue
203--------------------------------------------------------------------------------
205--------------------------------------------------------------------------------
208send :: Serializable a => ProcessId -> a -> Process ()
209-- This requires a lookup on every send. If we want to avoid that we need to
210-- modify serializable to allow for stateful (IO) deserialization.
213 let us = processId proc
214 node = processNode proc
215 nodeId = localNodeId node
216 destNode = (processNodeId them) in do
217 case destNode == nodeId of
218 True -> sendLocal them msg
219 False -> liftIO $ sendMessage (processNode proc)
220 (ProcessIdentifier (processId proc))
221 (ProcessIdentifier them)
224 -- We do not fire the trace event until after the sending is complete;
225 -- In the remote case, 'sendMessage' can block in the networking stack.
226 liftIO $ traceEvent (localTracer node)
227 (TraceEvSent them us (createUnencodedMessage msg))
229-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
230-- and (in the case when the destination process resides on the same local
231-- node) therefore ensure that the payload is fully evaluated before it is
233unsafeSend :: Serializable a => ProcessId -> a -> Process ()
234unsafeSend = Unsafe.send
236-- | Wait for a message of a specific type
237expect :: forall a. Serializable a => Process a
238expect = receiveWait [match return]
240--------------------------------------------------------------------------------
242--------------------------------------------------------------------------------
244-- | Create a new typed channel
245newChan :: Serializable a => Process (SendPort a, ReceivePort a)
248 liftIO . modifyMVar (processState proc) $ \st -> do
249 let lcid = st ^. channelCounter
250 let cid = SendPortId { sendPortProcessId = processId proc
251 , sendPortLocalId = lcid
253 let sport = SendPort cid
254 chan <- liftIO newTQueueIO
255 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
256 let rport = ReceivePort $ readTQueue chan
257 let tch = TypedChannel chan'
258 return ( (channelCounter ^: (+ 1))
259 . (typedChannelWithId lcid ^= Just tch)
264 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
265 finalizer st lcid = modifyMVar_ st $
266 return . (typedChannelWithId lcid ^= Nothing)
268-- | Send a message on a typed channel
269sendChan :: Serializable a => SendPort a -> a -> Process ()
270sendChan (SendPort cid) msg = do
272 let node = localNodeId (processNode proc)
273 destNode = processNodeId (sendPortProcessId cid) in do
274 case destNode == node of
275 True -> sendChanLocal cid msg
277 liftIO $ sendBinary (processNode proc)
278 (ProcessIdentifier (processId proc))
279 (SendPortIdentifier cid)
283-- | Send a message on a typed channel. This function makes /no/ attempt to
284-- serialize and (in the case when the @ReceivePort@ resides on the same local
285-- node) therefore ensure that the payload is fully evaluated before it is
287unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
288unsafeSendChan = Unsafe.sendChan
290-- | Wait for a message on a typed channel
291receiveChan :: Serializable a => ReceivePort a -> Process a
292receiveChan = liftIO . atomically . receiveSTM
294-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
295-- non-blocking check for a message.
296receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
297receiveChanTimeout 0 ch = liftIO . atomically $
298 (Just <$> receiveSTM ch) `orElse` return Nothing
299receiveChanTimeout n ch = liftIO . timeout n . atomically $
302-- | Merge a list of typed channels.
304-- The result port is left-biased: if there are messages available on more
305-- than one port, the first available message is returned.
306mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
307mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
309-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
311mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
312mergePortsRR = \ps -> do
313 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
314 return $ ReceivePort (rr psVar)
318 rotate (x:xs) = xs ++ [x]
320 rr :: TVar [STM a] -> STM a
323 a <- foldr1 orElse ps
324 writeTVar psVar (rotate ps)
327--------------------------------------------------------------------------------
328-- Advanced messaging --
329--------------------------------------------------------------------------------
331-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
332newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
334-- | Test the matches in order against each message in the queue
335receiveWait :: [Match b] -> Process b
337 queue <- processQueue <$> ask
338 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
341-- | Like 'receiveWait' but with a timeout.
343-- If the timeout is zero do a non-blocking check for matching messages. A
344-- non-zero timeout is applied only when waiting for incoming messages (that is,
345-- /after/ we have checked the messages that are already in the mailbox).
346receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
347receiveTimeout t ms = do
348 queue <- processQueue <$> ask
349 let blockSpec = if t == 0 then NonBlocking else Timeout t
350 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
352 Nothing -> return Nothing
353 Just proc -> Just <$> proc
355matchChan :: ReceivePort a -> (a -> Process b) -> Match b
356matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
358-- | Match against any message of the right type
359match :: forall a b. Serializable a => (a -> Process b) -> Match b
360match = matchIf (const True)
362-- | Match against any message of the right type that satisfies a predicate
363matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
364matchIf c p = Match $ MatchMsg $ \msg ->
365 case messageFingerprint msg == fingerprint (undefined :: a) of
368 (UnencodedMessage _ m) ->
369 let m' = unsafeCoerce m :: a in
373 (EncodedMessage _ _) ->
374 if (c decoded) then Just (p decoded) else Nothing
377 -- Make sure the value is fully decoded so that we don't hang to
378 -- bytestrings when the process calling 'matchIf' doesn't process
379 -- the values immediately
380 !decoded = decode (messageEncoding msg)
382-- | Match against any message, regardless of the underlying (contained) type
383matchMessage :: (Message -> Process Message) -> Match Message
384matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
386-- | Match against any message (regardless of underlying type) that satisfies a predicate
387matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
388matchMessageIf c p = Match $ MatchMsg $ \msg ->
393-- | Forward a raw 'Message' to the given 'ProcessId'.
394forward :: Message -> ProcessId -> Process ()
397 let node = processNode proc
399 nid = localNodeId node
400 destNode = (processNodeId them) in do
401 case destNode == nid of
402 True -> sendCtrlMsg Nothing (LocalSend them msg)
403 False -> liftIO $ sendPayload (processNode proc)
404 (ProcessIdentifier (processId proc))
405 (ProcessIdentifier them)
407 (messageToPayload msg)
408 -- We do not fire the trace event until after the sending is complete;
409 -- In the remote case, 'sendMessage' can block in the networking stack.
410 liftIO $ traceEvent (localTracer node)
411 (TraceEvSent them us msg)
414-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
415-- 'Serializable' - like the datum they contain - but also note, deserialising
416-- a 'Message' will yield a 'Message', not the type within it! To obtain the
417-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
422-- send self (wrapMessage "blah")
423-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
424-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
425-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
428wrapMessage :: Serializable a => a -> Message
429wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
431-- [note Serializable UnencodedMessage]
432-- if we attempt to serialise an UnencodedMessage, it will be converted to an
433-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
434-- Primitives.hs for the gory details
436-- | Attempt to unwrap a raw 'Message'.
437-- If the type of the decoded message payload matches the expected type, the
438-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
441-- This expression, for example, will evaluate to @Nothing@
442-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
444-- Whereas this expression, will yield @Just "foo"@
445-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
447unwrapMessage :: forall a. Serializable a => Message -> Process (Maybe a)
449 case messageFingerprint msg == fingerprint (undefined :: a) of
450 False -> return Nothing
452 (UnencodedMessage _ m) ->
453 let m' = unsafeCoerce m :: a
455 (EncodedMessage _ _) ->
456 return (Just (decoded))
458 decoded :: a -- note [decoding]
459 !decoded = decode (messageEncoding msg)
461-- | Attempt to handle a raw 'Message'.
462-- If the type of the message matches the type of the first argument to
463-- the supplied expression, then the message will be decoded and the expression
464-- evaluated against its value. If this runtime type checking fails however,
465-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
466-- evaluation proceeds, the resulting value with be wrapped with @Just@.
468-- Intended for use in `catchesExit` and `matchAny` primitives.
470handleMessage :: forall a b. (Serializable a)
471 => Message -> (a -> Process b) -> Process (Maybe b)
472handleMessage msg proc = do
473 case messageFingerprint msg == fingerprint (undefined :: a) of
474 False -> return Nothing
476 (UnencodedMessage _ m) ->
477 let m' = unsafeCoerce m :: a -- note [decoding]
478 in do { r <- proc m'; return (Just r) }
479 (EncodedMessage _ _) ->
480 do { r <- proc (decoded :: a); return (Just r) }
482 decoded :: a -- note [decoding]
483 !decoded = decode (messageEncoding msg)
485-- | Match against an arbitrary message. 'matchAny' removes the first available
486-- message from the process mailbox. To handle arbitrary /raw/ messages once
487-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
489matchAny :: forall b. (Message -> Process b) -> Match b
490matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
492-- | Match against an arbitrary message. Intended for use with 'handleMessage'
493-- and 'unwrapMessage', this function /only/ removes a message from the process
494-- mailbox, /if/ the supplied condition matches. The success (or failure) of
495-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
496-- here, i.e., if the condition evaluates to @True@ then the message will
497-- be removed from the process mailbox and decoded, but that does /not/
498-- guarantee that an expression passed to @handleMessage@ will pass the
499-- runtime type checks and therefore be evaluated.
501matchAnyIf :: forall a b. (Serializable a)
503 -> (Message -> Process b)
505matchAnyIf c p = Match $ MatchMsg $ \msg ->
506 case messageFingerprint msg == fingerprint (undefined :: a) of
507 True | check -> Just (p msg)
512 (EncodedMessage _ _) -> c decoded
513 (UnencodedMessage _ m') -> c (unsafeCoerce m')
515 decoded :: a -- note [decoding]
516 !decoded = decode (messageEncoding msg)
520For an EncodedMessage, we need to ensure the value is fully decoded so that
521we don't hang to bytestrings if the calling process doesn't evaluate
522immediately. For UnencodedMessage we know (because the fingerprint comparison
523succeeds) that unsafeCoerce will not fail.
526-- | Remove any message from the queue
527matchUnknown :: Process b -> Match b
528matchUnknown p = Match $ MatchMsg (const (Just p))
530-- | Receives messages and forwards them to @pid@ if @p msg == True@.
531delegate :: ProcessId -> (Message -> Bool) -> Process ()
534 matchAny (\m -> case (p m) of
535 True -> forward m pid
540-- | A straight relay that simply forwards all messages to the supplied pid.
541relay :: ProcessId -> Process ()
542relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
544-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
545-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
546-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
547-- fails, no action is taken and the proxy process will continue running.
548proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
552 next <- handleMessage m proc
554 Just True -> forward m pid
555 Just False -> return () -- explicitly ignored
556 Nothing -> return ()) -- un-routable
560--------------------------------------------------------------------------------
561-- Process management --
562--------------------------------------------------------------------------------
564-- | Thrown by 'terminate'
565data ProcessTerminationException = ProcessTerminationException
566 deriving (Show, Typeable)
568instance Exception ProcessTerminationException
570-- | Terminate immediately (throws a ProcessTerminationException)
571terminate :: Process a
572terminate = liftIO $ throwIO ProcessTerminationException
575-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
576die :: Serializable a => a -> Process b
578-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
579-- could be decoded by a handler passed to 'catchExit', re-thrown or even
580-- passed to another process via the tracing mechanism.
582 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
584-- | Forceful request to kill a process. Where 'exit' provides an exception
585-- that can be caught and handled, 'kill' throws an unexposed exception type
586-- which cannot be handled explicitly (by type).
587kill :: ProcessId -> String -> Process ()
588-- NOTE: We send the message to our local node controller, which will then
589-- forward it to a remote node controller (if applicable). Sending it directly
590-- to a remote node controller means that that the message may overtake a
591-- 'monitor' or 'link' request.
592kill them reason = sendCtrlMsg Nothing (Kill them reason)
594-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
595-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
596-- manner can be handled using the 'catchExit' family of functions.
597exit :: Serializable a => ProcessId -> a -> Process ()
598-- NOTE: We send the message to our local node controller, which will then
599-- forward it to a remote node controller (if applicable). Sending it directly
600-- to a remote node controller means that that the message may overtake a
601-- 'monitor' or 'link' request.
602exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
604-- | Catches 'ProcessExitException'. The handler will not be applied unless its
605-- type matches the encoded data stored in the exception (see the /reason/
606-- argument given to the 'exit' primitive). If the handler cannot be applied,
607-- the exception will be re-thrown.
609-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
610-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
611catchExit :: forall a b . (Show a, Serializable a)
613 -> (ProcessId -> a -> Process b)
615catchExit act exitHandler = catch act handleExit
617 handleExit ex@(ProcessExitException from msg) =
618 if messageFingerprint msg == fingerprint (undefined :: a)
619 then exitHandler from decoded
620 else liftIO $ throwIO ex
623 -- Make sure the value is fully decoded so that we don't hang to
624 -- bytestrings if the caller doesn't use the value immediately
625 !decoded = decode (messageEncoding msg)
627-- | Lift 'Control.Exception.catches' (almost).
629-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
630-- message, a handler must accept an input of the expected type. In order to
631-- handle a list of potentially different handlers (and therefore input types),
632-- a handler passed to 'catchesExit' must accept 'AbstractMessage' and return
633-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
635-- See 'maybeHandleMessage' and 'AsbtractMessage' for more details.
636catchesExit :: Process b
637 -> [(ProcessId -> Message -> (Process (Maybe b)))]
639catchesExit act handlers = catch act ((flip handleExit) handlers)
641 handleExit :: ProcessExitException
642 -> [(ProcessId -> Message -> Process (Maybe b))]
644 handleExit ex [] = liftIO $ throwIO ex
645 handleExit ex@(ProcessExitException from msg) (h:hs) = do
648 Nothing -> handleExit ex hs
651-- | Our own process ID
652getSelfPid :: Process ProcessId
653getSelfPid = processId <$> ask
655-- | Get the node ID of our local node
656getSelfNode :: Process NodeId
657getSelfNode = localNodeId . processNode <$> ask
659-- | Get information about the specified process
660getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
662 let them = processNodeId pid in do
664 dest <- mkNode them us
665 sendCtrlMsg dest $ GetInfo pid
667 match (\(p :: ProcessInfo) -> return $ Just p)
668 , match (\(_ :: ProcessInfoNone) -> return Nothing)
670 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
671 mkNode them us = case them == us of
672 True -> return Nothing
673 _ -> return $ Just them
675--------------------------------------------------------------------------------
676-- Monitoring and linking --
677--------------------------------------------------------------------------------
679-- | Link to a remote process (asynchronous)
681-- When process A links to process B (that is, process A calls
682-- @link pidB@) then an asynchronous exception will be thrown to process A
683-- when process B terminates (normally or abnormally), or when process A gets
684-- disconnected from process B. Although it is /technically/ possible to catch
685-- these exceptions, chances are if you find yourself trying to do so you should
686-- probably be using 'monitor' rather than 'link'. In particular, code such as
688-- > link pidB -- Link to process B
689-- > expect -- Wait for a message from process B
690-- > unlink pidB -- Unlink again
692-- doesn't quite do what one might expect: if process B sends a message to
693-- process A, and /subsequently terminates/, then process A might or might not
694-- be terminated too, depending on whether the exception is thrown before or
695-- after the 'unlink' (i.e., this code has a race condition).
697-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
698-- call to 'link' has no effect.
700-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
701-- Linking makes no distinction between normal and abnormal termination of
702-- the remote process.
703link :: ProcessId -> Process ()
704link = sendCtrlMsg Nothing . Link . ProcessIdentifier
706-- | Monitor another process (asynchronous)
708-- When process A monitors process B (that is, process A calls
709-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
710-- when process B terminates (normally or abnormally), or when process A gets
711-- disconnected from process B. You receive this message like any other (using
712-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
713-- 'DiedDisconnect', etc.).
715-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
716-- multiple monitors are set up, multiple notifications will be delivered
717-- and monitors can be disabled individually using 'unmonitor'.
718monitor :: ProcessId -> Process MonitorRef
719monitor = monitor' . ProcessIdentifier
721-- | Establishes temporary monitoring of another process.
723-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
724-- of @code@. Note: although monitoring is no longer active when
725-- @withMonitor@ returns, there might still be unreceived monitor
726-- messages in the queue.
728withMonitor :: ProcessId -> Process a -> Process a
729withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
730 -- unmonitor blocks waiting for the response, so there's a possibility
731 -- that an exception might interrupt withMonitor before the unmonitor
732 -- has completed. I think that's better than making the unmonitor
737-- This is synchronous in the sense that once it returns you are guaranteed
738-- that no exception will be raised if the remote process dies. However, it is
739-- asynchronous in the sense that we do not wait for a response from the remote
741unlink :: ProcessId -> Process ()
744 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
748-- | Remove a node link
750-- This has the same synchronous/asynchronous nature as 'unlink'.
751unlinkNode :: NodeId -> Process ()
754 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
758-- | Remove a channel (send port) link
760-- This has the same synchronous/asynchronous nature as 'unlink'.
761unlinkPort :: SendPort a -> Process ()
763 unlinkPortAsync sport
764 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
770-- This has the same synchronous/asynchronous nature as 'unlink'.
771unmonitor :: MonitorRef -> Process ()
774 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
778--------------------------------------------------------------------------------
779-- Exception handling --
780--------------------------------------------------------------------------------
782-- | Lift 'Control.Exception.catch'
783catch :: Exception e => Process a -> (e -> Process a) -> Process a
786 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
788-- | Lift 'Control.Exception.try'
789try :: Exception e => Process a -> Process (Either e a)
792 liftIO $ Ex.try (runLocalProcess lproc p)
794-- | Lift 'Control.Exception.mask'
795mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
798 liftIO $ Ex.mask $ \restore ->
799 runLocalProcess lproc (p (liftRestore restore))
801 liftRestore :: (forall a. IO a -> IO a)
802 -> (forall a. Process a -> Process a)
803 liftRestore restoreIO = \p2 -> do
805 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
807-- | Lift 'Control.Exception.onException'
808onException :: Process a -> Process b -> Process a
809onException p what = p `catch` \e -> do _ <- what
810 liftIO $ throwIO (e :: SomeException)
812-- | Lift 'Control.Exception.bracket'
813bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
814bracket before after thing =
815 mask $ \restore -> do
817 r <- restore (thing a) `onException` after a
821-- | Lift 'Control.Exception.bracket_'
822bracket_ :: Process a -> Process b -> Process c -> Process c
823bracket_ before after thing = bracket before (const after) (const thing)
825-- | Lift 'Control.Exception.finally'
826finally :: Process a -> Process b -> Process a
827finally a sequel = bracket_ (return ()) sequel a
829-- | You need this when using 'catches'
830data Handler a = forall e . Exception e => Handler (e -> Process a)
832instance Functor Handler where
833 fmap f (Handler h) = Handler (fmap f . h)
835-- | Lift 'Control.Exception.catches'
836catches :: Process a -> [Handler a] -> Process a
837catches proc handlers = proc `catch` catchesHandler handlers
839catchesHandler :: [Handler a] -> SomeException -> Process a
840catchesHandler handlers e = foldr tryHandler (throw e) handlers
841 where tryHandler (Handler handler) res
842 = case fromException e of
843 Just e' -> handler e'
846--------------------------------------------------------------------------------
848--------------------------------------------------------------------------------
850-- | Like 'expect' but with a timeout
851expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
852expectTimeout n = receiveTimeout n [match return]
854-- | Asynchronous version of 'spawn'
856-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
857spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
858spawnAsync nid proc = do
859 spawnRef <- getSpawnRef
860 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
863-- | Monitor a node (asynchronous)
864monitorNode :: NodeId -> Process MonitorRef
866 monitor' . NodeIdentifier
868-- | Monitor a typed channel (asynchronous)
869monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
870monitorPort (SendPort cid) =
871 monitor' (SendPortIdentifier cid)
873-- | Remove a monitor (asynchronous)
874unmonitorAsync :: MonitorRef -> Process ()
876 sendCtrlMsg Nothing . Unmonitor
878-- | Link to a node (asynchronous)
879linkNode :: NodeId -> Process ()
880linkNode = link' . NodeIdentifier
882-- | Link to a channel (asynchronous)
883linkPort :: SendPort a -> Process ()
884linkPort (SendPort cid) =
885 link' (SendPortIdentifier cid)
887-- | Remove a process link (asynchronous)
888unlinkAsync :: ProcessId -> Process ()
890 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
892-- | Remove a node link (asynchronous)
893unlinkNodeAsync :: NodeId -> Process ()
895 sendCtrlMsg Nothing . Unlink . NodeIdentifier
897-- | Remove a channel (send port) link (asynchronous)
898unlinkPortAsync :: SendPort a -> Process ()
899unlinkPortAsync (SendPort cid) =
900 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
902--------------------------------------------------------------------------------
904--------------------------------------------------------------------------------
908-- @say message@ sends a message (time, pid of the current process, message)
909-- to the process registered as 'logger'. By default, this process simply
910-- sends the string to 'stderr'. Individual Cloud Haskell backends might
911-- replace this with a different logger process, however.
912say :: String -> Process ()
914 now <- liftIO getCurrentTime
916 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
918--------------------------------------------------------------------------------
920--------------------------------------------------------------------------------
922-- | Register a process with the local registry (asynchronous).
923-- This version will wait until a response is gotten from the
924-- management process. The name must not already be registered.
925-- The process need not be on this node.
926-- A bad registration will result in a 'ProcessRegistrationException'
928-- The process to be registered does not have to be local itself.
929register :: String -> ProcessId -> Process ()
930register = registerImpl False
932-- | Like 'register', but will replace an existing registration.
933-- The name must already be registered.
934reregister :: String -> ProcessId -> Process ()
935reregister = registerImpl True
937registerImpl :: Bool -> String -> ProcessId -> Process ()
938registerImpl force label pid = do
940 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
941 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
942 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
945-- | Register a process with a remote registry (asynchronous).
947-- The process to be registered does not have to live on the same remote node.
948-- Reply wil come in the form of a 'RegisterReply' message
950-- See comments in 'whereisRemoteAsync'
951registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
952registerRemoteAsync nid label pid =
953 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
955reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
956reregisterRemoteAsync nid label pid =
957 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
959-- | Remove a process from the local registry (asynchronous).
960-- This version will wait until a response is gotten from the
961-- management process. The name must already be registered.
962unregister :: String -> Process ()
965 sendCtrlMsg Nothing (Register label mynid Nothing False)
966 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
967 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
970-- | Deal with the result from an attempted registration or unregistration
971-- by throwing an exception if necessary
972handleRegistrationReply :: String -> Bool -> Process ()
973handleRegistrationReply label ok =
975 liftIO $ throwIO $ ProcessRegistrationException label
977-- | Remove a process from a remote registry (asynchronous).
979-- Reply wil come in the form of a 'RegisterReply' message
981-- See comments in 'whereisRemoteAsync'
982unregisterRemoteAsync :: NodeId -> String -> Process ()
983unregisterRemoteAsync nid label =
984 sendCtrlMsg (Just nid) (Register label nid Nothing False)
986-- | Query the local process registry
987whereis :: String -> Process (Maybe ProcessId)
989 sendCtrlMsg Nothing (WhereIs label)
990 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
991 (\(WhereIsReply _ mPid) -> return mPid)
994-- | Query a remote process registry (asynchronous)
996-- Reply will come in the form of a 'WhereIsReply' message.
998-- There is currently no synchronous version of 'whereisRemoteAsync': if
999-- you implement one yourself, be sure to take into account that the remote
1000-- node might die or get disconnect before it can respond (i.e. you should
1001-- use 'monitorNode' and take appropriate action when you receive a
1002-- 'NodeMonitorNotification').
1003whereisRemoteAsync :: NodeId -> String -> Process ()
1004whereisRemoteAsync nid label =
1005 sendCtrlMsg (Just nid) (WhereIs label)
1007-- | Named send to a process in the local registry (asynchronous)
1008nsend :: Serializable a => String -> a -> Process ()
1010 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1012-- | Named send to a process in the local registry (asynchronous).
1013-- This function makes /no/ attempt to serialize and (in the case when the
1014-- destination process resides on the same local node) therefore ensure that
1015-- the payload is fully evaluated before it is delivered.
1016unsafeNSend :: Serializable a => String -> a -> Process ()
1017unsafeNSend = Unsafe.nsend
1019-- | Named send to a process in a remote registry (asynchronous)
1020nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1021nsendRemote nid label msg =
1022 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1024--------------------------------------------------------------------------------
1026--------------------------------------------------------------------------------
1028-- | Resolve a static value
1029unStatic :: Typeable a => Static a -> Process a
1031 rtable <- remoteTable . processNode <$> ask
1032 case Static.unstatic rtable static of
1033 Left err -> fail $ "Could not resolve static value: " ++ err
1036-- | Resolve a closure
1037unClosure :: Typeable a => Closure a -> Process a
1038unClosure closure = do
1039 rtable <- remoteTable . processNode <$> ask
1040 case Static.unclosure rtable closure of
1041 Left err -> fail $ "Could not resolve closure: " ++ err
1044--------------------------------------------------------------------------------
1046--------------------------------------------------------------------------------
1048-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1049-- message passing. However, when network connections get disrupted this
1050-- illusion cannot always be maintained. Once a network connection breaks (even
1051-- temporarily) no further communication on that connection will be possible.
1052-- For example, if process A sends a message to process B, and A is then
1053-- notified (by monitor notification) that it got disconnected from B, A will
1054-- not be able to send any further messages to B, /unless/ A explicitly
1055-- indicates that it is acceptable to attempt to reconnect to B using the
1056-- Cloud Haskell 'reconnect' primitive.
1058-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1059-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1060-- then receives a monitor notification that its connection to B has been lost,
1061-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1062-- and m3 but not m2.
1064-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1065-- to attempt to reconnect on the next send/. In particular, if no further
1066-- communication attempts are made to B then A can use reconnect to clean up
1067-- its connection to B.
1068reconnect :: ProcessId -> Process ()
1071 node <- processNode <$> ask
1072 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1074-- | Reconnect to a sendport. See 'reconnect' for more information.
1075reconnectPort :: SendPort a -> Process ()
1076reconnectPort them = do
1078 node <- processNode <$> ask
1079 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1081--------------------------------------------------------------------------------
1082-- Auxiliary functions --
1083--------------------------------------------------------------------------------
1085sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1087 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1089sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1090sendChanLocal spId msg =
1091 -- we *must* fully serialize/encode the message here, because
1092 -- attempting to use `unsafeCoerce' in the node controller
1093 -- won't work since we know nothing about the required type
1094 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1096getMonitorRefFor :: Identifier -> Process MonitorRef
1097getMonitorRefFor ident = do
1099 liftIO $ modifyMVar (processState proc) $ \st -> do
1100 let counter = st ^. monitorCounter
1101 return ( monitorCounter ^: (+ 1) $ st
1102 , MonitorRef ident counter
1105getSpawnRef :: Process SpawnRef
1108 liftIO $ modifyMVar (processState proc) $ \st -> do
1109 let counter = st ^. spawnCounter
1110 return ( spawnCounter ^: (+ 1) $ st
1114-- | Monitor a process/node/channel
1115monitor' :: Identifier -> Process MonitorRef
1117 monitorRef <- getMonitorRefFor ident
1118 sendCtrlMsg Nothing $ Monitor monitorRef
1121-- | Link to a process/node/channel
1122link' :: Identifier -> Process ()
1123link' = 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