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 qualified Control.Distributed.Process.Management.Bus as Mx
197import Control.Distributed.Process.Management.Types
200import Control.Distributed.Process.Management.Trace.Types
203import Control.Distributed.Process.Internal.WeakTQueue
211--------------------------------------------------------------------------------
213--------------------------------------------------------------------------------
216send :: Serializable a => ProcessId -> a -> Process ()
217-- This requires a lookup on every send. If we want to avoid that we need to
218-- modify serializable to allow for stateful (IO) deserialization.
221 let us = processId proc
222 node = processNode proc
223 nodeId = localNodeId node
224 destNode = (processNodeId them) in do
225 case destNode == nodeId of
226 True -> sendLocal them msg
227 False -> liftIO $ sendMessage (processNode proc)
228 (ProcessIdentifier (processId proc))
229 (ProcessIdentifier them)
232 -- We do not fire the trace event until after sending completes, since
233 -- in the remote case, 'sendMessage' can block in the networking stack.
235 let msg = createUnencodedMessage $ (MxSent them us (createUnencodedMessage msg))
236 Mx.publishEvent (localEventBus node) msg
238-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
239-- and (in the case when the destination process resides on the same local
240-- node) therefore ensure that the payload is fully evaluated before it is
242unsafeSend :: Serializable a => ProcessId -> a -> Process ()
243unsafeSend = Unsafe.send
245-- | Wait for a message of a specific type
246expect :: forall a. Serializable a => Process a
247expect = receiveWait [match return]
249--------------------------------------------------------------------------------
251--------------------------------------------------------------------------------
253-- | Create a new typed channel
254newChan :: Serializable a => Process (SendPort a, ReceivePort a)
257 liftIO . modifyMVar (processState proc) $ \st -> do
258 let lcid = st ^. channelCounter
259 let cid = SendPortId { sendPortProcessId = processId proc
260 , sendPortLocalId = lcid
262 let sport = SendPort cid
263 chan <- liftIO newTQueueIO
264 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
265 let rport = ReceivePort $ readTQueue chan
266 let tch = TypedChannel chan'
267 return ( (channelCounter ^: (+ 1))
268 . (typedChannelWithId lcid ^= Just tch)
273 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
274 finalizer st lcid = modifyMVar_ st $
275 return . (typedChannelWithId lcid ^= Nothing)
277-- | Send a message on a typed channel
278sendChan :: Serializable a => SendPort a -> a -> Process ()
279sendChan (SendPort cid) msg = do
281 let node = localNodeId (processNode proc)
282 destNode = processNodeId (sendPortProcessId cid) in do
283 case destNode == node of
284 True -> sendChanLocal cid msg
286 liftIO $ sendBinary (processNode proc)
287 (ProcessIdentifier (processId proc))
288 (SendPortIdentifier cid)
292-- | Send a message on a typed channel. This function makes /no/ attempt to
293-- serialize and (in the case when the @ReceivePort@ resides on the same local
294-- node) therefore ensure that the payload is fully evaluated before it is
296unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
297unsafeSendChan = Unsafe.sendChan
299-- | Wait for a message on a typed channel
300receiveChan :: Serializable a => ReceivePort a -> Process a
301receiveChan = liftIO . atomically . receiveSTM
303-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
304-- non-blocking check for a message.
305receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
306receiveChanTimeout 0 ch = liftIO . atomically $
307 (Just <$> receiveSTM ch) `orElse` return Nothing
308receiveChanTimeout n ch = liftIO . timeout n . atomically $
311-- | Merge a list of typed channels.
313-- The result port is left-biased: if there are messages available on more
314-- than one port, the first available message is returned.
315mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
316mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
318-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
320mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
321mergePortsRR = \ps -> do
322 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
323 return $ ReceivePort (rr psVar)
327 rotate (x:xs) = xs ++ [x]
329 rr :: TVar [STM a] -> STM a
332 a <- foldr1 orElse ps
333 writeTVar psVar (rotate ps)
336--------------------------------------------------------------------------------
337-- Advanced messaging --
338--------------------------------------------------------------------------------
340-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
341newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
343-- | Test the matches in order against each message in the queue
344receiveWait :: [Match b] -> Process b
346 queue <- processQueue <$> ask
347 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
350-- | Like 'receiveWait' but with a timeout.
352-- If the timeout is zero do a non-blocking check for matching messages. A
353-- non-zero timeout is applied only when waiting for incoming messages (that is,
354-- /after/ we have checked the messages that are already in the mailbox).
355receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
356receiveTimeout t ms = do
357 queue <- processQueue <$> ask
358 let blockSpec = if t == 0 then NonBlocking else Timeout t
359 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
361 Nothing -> return Nothing
362 Just proc -> Just <$> proc
364matchChan :: ReceivePort a -> (a -> Process b) -> Match b
365matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
367-- | Match against any message of the right type
368match :: forall a b. Serializable a => (a -> Process b) -> Match b
369match = matchIf (const True)
371-- | Match against any message of the right type that satisfies a predicate
372matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
373matchIf c p = Match $ MatchMsg $ \msg ->
374 case messageFingerprint msg == fingerprint (undefined :: a) of
377 (UnencodedMessage _ m) ->
378 let m' = unsafeCoerce m :: a in
382 (EncodedMessage _ _) ->
383 if (c decoded) then Just (p decoded) else Nothing
386 -- Make sure the value is fully decoded so that we don't hang to
387 -- bytestrings when the process calling 'matchIf' doesn't process
388 -- the values immediately
389 !decoded = decode (messageEncoding msg)
391-- | Match against any message, regardless of the underlying (contained) type
392matchMessage :: (Message -> Process Message) -> Match Message
393matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
395-- | Match against any message (regardless of underlying type) that satisfies a predicate
396matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
397matchMessageIf c p = Match $ MatchMsg $ \msg ->
402-- | Forward a raw 'Message' to the given 'ProcessId'.
403forward :: Message -> ProcessId -> Process ()
406 let node = processNode proc
408 nid = localNodeId node
409 destNode = (processNodeId them) in do
410 case destNode == nid of
411 True -> sendCtrlMsg Nothing (LocalSend them msg)
412 False -> liftIO $ sendPayload (processNode proc)
413 (ProcessIdentifier (processId proc))
414 (ProcessIdentifier them)
416 (messageToPayload msg)
417 -- We do not fire the trace event until after the sending is complete;
418 -- In the remote case, 'sendMessage' can block in the networking stack.
419 liftIO $ Mx.publishEvent (localEventBus node)
420 (createUnencodedMessage (MxSent them us msg))
423-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
424-- 'Serializable' - like the datum they contain - but also note, deserialising
425-- a 'Message' will yield a 'Message', not the type within it! To obtain the
426-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
431-- send self (wrapMessage "blah")
432-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
433-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
434-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
437wrapMessage :: Serializable a => a -> Message
438wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
440-- [note Serializable UnencodedMessage]
441-- if we attempt to serialise an UnencodedMessage, it will be converted to an
442-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
443-- Primitives.hs for the gory details
445-- | Attempt to unwrap a raw 'Message'.
446-- If the type of the decoded message payload matches the expected type, the
447-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
450-- This expression, for example, will evaluate to @Nothing@
451-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
453-- Whereas this expression, will yield @Just "foo"@
454-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
456unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
458 case messageFingerprint msg == fingerprint (undefined :: a) of
459 False -> return Nothing :: m (Maybe a)
461 (UnencodedMessage _ ms) ->
462 let ms' = unsafeCoerce ms :: a
464 (EncodedMessage _ _) ->
465 return (Just (decoded))
467 decoded :: a -- note [decoding]
468 !decoded = decode (messageEncoding msg)
470-- | Attempt to handle a raw 'Message'.
471-- If the type of the message matches the type of the first argument to
472-- the supplied expression, then the message will be decoded and the expression
473-- evaluated against its value. If this runtime type checking fails however,
474-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
475-- evaluation proceeds, the resulting value with be wrapped with @Just@.
477-- Intended for use in `catchesExit` and `matchAny` primitives.
479handleMessage :: forall m a b. (Monad m, Serializable a)
480 => Message -> (a -> m b) -> m (Maybe b)
481handleMessage msg proc = handleMessageIf msg (const True) proc
483handleMessageIf :: forall m a b . (Monad m, Serializable a)
488handleMessageIf msg c proc = do
489 case messageFingerprint msg == fingerprint (undefined :: a) of
490 False -> return Nothing :: m (Maybe b)
492 (UnencodedMessage _ ms) ->
493 let ms' = unsafeCoerce ms :: a in
495 True -> do { r <- proc ms'; return (Just r) }
496 False -> return Nothing :: m (Maybe b)
497 (EncodedMessage _ _) ->
499 True -> do { r <- proc (decoded :: a); return (Just r) }
500 False -> return Nothing :: m (Maybe b)
502 decoded :: a -- note [decoding]
503 !decoded = decode (messageEncoding msg)
505-- | Match against an arbitrary message. 'matchAny' removes the first available
506-- message from the process mailbox. To handle arbitrary /raw/ messages once
507-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
509matchAny :: forall b. (Message -> Process b) -> Match b
510matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
512-- | Match against an arbitrary message. Intended for use with 'handleMessage'
513-- and 'unwrapMessage', this function /only/ removes a message from the process
514-- mailbox, /if/ the supplied condition matches. The success (or failure) of
515-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
516-- here, i.e., if the condition evaluates to @True@ then the message will
517-- be removed from the process mailbox and decoded, but that does /not/
518-- guarantee that an expression passed to @handleMessage@ will pass the
519-- runtime type checks and therefore be evaluated.
521matchAnyIf :: forall a b. (Serializable a)
523 -> (Message -> Process b)
525matchAnyIf c p = Match $ MatchMsg $ \msg ->
526 case messageFingerprint msg == fingerprint (undefined :: a) of
527 True | check -> Just (p msg)
532 (EncodedMessage _ _) -> c decoded
533 (UnencodedMessage _ m') -> c (unsafeCoerce m')
535 decoded :: a -- note [decoding]
536 !decoded = decode (messageEncoding msg)
540For an EncodedMessage, we need to ensure the value is fully decoded so that
541we don't hang to bytestrings if the calling process doesn't evaluate
542immediately. For UnencodedMessage we know (because the fingerprint comparison
543succeeds) that unsafeCoerce will not fail.
546-- | Remove any message from the queue
547matchUnknown :: Process b -> Match b
548matchUnknown p = Match $ MatchMsg (const (Just p))
550-- | Receives messages and forwards them to @pid@ if @p msg == True@.
551delegate :: ProcessId -> (Message -> Bool) -> Process ()
554 matchAny (\m -> case (p m) of
555 True -> forward m pid
560-- | A straight relay that simply forwards all messages to the supplied pid.
561relay :: ProcessId -> Process ()
562relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
564-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
565-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
566-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
567-- fails, no action is taken and the proxy process will continue running.
568proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
572 next <- handleMessage m proc
574 Just True -> forward m pid
575 Just False -> return () -- explicitly ignored
576 Nothing -> return ()) -- un-routable
580--------------------------------------------------------------------------------
581-- Process management --
582--------------------------------------------------------------------------------
584-- | Thrown by 'terminate'
585data ProcessTerminationException = ProcessTerminationException
586 deriving (Show, Typeable)
588instance Exception ProcessTerminationException
590-- | Terminate immediately (throws a ProcessTerminationException)
591terminate :: Process a
592terminate = liftIO $ throwIO ProcessTerminationException
595-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
596die :: Serializable a => a -> Process b
598-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
599-- could be decoded by a handler passed to 'catchExit', re-thrown or even
600-- passed to another process via the tracing mechanism.
602 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
604-- | Forceful request to kill a process. Where 'exit' provides an exception
605-- that can be caught and handled, 'kill' throws an unexposed exception type
606-- which cannot be handled explicitly (by type).
607kill :: ProcessId -> String -> Process ()
608-- NOTE: We send the message to our local node controller, which will then
609-- forward it to a remote node controller (if applicable). Sending it directly
610-- to a remote node controller means that that the message may overtake a
611-- 'monitor' or 'link' request.
612kill them reason = sendCtrlMsg Nothing (Kill them reason)
614-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
615-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
616-- manner can be handled using the 'catchExit' family of functions.
617exit :: Serializable a => ProcessId -> a -> Process ()
618-- NOTE: We send the message to our local node controller, which will then
619-- forward it to a remote node controller (if applicable). Sending it directly
620-- to a remote node controller means that that the message may overtake a
621-- 'monitor' or 'link' request.
622exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
624-- | Catches 'ProcessExitException'. The handler will not be applied unless its
625-- type matches the encoded data stored in the exception (see the /reason/
626-- argument given to the 'exit' primitive). If the handler cannot be applied,
627-- the exception will be re-thrown.
629-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
630-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
631catchExit :: forall a b . (Show a, Serializable a)
633 -> (ProcessId -> a -> Process b)
635catchExit act exitHandler = catch act handleExit
637 handleExit ex@(ProcessExitException from msg) =
638 if messageFingerprint msg == fingerprint (undefined :: a)
639 then exitHandler from decoded
640 else liftIO $ throwIO ex
643 -- Make sure the value is fully decoded so that we don't hang to
644 -- bytestrings if the caller doesn't use the value immediately
645 !decoded = decode (messageEncoding msg)
647-- | Lift 'Control.Exception.catches' (almost).
649-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
650-- message, a handler must accept inputs of the expected type. In order to
651-- handle a list of potentially different handlers (and therefore input types),
652-- a handler passed to 'catchesExit' must accept 'Message' and return
653-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
655-- See 'maybeHandleMessage' and 'Message' for more details.
656catchesExit :: Process b
657 -> [(ProcessId -> Message -> (Process (Maybe b)))]
659catchesExit act handlers = catch act ((flip handleExit) handlers)
661 handleExit :: ProcessExitException
662 -> [(ProcessId -> Message -> Process (Maybe b))]
664 handleExit ex [] = liftIO $ throwIO ex
665 handleExit ex@(ProcessExitException from msg) (h:hs) = do
668 Nothing -> handleExit ex hs
671-- | Our own process ID
672getSelfPid :: Process ProcessId
673getSelfPid = processId <$> ask
675-- | Get the node ID of our local node
676getSelfNode :: Process NodeId
677getSelfNode = localNodeId . processNode <$> ask
679-- | Get information about the specified process
680getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
682 let them = processNodeId pid in do
684 dest <- mkNode them us
685 sendCtrlMsg dest $ GetInfo pid
687 match (\(p :: ProcessInfo) -> return $ Just p)
688 , match (\(_ :: ProcessInfoNone) -> return Nothing)
690 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
691 mkNode them us = case them == us of
692 True -> return Nothing
693 _ -> return $ Just them
695--------------------------------------------------------------------------------
696-- Monitoring and linking --
697--------------------------------------------------------------------------------
699-- | Link to a remote process (asynchronous)
701-- When process A links to process B (that is, process A calls
702-- @link pidB@) then an asynchronous exception will be thrown to process A
703-- when process B terminates (normally or abnormally), or when process A gets
704-- disconnected from process B. Although it is /technically/ possible to catch
705-- these exceptions, chances are if you find yourself trying to do so you should
706-- probably be using 'monitor' rather than 'link'. In particular, code such as
708-- > link pidB -- Link to process B
709-- > expect -- Wait for a message from process B
710-- > unlink pidB -- Unlink again
712-- doesn't quite do what one might expect: if process B sends a message to
713-- process A, and /subsequently terminates/, then process A might or might not
714-- be terminated too, depending on whether the exception is thrown before or
715-- after the 'unlink' (i.e., this code has a race condition).
717-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
718-- call to 'link' has no effect.
720-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
721-- Linking makes no distinction between normal and abnormal termination of
722-- the remote process.
723link :: ProcessId -> Process ()
724link = sendCtrlMsg Nothing . Link . ProcessIdentifier
726-- | Monitor another process (asynchronous)
728-- When process A monitors process B (that is, process A calls
729-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
730-- when process B terminates (normally or abnormally), or when process A gets
731-- disconnected from process B. You receive this message like any other (using
732-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
733-- 'DiedDisconnect', etc.).
735-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
736-- multiple monitors are set up, multiple notifications will be delivered
737-- and monitors can be disabled individually using 'unmonitor'.
738monitor :: ProcessId -> Process MonitorRef
739monitor = monitor' . ProcessIdentifier
741-- | Establishes temporary monitoring of another process.
743-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
744-- of @code@. Note: although monitoring is no longer active when
745-- @withMonitor@ returns, there might still be unreceived monitor
746-- messages in the queue.
748withMonitor :: ProcessId -> Process a -> Process a
749withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
750 -- unmonitor blocks waiting for the response, so there's a possibility
751 -- that an exception might interrupt withMonitor before the unmonitor
752 -- has completed. I think that's better than making the unmonitor
757-- This is synchronous in the sense that once it returns you are guaranteed
758-- that no exception will be raised if the remote process dies. However, it is
759-- asynchronous in the sense that we do not wait for a response from the remote
761unlink :: ProcessId -> Process ()
764 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
768-- | Remove a node link
770-- This has the same synchronous/asynchronous nature as 'unlink'.
771unlinkNode :: NodeId -> Process ()
774 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
778-- | Remove a channel (send port) link
780-- This has the same synchronous/asynchronous nature as 'unlink'.
781unlinkPort :: SendPort a -> Process ()
783 unlinkPortAsync sport
784 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
790-- This has the same synchronous/asynchronous nature as 'unlink'.
791unmonitor :: MonitorRef -> Process ()
794 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
798--------------------------------------------------------------------------------
799-- Exception handling --
800--------------------------------------------------------------------------------
802-- | Lift 'Control.Exception.catch'
803catch :: Exception e => Process a -> (e -> Process a) -> Process a
806 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
808-- | Lift 'Control.Exception.try'
809try :: Exception e => Process a -> Process (Either e a)
812 liftIO $ Ex.try (runLocalProcess lproc p)
814-- | Lift 'Control.Exception.mask'
815mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
818 liftIO $ Ex.mask $ \restore ->
819 runLocalProcess lproc (p (liftRestore restore))
821 liftRestore :: (forall a. IO a -> IO a)
822 -> (forall a. Process a -> Process a)
823 liftRestore restoreIO = \p2 -> do
825 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
827-- | Lift 'Control.Exception.onException'
828onException :: Process a -> Process b -> Process a
829onException p what = p `catch` \e -> do _ <- what
830 liftIO $ throwIO (e :: SomeException)
832-- | Lift 'Control.Exception.bracket'
833bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
834bracket before after thing =
835 mask $ \restore -> do
837 r <- restore (thing a) `onException` after a
841-- | Lift 'Control.Exception.bracket_'
842bracket_ :: Process a -> Process b -> Process c -> Process c
843bracket_ before after thing = bracket before (const after) (const thing)
845-- | Lift 'Control.Exception.finally'
846finally :: Process a -> Process b -> Process a
847finally a sequel = bracket_ (return ()) sequel a
849-- | You need this when using 'catches'
850data Handler a = forall e . Exception e => Handler (e -> Process a)
852instance Functor Handler where
853 fmap f (Handler h) = Handler (fmap f . h)
855-- | Lift 'Control.Exception.catches'
856catches :: Process a -> [Handler a] -> Process a
857catches proc handlers = proc `catch` catchesHandler handlers
859catchesHandler :: [Handler a] -> SomeException -> Process a
860catchesHandler handlers e = foldr tryHandler (throw e) handlers
861 where tryHandler (Handler handler) res
862 = case fromException e of
863 Just e' -> handler e'
866--------------------------------------------------------------------------------
868--------------------------------------------------------------------------------
870-- | Like 'expect' but with a timeout
871expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
872expectTimeout n = receiveTimeout n [match return]
874-- | Asynchronous version of 'spawn'
876-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
877spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
878spawnAsync nid proc = do
879 spawnRef <- getSpawnRef
880 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
883-- | Monitor a node (asynchronous)
884monitorNode :: NodeId -> Process MonitorRef
886 monitor' . NodeIdentifier
888-- | Monitor a typed channel (asynchronous)
889monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
890monitorPort (SendPort cid) =
891 monitor' (SendPortIdentifier cid)
893-- | Remove a monitor (asynchronous)
894unmonitorAsync :: MonitorRef -> Process ()
896 sendCtrlMsg Nothing . Unmonitor
898-- | Link to a node (asynchronous)
899linkNode :: NodeId -> Process ()
900linkNode = link' . NodeIdentifier
902-- | Link to a channel (asynchronous)
903linkPort :: SendPort a -> Process ()
904linkPort (SendPort cid) =
905 link' (SendPortIdentifier cid)
907-- | Remove a process link (asynchronous)
908unlinkAsync :: ProcessId -> Process ()
910 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
912-- | Remove a node link (asynchronous)
913unlinkNodeAsync :: NodeId -> Process ()
915 sendCtrlMsg Nothing . Unlink . NodeIdentifier
917-- | Remove a channel (send port) link (asynchronous)
918unlinkPortAsync :: SendPort a -> Process ()
919unlinkPortAsync (SendPort cid) =
920 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
922--------------------------------------------------------------------------------
924--------------------------------------------------------------------------------
928-- @say message@ sends a message (time, pid of the current process, message)
929-- to the process registered as 'logger'. By default, this process simply
930-- sends the string to 'stderr'. Individual Cloud Haskell backends might
931-- replace this with a different logger process, however.
932say :: String -> Process ()
934 now <- liftIO getCurrentTime
936 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
938--------------------------------------------------------------------------------
940--------------------------------------------------------------------------------
942-- | Register a process with the local registry (asynchronous).
943-- This version will wait until a response is gotten from the
944-- management process. The name must not already be registered.
945-- The process need not be on this node.
946-- A bad registration will result in a 'ProcessRegistrationException'
948-- The process to be registered does not have to be local itself.
949register :: String -> ProcessId -> Process ()
950register = registerImpl False
952-- | Like 'register', but will replace an existing registration.
953-- The name must already be registered.
954reregister :: String -> ProcessId -> Process ()
955reregister = registerImpl True
957registerImpl :: Bool -> String -> ProcessId -> Process ()
958registerImpl force label pid = do
960 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
961 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
962 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
965-- | Register a process with a remote registry (asynchronous).
967-- The process to be registered does not have to live on the same remote node.
968-- Reply wil come in the form of a 'RegisterReply' message
970-- See comments in 'whereisRemoteAsync'
971registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
972registerRemoteAsync nid label pid =
973 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
975reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
976reregisterRemoteAsync nid label pid =
977 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
979-- | Remove a process from the local registry (asynchronous).
980-- This version will wait until a response is gotten from the
981-- management process. The name must already be registered.
982unregister :: String -> Process ()
985 sendCtrlMsg Nothing (Register label mynid Nothing False)
986 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
987 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
990-- | Deal with the result from an attempted registration or unregistration
991-- by throwing an exception if necessary
992handleRegistrationReply :: String -> Bool -> Process ()
993handleRegistrationReply label ok =
995 liftIO $ throwIO $ ProcessRegistrationException label
997-- | Remove a process from a remote registry (asynchronous).
999-- Reply wil come in the form of a 'RegisterReply' message
1001-- See comments in 'whereisRemoteAsync'
1002unregisterRemoteAsync :: NodeId -> String -> Process ()
1003unregisterRemoteAsync nid label =
1004 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1006-- | Query the local process registry
1007whereis :: String -> Process (Maybe ProcessId)
1009 sendCtrlMsg Nothing (WhereIs label)
1010 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1011 (\(WhereIsReply _ mPid) -> return mPid)
1014-- | Query a remote process registry (asynchronous)
1016-- Reply will come in the form of a 'WhereIsReply' message.
1018-- There is currently no synchronous version of 'whereisRemoteAsync': if
1019-- you implement one yourself, be sure to take into account that the remote
1020-- node might die or get disconnect before it can respond (i.e. you should
1021-- use 'monitorNode' and take appropriate action when you receive a
1022-- 'NodeMonitorNotification').
1023whereisRemoteAsync :: NodeId -> String -> Process ()
1024whereisRemoteAsync nid label =
1025 sendCtrlMsg (Just nid) (WhereIs label)
1027-- | Named send to a process in the local registry (asynchronous)
1028nsend :: Serializable a => String -> a -> Process ()
1030 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1032-- | Named send to a process in the local registry (asynchronous).
1033-- This function makes /no/ attempt to serialize and (in the case when the
1034-- destination process resides on the same local node) therefore ensure that
1035-- the payload is fully evaluated before it is delivered.
1036unsafeNSend :: Serializable a => String -> a -> Process ()
1037unsafeNSend = Unsafe.nsend
1039-- | Named send to a process in a remote registry (asynchronous)
1040nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1041nsendRemote nid label msg =
1042 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1044--------------------------------------------------------------------------------
1046--------------------------------------------------------------------------------
1048-- | Resolve a static value
1049unStatic :: Typeable a => Static a -> Process a
1051 rtable <- remoteTable . processNode <$> ask
1052 case Static.unstatic rtable static of
1053 Left err -> fail $ "Could not resolve static value: " ++ err
1056-- | Resolve a closure
1057unClosure :: Typeable a => Closure a -> Process a
1058unClosure closure = do
1059 rtable <- remoteTable . processNode <$> ask
1060 case Static.unclosure rtable closure of
1061 Left err -> fail $ "Could not resolve closure: " ++ err
1064--------------------------------------------------------------------------------
1066--------------------------------------------------------------------------------
1068-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1069-- message passing. However, when network connections get disrupted this
1070-- illusion cannot always be maintained. Once a network connection breaks (even
1071-- temporarily) no further communication on that connection will be possible.
1072-- For example, if process A sends a message to process B, and A is then
1073-- notified (by monitor notification) that it got disconnected from B, A will
1074-- not be able to send any further messages to B, /unless/ A explicitly
1075-- indicates that it is acceptable to attempt to reconnect to B using the
1076-- Cloud Haskell 'reconnect' primitive.
1078-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1079-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1080-- then receives a monitor notification that its connection to B has been lost,
1081-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1082-- and m3 but not m2.
1084-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1085-- to attempt to reconnect on the next send/. In particular, if no further
1086-- communication attempts are made to B then A can use reconnect to clean up
1087-- its connection to B.
1088reconnect :: ProcessId -> Process ()
1091 node <- processNode <$> ask
1092 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1094-- | Reconnect to a sendport. See 'reconnect' for more information.
1095reconnectPort :: SendPort a -> Process ()
1096reconnectPort them = do
1098 node <- processNode <$> ask
1099 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1101--------------------------------------------------------------------------------
1102-- Auxiliary functions --
1103--------------------------------------------------------------------------------
1105sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1107 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1109sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1110sendChanLocal spId msg =
1111 -- we *must* fully serialize/encode the message here, because
1112 -- attempting to use `unsafeCoerce' in the node controller
1113 -- won't work since we know nothing about the required type
1114 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1116getMonitorRefFor :: Identifier -> Process MonitorRef
1117getMonitorRefFor ident = do
1119 liftIO $ modifyMVar (processState proc) $ \st -> do
1120 let counter = st ^. monitorCounter
1121 return ( monitorCounter ^: (+ 1) $ st
1122 , MonitorRef ident counter
1125getSpawnRef :: Process SpawnRef
1128 liftIO $ modifyMVar (processState proc) $ \st -> do
1129 let counter = st ^. spawnCounter
1130 return ( spawnCounter ^: (+ 1) $ st
1134-- | Monitor a process/node/channel
1135monitor' :: Identifier -> Process MonitorRef
1137 monitorRef <- getMonitorRefFor ident
1138 sendCtrlMsg Nothing $ Monitor monitorRef
1141-- | Link to a process/node/channel
1142link' :: Identifier -> Process ()
1143link' = 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