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.Internal.Trace.Types
198import Control.Distributed.Process.Internal.WeakTQueue
206--------------------------------------------------------------------------------
208--------------------------------------------------------------------------------
211send :: Serializable a => ProcessId -> a -> Process ()
212-- This requires a lookup on every send. If we want to avoid that we need to
213-- modify serializable to allow for stateful (IO) deserialization.
216 let us = processId proc
217 node = processNode proc
218 nodeId = localNodeId node
219 destNode = (processNodeId them) in do
220 case destNode == nodeId of
221 True -> sendLocal them msg
222 False -> liftIO $ sendMessage (processNode proc)
223 (ProcessIdentifier (processId proc))
224 (ProcessIdentifier them)
227 -- We do not fire the trace event until after the sending is complete;
228 -- In the remote case, 'sendMessage' can block in the networking stack.
229 liftIO $ traceEvent (localEventBus node)
230 (TraceEvSent them us (createUnencodedMessage msg))
232-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
233-- and (in the case when the destination process resides on the same local
234-- node) therefore ensure that the payload is fully evaluated before it is
236unsafeSend :: Serializable a => ProcessId -> a -> Process ()
237unsafeSend = Unsafe.send
239-- | Wait for a message of a specific type
240expect :: forall a. Serializable a => Process a
241expect = receiveWait [match return]
243--------------------------------------------------------------------------------
245--------------------------------------------------------------------------------
247-- | Create a new typed channel
248newChan :: Serializable a => Process (SendPort a, ReceivePort a)
251 liftIO . modifyMVar (processState proc) $ \st -> do
252 let lcid = st ^. channelCounter
253 let cid = SendPortId { sendPortProcessId = processId proc
254 , sendPortLocalId = lcid
256 let sport = SendPort cid
257 chan <- liftIO newTQueueIO
258 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
259 let rport = ReceivePort $ readTQueue chan
260 let tch = TypedChannel chan'
261 return ( (channelCounter ^: (+ 1))
262 . (typedChannelWithId lcid ^= Just tch)
267 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
268 finalizer st lcid = modifyMVar_ st $
269 return . (typedChannelWithId lcid ^= Nothing)
271-- | Send a message on a typed channel
272sendChan :: Serializable a => SendPort a -> a -> Process ()
273sendChan (SendPort cid) msg = do
275 let node = localNodeId (processNode proc)
276 destNode = processNodeId (sendPortProcessId cid) in do
277 case destNode == node of
278 True -> sendChanLocal cid msg
280 liftIO $ sendBinary (processNode proc)
281 (ProcessIdentifier (processId proc))
282 (SendPortIdentifier cid)
286-- | Send a message on a typed channel. This function makes /no/ attempt to
287-- serialize and (in the case when the @ReceivePort@ resides on the same local
288-- node) therefore ensure that the payload is fully evaluated before it is
290unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
291unsafeSendChan = Unsafe.sendChan
293-- | Wait for a message on a typed channel
294receiveChan :: Serializable a => ReceivePort a -> Process a
295receiveChan = liftIO . atomically . receiveSTM
297-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
298-- non-blocking check for a message.
299receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
300receiveChanTimeout 0 ch = liftIO . atomically $
301 (Just <$> receiveSTM ch) `orElse` return Nothing
302receiveChanTimeout n ch = liftIO . timeout n . atomically $
305-- | Merge a list of typed channels.
307-- The result port is left-biased: if there are messages available on more
308-- than one port, the first available message is returned.
309mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
310mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
312-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
314mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
315mergePortsRR = \ps -> do
316 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
317 return $ ReceivePort (rr psVar)
321 rotate (x:xs) = xs ++ [x]
323 rr :: TVar [STM a] -> STM a
326 a <- foldr1 orElse ps
327 writeTVar psVar (rotate ps)
330--------------------------------------------------------------------------------
331-- Advanced messaging --
332--------------------------------------------------------------------------------
334-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
335newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
337-- | Test the matches in order against each message in the queue
338receiveWait :: [Match b] -> Process b
340 queue <- processQueue <$> ask
341 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
344-- | Like 'receiveWait' but with a timeout.
346-- If the timeout is zero do a non-blocking check for matching messages. A
347-- non-zero timeout is applied only when waiting for incoming messages (that is,
348-- /after/ we have checked the messages that are already in the mailbox).
349receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
350receiveTimeout t ms = do
351 queue <- processQueue <$> ask
352 let blockSpec = if t == 0 then NonBlocking else Timeout t
353 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
355 Nothing -> return Nothing
356 Just proc -> Just <$> proc
358matchChan :: ReceivePort a -> (a -> Process b) -> Match b
359matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
361-- | Match against any message of the right type
362match :: forall a b. Serializable a => (a -> Process b) -> Match b
363match = matchIf (const True)
365-- | Match against any message of the right type that satisfies a predicate
366matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
367matchIf c p = Match $ MatchMsg $ \msg ->
368 case messageFingerprint msg == fingerprint (undefined :: a) of
371 (UnencodedMessage _ m) ->
372 let m' = unsafeCoerce m :: a in
376 (EncodedMessage _ _) ->
377 if (c decoded) then Just (p decoded) else Nothing
380 -- Make sure the value is fully decoded so that we don't hang to
381 -- bytestrings when the process calling 'matchIf' doesn't process
382 -- the values immediately
383 !decoded = decode (messageEncoding msg)
385-- | Match against any message, regardless of the underlying (contained) type
386matchMessage :: (Message -> Process Message) -> Match Message
387matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
389-- | Match against any message (regardless of underlying type) that satisfies a predicate
390matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
391matchMessageIf c p = Match $ MatchMsg $ \msg ->
396-- | Forward a raw 'Message' to the given 'ProcessId'.
397forward :: Message -> ProcessId -> Process ()
400 let node = processNode proc
402 nid = localNodeId node
403 destNode = (processNodeId them) in do
404 case destNode == nid of
405 True -> sendCtrlMsg Nothing (LocalSend them msg)
406 False -> liftIO $ sendPayload (processNode proc)
407 (ProcessIdentifier (processId proc))
408 (ProcessIdentifier them)
410 (messageToPayload msg)
411 -- We do not fire the trace event until after the sending is complete;
412 -- In the remote case, 'sendMessage' can block in the networking stack.
413 liftIO $ traceEvent (localEventBus node)
414 (TraceEvSent them us msg)
417-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
418-- 'Serializable' - like the datum they contain - but also note, deserialising
419-- a 'Message' will yield a 'Message', not the type within it! To obtain the
420-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
425-- send self (wrapMessage "blah")
426-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
427-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
428-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
431wrapMessage :: Serializable a => a -> Message
432wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
434-- [note Serializable UnencodedMessage]
435-- if we attempt to serialise an UnencodedMessage, it will be converted to an
436-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
437-- Primitives.hs for the gory details
439-- | Attempt to unwrap a raw 'Message'.
440-- If the type of the decoded message payload matches the expected type, the
441-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
444-- This expression, for example, will evaluate to @Nothing@
445-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
447-- Whereas this expression, will yield @Just "foo"@
448-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
450unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
452 case messageFingerprint msg == fingerprint (undefined :: a) of
453 False -> return Nothing :: m (Maybe a)
455 (UnencodedMessage _ ms) ->
456 let ms' = unsafeCoerce ms :: a
458 (EncodedMessage _ _) ->
459 return (Just (decoded))
461 decoded :: a -- note [decoding]
462 !decoded = decode (messageEncoding msg)
464-- | Attempt to handle a raw 'Message'.
465-- If the type of the message matches the type of the first argument to
466-- the supplied expression, then the message will be decoded and the expression
467-- evaluated against its value. If this runtime type checking fails however,
468-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
469-- evaluation proceeds, the resulting value with be wrapped with @Just@.
471-- Intended for use in `catchesExit` and `matchAny` primitives.
473handleMessage :: forall m a b. (Monad m, Serializable a)
474 => Message -> (a -> m b) -> m (Maybe b)
475handleMessage msg proc = handleMessageIf msg (const True) proc
477handleMessageIf :: forall m a b . (Monad m, Serializable a)
482handleMessageIf msg c proc = do
483 case messageFingerprint msg == fingerprint (undefined :: a) of
484 False -> return Nothing :: m (Maybe b)
486 (UnencodedMessage _ ms) ->
487 let ms' = unsafeCoerce ms :: a in
489 True -> do { r <- proc ms'; return (Just r) }
490 False -> return Nothing :: m (Maybe b)
491 (EncodedMessage _ _) ->
493 True -> do { r <- proc (decoded :: a); return (Just r) }
494 False -> return Nothing :: m (Maybe b)
496 decoded :: a -- note [decoding]
497 !decoded = decode (messageEncoding msg)
499-- | Match against an arbitrary message. 'matchAny' removes the first available
500-- message from the process mailbox. To handle arbitrary /raw/ messages once
501-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
503matchAny :: forall b. (Message -> Process b) -> Match b
504matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
506-- | Match against an arbitrary message. Intended for use with 'handleMessage'
507-- and 'unwrapMessage', this function /only/ removes a message from the process
508-- mailbox, /if/ the supplied condition matches. The success (or failure) of
509-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
510-- here, i.e., if the condition evaluates to @True@ then the message will
511-- be removed from the process mailbox and decoded, but that does /not/
512-- guarantee that an expression passed to @handleMessage@ will pass the
513-- runtime type checks and therefore be evaluated.
515matchAnyIf :: forall a b. (Serializable a)
517 -> (Message -> Process b)
519matchAnyIf c p = Match $ MatchMsg $ \msg ->
520 case messageFingerprint msg == fingerprint (undefined :: a) of
521 True | check -> Just (p msg)
526 (EncodedMessage _ _) -> c decoded
527 (UnencodedMessage _ m') -> c (unsafeCoerce m')
529 decoded :: a -- note [decoding]
530 !decoded = decode (messageEncoding msg)
534For an EncodedMessage, we need to ensure the value is fully decoded so that
535we don't hang to bytestrings if the calling process doesn't evaluate
536immediately. For UnencodedMessage we know (because the fingerprint comparison
537succeeds) that unsafeCoerce will not fail.
540-- | Remove any message from the queue
541matchUnknown :: Process b -> Match b
542matchUnknown p = Match $ MatchMsg (const (Just p))
544-- | Receives messages and forwards them to @pid@ if @p msg == True@.
545delegate :: ProcessId -> (Message -> Bool) -> Process ()
548 matchAny (\m -> case (p m) of
549 True -> forward m pid
554-- | A straight relay that simply forwards all messages to the supplied pid.
555relay :: ProcessId -> Process ()
556relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
558-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
559-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
560-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
561-- fails, no action is taken and the proxy process will continue running.
562proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
566 next <- handleMessage m proc
568 Just True -> forward m pid
569 Just False -> return () -- explicitly ignored
570 Nothing -> return ()) -- un-routable
574--------------------------------------------------------------------------------
575-- Process management --
576--------------------------------------------------------------------------------
578-- | Thrown by 'terminate'
579data ProcessTerminationException = ProcessTerminationException
580 deriving (Show, Typeable)
582instance Exception ProcessTerminationException
584-- | Terminate immediately (throws a ProcessTerminationException)
585terminate :: Process a
586terminate = liftIO $ throwIO ProcessTerminationException
589-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
590die :: Serializable a => a -> Process b
592-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
593-- could be decoded by a handler passed to 'catchExit', re-thrown or even
594-- passed to another process via the tracing mechanism.
596 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
598-- | Forceful request to kill a process. Where 'exit' provides an exception
599-- that can be caught and handled, 'kill' throws an unexposed exception type
600-- which cannot be handled explicitly (by type).
601kill :: ProcessId -> String -> Process ()
602-- NOTE: We send the message to our local node controller, which will then
603-- forward it to a remote node controller (if applicable). Sending it directly
604-- to a remote node controller means that that the message may overtake a
605-- 'monitor' or 'link' request.
606kill them reason = sendCtrlMsg Nothing (Kill them reason)
608-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
609-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
610-- manner can be handled using the 'catchExit' family of functions.
611exit :: Serializable a => ProcessId -> a -> Process ()
612-- NOTE: We send the message to our local node controller, which will then
613-- forward it to a remote node controller (if applicable). Sending it directly
614-- to a remote node controller means that that the message may overtake a
615-- 'monitor' or 'link' request.
616exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
618-- | Catches 'ProcessExitException'. The handler will not be applied unless its
619-- type matches the encoded data stored in the exception (see the /reason/
620-- argument given to the 'exit' primitive). If the handler cannot be applied,
621-- the exception will be re-thrown.
623-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
624-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
625catchExit :: forall a b . (Show a, Serializable a)
627 -> (ProcessId -> a -> Process b)
629catchExit act exitHandler = catch act handleExit
631 handleExit ex@(ProcessExitException from msg) =
632 if messageFingerprint msg == fingerprint (undefined :: a)
633 then exitHandler from decoded
634 else liftIO $ throwIO ex
637 -- Make sure the value is fully decoded so that we don't hang to
638 -- bytestrings if the caller doesn't use the value immediately
639 !decoded = decode (messageEncoding msg)
641-- | Lift 'Control.Exception.catches' (almost).
643-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
644-- message, a handler must accept inputs of the expected type. In order to
645-- handle a list of potentially different handlers (and therefore input types),
646-- a handler passed to 'catchesExit' must accept 'Message' and return
647-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
649-- See 'maybeHandleMessage' and 'Message' for more details.
650catchesExit :: Process b
651 -> [(ProcessId -> Message -> (Process (Maybe b)))]
653catchesExit act handlers = catch act ((flip handleExit) handlers)
655 handleExit :: ProcessExitException
656 -> [(ProcessId -> Message -> Process (Maybe b))]
658 handleExit ex [] = liftIO $ throwIO ex
659 handleExit ex@(ProcessExitException from msg) (h:hs) = do
662 Nothing -> handleExit ex hs
665-- | Our own process ID
666getSelfPid :: Process ProcessId
667getSelfPid = processId <$> ask
669-- | Get the node ID of our local node
670getSelfNode :: Process NodeId
671getSelfNode = localNodeId . processNode <$> ask
673-- | Get information about the specified process
674getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
676 let them = processNodeId pid in do
678 dest <- mkNode them us
679 sendCtrlMsg dest $ GetInfo pid
681 match (\(p :: ProcessInfo) -> return $ Just p)
682 , match (\(_ :: ProcessInfoNone) -> return Nothing)
684 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
685 mkNode them us = case them == us of
686 True -> return Nothing
687 _ -> return $ Just them
689--------------------------------------------------------------------------------
690-- Monitoring and linking --
691--------------------------------------------------------------------------------
693-- | Link to a remote process (asynchronous)
695-- When process A links to process B (that is, process A calls
696-- @link pidB@) then an asynchronous exception will be thrown to process A
697-- when process B terminates (normally or abnormally), or when process A gets
698-- disconnected from process B. Although it is /technically/ possible to catch
699-- these exceptions, chances are if you find yourself trying to do so you should
700-- probably be using 'monitor' rather than 'link'. In particular, code such as
702-- > link pidB -- Link to process B
703-- > expect -- Wait for a message from process B
704-- > unlink pidB -- Unlink again
706-- doesn't quite do what one might expect: if process B sends a message to
707-- process A, and /subsequently terminates/, then process A might or might not
708-- be terminated too, depending on whether the exception is thrown before or
709-- after the 'unlink' (i.e., this code has a race condition).
711-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
712-- call to 'link' has no effect.
714-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
715-- Linking makes no distinction between normal and abnormal termination of
716-- the remote process.
717link :: ProcessId -> Process ()
718link = sendCtrlMsg Nothing . Link . ProcessIdentifier
720-- | Monitor another process (asynchronous)
722-- When process A monitors process B (that is, process A calls
723-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
724-- when process B terminates (normally or abnormally), or when process A gets
725-- disconnected from process B. You receive this message like any other (using
726-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
727-- 'DiedDisconnect', etc.).
729-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
730-- multiple monitors are set up, multiple notifications will be delivered
731-- and monitors can be disabled individually using 'unmonitor'.
732monitor :: ProcessId -> Process MonitorRef
733monitor = monitor' . ProcessIdentifier
735-- | Establishes temporary monitoring of another process.
737-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
738-- of @code@. Note: although monitoring is no longer active when
739-- @withMonitor@ returns, there might still be unreceived monitor
740-- messages in the queue.
742withMonitor :: ProcessId -> Process a -> Process a
743withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
744 -- unmonitor blocks waiting for the response, so there's a possibility
745 -- that an exception might interrupt withMonitor before the unmonitor
746 -- has completed. I think that's better than making the unmonitor
751-- This is synchronous in the sense that once it returns you are guaranteed
752-- that no exception will be raised if the remote process dies. However, it is
753-- asynchronous in the sense that we do not wait for a response from the remote
755unlink :: ProcessId -> Process ()
758 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
762-- | Remove a node link
764-- This has the same synchronous/asynchronous nature as 'unlink'.
765unlinkNode :: NodeId -> Process ()
768 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
772-- | Remove a channel (send port) link
774-- This has the same synchronous/asynchronous nature as 'unlink'.
775unlinkPort :: SendPort a -> Process ()
777 unlinkPortAsync sport
778 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
784-- This has the same synchronous/asynchronous nature as 'unlink'.
785unmonitor :: MonitorRef -> Process ()
788 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
792--------------------------------------------------------------------------------
793-- Exception handling --
794--------------------------------------------------------------------------------
796-- | Lift 'Control.Exception.catch'
797catch :: Exception e => Process a -> (e -> Process a) -> Process a
800 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
802-- | Lift 'Control.Exception.try'
803try :: Exception e => Process a -> Process (Either e a)
806 liftIO $ Ex.try (runLocalProcess lproc p)
808-- | Lift 'Control.Exception.mask'
809mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
812 liftIO $ Ex.mask $ \restore ->
813 runLocalProcess lproc (p (liftRestore restore))
815 liftRestore :: (forall a. IO a -> IO a)
816 -> (forall a. Process a -> Process a)
817 liftRestore restoreIO = \p2 -> do
819 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
821-- | Lift 'Control.Exception.onException'
822onException :: Process a -> Process b -> Process a
823onException p what = p `catch` \e -> do _ <- what
824 liftIO $ throwIO (e :: SomeException)
826-- | Lift 'Control.Exception.bracket'
827bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
828bracket before after thing =
829 mask $ \restore -> do
831 r <- restore (thing a) `onException` after a
835-- | Lift 'Control.Exception.bracket_'
836bracket_ :: Process a -> Process b -> Process c -> Process c
837bracket_ before after thing = bracket before (const after) (const thing)
839-- | Lift 'Control.Exception.finally'
840finally :: Process a -> Process b -> Process a
841finally a sequel = bracket_ (return ()) sequel a
843-- | You need this when using 'catches'
844data Handler a = forall e . Exception e => Handler (e -> Process a)
846instance Functor Handler where
847 fmap f (Handler h) = Handler (fmap f . h)
849-- | Lift 'Control.Exception.catches'
850catches :: Process a -> [Handler a] -> Process a
851catches proc handlers = proc `catch` catchesHandler handlers
853catchesHandler :: [Handler a] -> SomeException -> Process a
854catchesHandler handlers e = foldr tryHandler (throw e) handlers
855 where tryHandler (Handler handler) res
856 = case fromException e of
857 Just e' -> handler e'
860--------------------------------------------------------------------------------
862--------------------------------------------------------------------------------
864-- | Like 'expect' but with a timeout
865expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
866expectTimeout n = receiveTimeout n [match return]
868-- | Asynchronous version of 'spawn'
870-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
871spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
872spawnAsync nid proc = do
873 spawnRef <- getSpawnRef
874 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
877-- | Monitor a node (asynchronous)
878monitorNode :: NodeId -> Process MonitorRef
880 monitor' . NodeIdentifier
882-- | Monitor a typed channel (asynchronous)
883monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
884monitorPort (SendPort cid) =
885 monitor' (SendPortIdentifier cid)
887-- | Remove a monitor (asynchronous)
888unmonitorAsync :: MonitorRef -> Process ()
890 sendCtrlMsg Nothing . Unmonitor
892-- | Link to a node (asynchronous)
893linkNode :: NodeId -> Process ()
894linkNode = link' . NodeIdentifier
896-- | Link to a channel (asynchronous)
897linkPort :: SendPort a -> Process ()
898linkPort (SendPort cid) =
899 link' (SendPortIdentifier cid)
901-- | Remove a process link (asynchronous)
902unlinkAsync :: ProcessId -> Process ()
904 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
906-- | Remove a node link (asynchronous)
907unlinkNodeAsync :: NodeId -> Process ()
909 sendCtrlMsg Nothing . Unlink . NodeIdentifier
911-- | Remove a channel (send port) link (asynchronous)
912unlinkPortAsync :: SendPort a -> Process ()
913unlinkPortAsync (SendPort cid) =
914 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
916--------------------------------------------------------------------------------
918--------------------------------------------------------------------------------
922-- @say message@ sends a message (time, pid of the current process, message)
923-- to the process registered as 'logger'. By default, this process simply
924-- sends the string to 'stderr'. Individual Cloud Haskell backends might
925-- replace this with a different logger process, however.
926say :: String -> Process ()
928 now <- liftIO getCurrentTime
930 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
932--------------------------------------------------------------------------------
934--------------------------------------------------------------------------------
936-- | Register a process with the local registry (asynchronous).
937-- This version will wait until a response is gotten from the
938-- management process. The name must not already be registered.
939-- The process need not be on this node.
940-- A bad registration will result in a 'ProcessRegistrationException'
942-- The process to be registered does not have to be local itself.
943register :: String -> ProcessId -> Process ()
944register = registerImpl False
946-- | Like 'register', but will replace an existing registration.
947-- The name must already be registered.
948reregister :: String -> ProcessId -> Process ()
949reregister = registerImpl True
951registerImpl :: Bool -> String -> ProcessId -> Process ()
952registerImpl force label pid = do
954 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
955 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
956 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
959-- | Register a process with a remote registry (asynchronous).
961-- The process to be registered does not have to live on the same remote node.
962-- Reply wil come in the form of a 'RegisterReply' message
964-- See comments in 'whereisRemoteAsync'
965registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
966registerRemoteAsync nid label pid =
967 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
969reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
970reregisterRemoteAsync nid label pid =
971 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
973-- | Remove a process from the local registry (asynchronous).
974-- This version will wait until a response is gotten from the
975-- management process. The name must already be registered.
976unregister :: String -> Process ()
979 sendCtrlMsg Nothing (Register label mynid Nothing False)
980 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
981 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
984-- | Deal with the result from an attempted registration or unregistration
985-- by throwing an exception if necessary
986handleRegistrationReply :: String -> Bool -> Process ()
987handleRegistrationReply label ok =
989 liftIO $ throwIO $ ProcessRegistrationException label
991-- | Remove a process from a remote registry (asynchronous).
993-- Reply wil come in the form of a 'RegisterReply' message
995-- See comments in 'whereisRemoteAsync'
996unregisterRemoteAsync :: NodeId -> String -> Process ()
997unregisterRemoteAsync nid label =
998 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1000-- | Query the local process registry
1001whereis :: String -> Process (Maybe ProcessId)
1003 sendCtrlMsg Nothing (WhereIs label)
1004 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1005 (\(WhereIsReply _ mPid) -> return mPid)
1008-- | Query a remote process registry (asynchronous)
1010-- Reply will come in the form of a 'WhereIsReply' message.
1012-- There is currently no synchronous version of 'whereisRemoteAsync': if
1013-- you implement one yourself, be sure to take into account that the remote
1014-- node might die or get disconnect before it can respond (i.e. you should
1015-- use 'monitorNode' and take appropriate action when you receive a
1016-- 'NodeMonitorNotification').
1017whereisRemoteAsync :: NodeId -> String -> Process ()
1018whereisRemoteAsync nid label =
1019 sendCtrlMsg (Just nid) (WhereIs label)
1021-- | Named send to a process in the local registry (asynchronous)
1022nsend :: Serializable a => String -> a -> Process ()
1024 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1026-- | Named send to a process in the local registry (asynchronous).
1027-- This function makes /no/ attempt to serialize and (in the case when the
1028-- destination process resides on the same local node) therefore ensure that
1029-- the payload is fully evaluated before it is delivered.
1030unsafeNSend :: Serializable a => String -> a -> Process ()
1031unsafeNSend = Unsafe.nsend
1033-- | Named send to a process in a remote registry (asynchronous)
1034nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1035nsendRemote nid label msg =
1036 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1038--------------------------------------------------------------------------------
1040--------------------------------------------------------------------------------
1042-- | Resolve a static value
1043unStatic :: Typeable a => Static a -> Process a
1045 rtable <- remoteTable . processNode <$> ask
1046 case Static.unstatic rtable static of
1047 Left err -> fail $ "Could not resolve static value: " ++ err
1050-- | Resolve a closure
1051unClosure :: Typeable a => Closure a -> Process a
1052unClosure closure = do
1053 rtable <- remoteTable . processNode <$> ask
1054 case Static.unclosure rtable closure of
1055 Left err -> fail $ "Could not resolve closure: " ++ err
1058--------------------------------------------------------------------------------
1060--------------------------------------------------------------------------------
1062-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1063-- message passing. However, when network connections get disrupted this
1064-- illusion cannot always be maintained. Once a network connection breaks (even
1065-- temporarily) no further communication on that connection will be possible.
1066-- For example, if process A sends a message to process B, and A is then
1067-- notified (by monitor notification) that it got disconnected from B, A will
1068-- not be able to send any further messages to B, /unless/ A explicitly
1069-- indicates that it is acceptable to attempt to reconnect to B using the
1070-- Cloud Haskell 'reconnect' primitive.
1072-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1073-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1074-- then receives a monitor notification that its connection to B has been lost,
1075-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1076-- and m3 but not m2.
1078-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1079-- to attempt to reconnect on the next send/. In particular, if no further
1080-- communication attempts are made to B then A can use reconnect to clean up
1081-- its connection to B.
1082reconnect :: ProcessId -> Process ()
1085 node <- processNode <$> ask
1086 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1088-- | Reconnect to a sendport. See 'reconnect' for more information.
1089reconnectPort :: SendPort a -> Process ()
1090reconnectPort them = do
1092 node <- processNode <$> ask
1093 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1095--------------------------------------------------------------------------------
1096-- Auxiliary functions --
1097--------------------------------------------------------------------------------
1099sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1101 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1103sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1104sendChanLocal spId msg =
1105 -- we *must* fully serialize/encode the message here, because
1106 -- attempting to use `unsafeCoerce' in the node controller
1107 -- won't work since we know nothing about the required type
1108 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1110getMonitorRefFor :: Identifier -> Process MonitorRef
1111getMonitorRefFor ident = do
1113 liftIO $ modifyMVar (processState proc) $ \st -> do
1114 let counter = st ^. monitorCounter
1115 return ( monitorCounter ^: (+ 1) $ st
1116 , MonitorRef ident counter
1119getSpawnRef :: Process SpawnRef
1122 liftIO $ modifyMVar (processState proc) $ \st -> do
1123 let counter = st ^. spawnCounter
1124 return ( spawnCounter ^: (+ 1) $ st
1128-- | Monitor a process/node/channel
1129monitor' :: Identifier -> Process MonitorRef
1131 monitorRef <- getMonitorRefFor ident
1132 sendCtrlMsg Nothing $ Monitor monitorRef
1135-- | Link to a process/node/channel
1136link' :: Identifier -> Process ()
1137link' = 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