2{-# LANGUAGE RankNTypes #-}
3{-# LANGUAGE DeriveDataTypeable #-}
4{-# LANGUAGE ScopedTypeVariables #-}
5{-# LANGUAGE ExistentialQuantification #-}
6{-# LANGUAGE BangPatterns #-}
8-- | Cloud Haskell primitives
10-- We define these in a separate module so that we don't have to rely on
11-- the closure combinators
12module Control.Distributed.Process.Internal.Primitives
13 ( -- * Basic messaging
22 -- * Unsafe messaging variants
26 -- * Advanced messaging
51 -- * Process management
53 , ProcessTerminationException(..)
59 -- keep the exception constructor hidden, so that handling exit
60 -- reasons /must/ take place via the 'catchExit' family of primitives
61 , ProcessExitException()
69 -- * Monitoring and linking
84 , reregisterRemoteAsync
85 , unregisterRemoteAsync
91 -- * Exception handling
114 -- * Internal Exports
118#if ! MIN_VERSION_base(4,6,0)
119import Prelude hiding (catch)
122import Data.Binary (decode)
123import Data.Time.Clock (getCurrentTime)
124import Data.Time.Format (formatTime)
125import System.Locale (defaultTimeLocale)
126import System.Timeout (timeout)
127import Control.Monad (when)
128import Control.Monad.Reader (ask)
129import Control.Monad.IO.Class (MonadIO, liftIO)
130import Control.Applicative ((<$>))
131import Control.Exception (Exception(..), throw, throwIO, SomeException)
132import qualified Control.Exception as Ex (catch, mask, try)
133import Control.Distributed.Process.Internal.StrictMVar
138import Control.Concurrent.STM
147import Control.Distributed.Process.Internal.CQueue
152import Control.Distributed.Process.Serializable (Serializable, fingerprint)
153import Data.Accessor ((^.), (^:), (^=))
154import Control.Distributed.Static
158import Data.Rank1Typeable (Typeable)
159import qualified Control.Distributed.Static as Static (unstatic, unclosure)
160import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
161import Control.Distributed.Process.Internal.Types
171 , NodeMonitorNotification(..)
181 , ProcessExitException(..)
184 , DidUnlinkProcess(..)
189 , ProcessRegistrationException(..)
191 , ProcessInfoNone(..)
195 , createUnencodedMessage
197 , ImplicitReconnect( NoImplicitReconnect)
202import Control.Distributed.Process.Internal.Messaging
209import Control.Distributed.Process.Management.Types
212import Control.Distributed.Process.Management.Trace.Types
215import Control.Distributed.Process.Internal.WeakTQueue
223--------------------------------------------------------------------------------
225--------------------------------------------------------------------------------
228send :: Serializable a => ProcessId -> a -> Process ()
229-- This requires a lookup on every send. If we want to avoid that we need to
230-- modify serializable to allow for stateful (IO) deserialization.
233 let us = processId proc
234 node = processNode proc
235 nodeId = localNodeId node
236 destNode = (processNodeId them) in do
237 case destNode == nodeId of
238 True -> sendLocal them msg
239 False -> liftIO $ sendMessage (processNode proc)
240 (ProcessIdentifier (processId proc))
241 (ProcessIdentifier them)
244 -- We do not fire the trace event until after the sending is complete;
245 -- In the remote case, 'sendMessage' can block in the networking stack.
246 liftIO $ traceEvent (localEventBus node)
247 (MxSent them us (createUnencodedMessage msg))
249-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
250-- and (in the case when the destination process resides on the same local
251-- node) therefore ensure that the payload is fully evaluated before it is
253unsafeSend :: Serializable a => ProcessId -> a -> Process ()
254unsafeSend = Unsafe.send
256-- | Wait for a message of a specific type
257expect :: forall a. Serializable a => Process a
258expect = receiveWait [match return]
260--------------------------------------------------------------------------------
262--------------------------------------------------------------------------------
264-- | Create a new typed channel
265newChan :: Serializable a => Process (SendPort a, ReceivePort a)
268 liftIO . modifyMVar (processState proc) $ \st -> do
269 let lcid = st ^. channelCounter
270 let cid = SendPortId { sendPortProcessId = processId proc
271 , sendPortLocalId = lcid
273 let sport = SendPort cid
274 chan <- liftIO newTQueueIO
275 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
276 let rport = ReceivePort $ readTQueue chan
277 let tch = TypedChannel chan'
278 return ( (channelCounter ^: (+ 1))
279 . (typedChannelWithId lcid ^= Just tch)
284 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
285 finalizer st lcid = modifyMVar_ st $
286 return . (typedChannelWithId lcid ^= Nothing)
288-- | Send a message on a typed channel
289sendChan :: Serializable a => SendPort a -> a -> Process ()
290sendChan (SendPort cid) msg = do
292 let node = localNodeId (processNode proc)
293 destNode = processNodeId (sendPortProcessId cid) in do
294 case destNode == node of
295 True -> sendChanLocal cid msg
297 liftIO $ sendBinary (processNode proc)
298 (ProcessIdentifier (processId proc))
299 (SendPortIdentifier cid)
303-- | Send a message on a typed channel. This function makes /no/ attempt to
304-- serialize and (in the case when the @ReceivePort@ resides on the same local
305-- node) therefore ensure that the payload is fully evaluated before it is
307unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
308unsafeSendChan = Unsafe.sendChan
310-- | Wait for a message on a typed channel
311receiveChan :: Serializable a => ReceivePort a -> Process a
312receiveChan = liftIO . atomically . receiveSTM
314-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
315-- non-blocking check for a message.
316receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
317receiveChanTimeout 0 ch = liftIO . atomically $
318 (Just <$> receiveSTM ch) `orElse` return Nothing
319receiveChanTimeout n ch = liftIO . timeout n . atomically $
322-- | Merge a list of typed channels.
324-- The result port is left-biased: if there are messages available on more
325-- than one port, the first available message is returned.
326mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
327mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
329-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
331mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
332mergePortsRR = \ps -> do
333 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
334 return $ ReceivePort (rr psVar)
338 rotate (x:xs) = xs ++ [x]
340 rr :: TVar [STM a] -> STM a
343 a <- foldr1 orElse ps
344 writeTVar psVar (rotate ps)
347--------------------------------------------------------------------------------
348-- Advanced messaging --
349--------------------------------------------------------------------------------
351-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
352newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
354-- | Test the matches in order against each message in the queue
355receiveWait :: [Match b] -> Process b
357 queue <- processQueue <$> ask
358 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
361-- | Like 'receiveWait' but with a timeout.
363-- If the timeout is zero do a non-blocking check for matching messages. A
364-- non-zero timeout is applied only when waiting for incoming messages (that is,
365-- /after/ we have checked the messages that are already in the mailbox).
366receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
367receiveTimeout t ms = do
368 queue <- processQueue <$> ask
369 let blockSpec = if t == 0 then NonBlocking else Timeout t
370 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
372 Nothing -> return Nothing
373 Just proc -> Just <$> proc
375-- | Match on a typed channel
376matchChan :: ReceivePort a -> (a -> Process b) -> Match b
377matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
379-- | Match on an arbitrary STM action.
381-- This rather unusaul /match primitive/ allows us to compose arbitrary STM
382-- actions with checks against our process' mailbox and/or any typed channel
383-- @ReceivePort@s we may hold.
385-- This allows us to process multiple input streams /along with our mailbox/,
386-- in just the same way that 'matchChan' supports checking both the mailbox
387-- /and/ an arbitrary set of typed channels in one atomic transaction.
389-- Note there are no ordering guarnatees with respect to these disparate input
392matchSTM :: STM a -> (a -> Process b) -> Match b
393matchSTM stm fn = Match $ MatchChan (fmap fn stm)
395-- | Match against any message of the right type
396match :: forall a b. Serializable a => (a -> Process b) -> Match b
397match = matchIf (const True)
399-- | Match against any message of the right type that satisfies a predicate
400matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
401matchIf c p = Match $ MatchMsg $ \msg ->
402 case messageFingerprint msg == fingerprint (undefined :: a) of
405 (UnencodedMessage _ m) ->
406 let m' = unsafeCoerce m :: a in
410 (EncodedMessage _ _) ->
411 if (c decoded) then Just (p decoded) else Nothing
414 -- Make sure the value is fully decoded so that we don't hang to
415 -- bytestrings when the process calling 'matchIf' doesn't process
416 -- the values immediately
417 !decoded = decode (messageEncoding msg)
419-- | Match against any message, regardless of the underlying (contained) type
420matchMessage :: (Message -> Process Message) -> Match Message
421matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
423-- | Match against any message (regardless of underlying type) that satisfies a predicate
424matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
425matchMessageIf c p = Match $ MatchMsg $ \msg ->
430-- | Forward a raw 'Message' to the given 'ProcessId'.
431forward :: Message -> ProcessId -> Process ()
434 let node = processNode proc
436 nid = localNodeId node
437 destNode = (processNodeId them) in do
438 case destNode == nid of
439 True -> sendCtrlMsg Nothing (LocalSend them msg)
440 False -> liftIO $ sendPayload (processNode proc)
441 (ProcessIdentifier (processId proc))
442 (ProcessIdentifier them)
444 (messageToPayload msg)
445 -- We do not fire the trace event until after the sending is complete;
446 -- In the remote case, 'sendMessage' can block in the networking stack.
447 liftIO $ traceEvent (localEventBus node)
451-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
452-- 'Serializable' - like the datum they contain - but also note, deserialising
453-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
454-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
459-- send self (wrapMessage "blah")
460-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
461-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
462-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
465wrapMessage :: Serializable a => a -> Message
466wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
468-- | This is the /unsafe/ variant of 'wrapMessage'. See
469-- "Control.Distributed.Process.UnsafePrimitives" for details.
470unsafeWrapMessage :: Serializable a => a -> Message
471unsafeWrapMessage = Unsafe.wrapMessage
473-- [note Serializable UnencodedMessage]
474-- if we attempt to serialise an UnencodedMessage, it will be converted to an
475-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
476-- Primitives.hs for the gory details
478-- | Attempt to unwrap a raw 'Message'.
479-- If the type of the decoded message payload matches the expected type, the
480-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
483-- This expression, for example, will evaluate to @Nothing@
484-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
486-- Whereas this expression, will yield @Just "foo"@
487-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
489unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
491 case messageFingerprint msg == fingerprint (undefined :: a) of
492 False -> return Nothing :: m (Maybe a)
494 (UnencodedMessage _ ms) ->
495 let ms' = unsafeCoerce ms :: a
497 (EncodedMessage _ _) ->
498 return (Just (decoded))
500 decoded :: a -- note [decoding]
501 !decoded = decode (messageEncoding msg)
503-- | Attempt to handle a raw 'Message'.
504-- If the type of the message matches the type of the first argument to
505-- the supplied expression, then the message will be decoded and the expression
506-- evaluated against its value. If this runtime type checking fails however,
507-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
508-- evaluation proceeds, the resulting value with be wrapped with @Just@.
510-- Intended for use in `catchesExit` and `matchAny` primitives.
512handleMessage :: forall m a b. (Monad m, Serializable a)
513 => Message -> (a -> m b) -> m (Maybe b)
514handleMessage msg proc = handleMessageIf msg (const True) proc
516-- | Conditionally handle a raw 'Message'.
517-- If the predicate @(a -> Bool)@ evaluates to @True@, invokes the supplied
518-- handler, other returns @Nothing@ to indicate failure. See 'handleMessage'
519-- for further information about runtime type checking.
521handleMessageIf :: forall m a b . (Monad m, Serializable a)
526handleMessageIf msg c proc = do
527 case messageFingerprint msg == fingerprint (undefined :: a) of
528 False -> return Nothing :: m (Maybe b)
530 (UnencodedMessage _ ms) ->
531 let ms' = unsafeCoerce ms :: a in
533 True -> do { r <- proc ms'; return (Just r) }
534 False -> return Nothing :: m (Maybe b)
535 (EncodedMessage _ _) ->
537 True -> do { r <- proc (decoded :: a); return (Just r) }
538 False -> return Nothing :: m (Maybe b)
540 decoded :: a -- note [decoding]
541 !decoded = decode (messageEncoding msg)
543-- | As 'handleMessage' but ignores result, which is useful if you don't
544-- care whether or not the handler succeeded.
546handleMessage_ :: forall m a . (Monad m, Serializable a)
547 => Message -> (a -> m ()) -> m ()
548handleMessage_ msg proc = handleMessageIf_ msg (const True) proc
550-- | Conditional version of 'handleMessage_'.
552handleMessageIf_ :: forall m a . (Monad m, Serializable a)
557handleMessageIf_ msg c proc = handleMessageIf msg c proc >> return ()
559-- | Match against an arbitrary message. 'matchAny' removes the first available
560-- message from the process mailbox. To handle arbitrary /raw/ messages once
561-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
563matchAny :: forall b. (Message -> Process b) -> Match b
564matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
566-- | Match against an arbitrary message. Intended for use with 'handleMessage'
567-- and 'unwrapMessage', this function /only/ removes a message from the process
568-- mailbox, /if/ the supplied condition matches. The success (or failure) of
569-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
570-- here, i.e., if the condition evaluates to @True@ then the message will
571-- be removed from the process mailbox and decoded, but that does /not/
572-- guarantee that an expression passed to @handleMessage@ will pass the
573-- runtime type checks and therefore be evaluated.
575matchAnyIf :: forall a b. (Serializable a)
577 -> (Message -> Process b)
579matchAnyIf c p = Match $ MatchMsg $ \msg ->
580 case messageFingerprint msg == fingerprint (undefined :: a) of
581 True | check -> Just (p msg)
586 (EncodedMessage _ _) -> c decoded
587 (UnencodedMessage _ m') -> c (unsafeCoerce m')
589 decoded :: a -- note [decoding]
590 !decoded = decode (messageEncoding msg)
594For an EncodedMessage, we need to ensure the value is fully decoded so that
595we don't hang to bytestrings if the calling process doesn't evaluate
596immediately. For UnencodedMessage we know (because the fingerprint comparison
597succeeds) that unsafeCoerce will not fail.
600-- | Remove any message from the queue
601matchUnknown :: Process b -> Match b
602matchUnknown p = Match $ MatchMsg (const (Just p))
604-- | Receives messages and forwards them to @pid@ if @p msg == True@.
605delegate :: ProcessId -> (Message -> Bool) -> Process ()
608 matchAny (\m -> case (p m) of
609 True -> forward m pid
614-- | A straight relay that forwards all messages to the supplied pid.
615relay :: ProcessId -> Process ()
616relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
618-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
619-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
620-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
621-- fails, no action is taken and the proxy process will continue running.
622proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
626 next <- handleMessage m proc
628 Just True -> forward m pid
629 Just False -> return () -- explicitly ignored
630 Nothing -> return ()) -- un-routable
634--------------------------------------------------------------------------------
635-- Process management --
636--------------------------------------------------------------------------------
638-- | Thrown by 'terminate'
639data ProcessTerminationException = ProcessTerminationException
640 deriving (Show, Typeable)
642instance Exception ProcessTerminationException
644-- | Terminate immediately (throws a ProcessTerminationException)
645terminate :: Process a
646terminate = liftIO $ throwIO ProcessTerminationException
649-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
650die :: Serializable a => a -> Process b
652-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
653-- could be decoded by a handler passed to 'catchExit', re-thrown or even
654-- passed to another process via the tracing mechanism.
656 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
658-- | Forceful request to kill a process. Where 'exit' provides an exception
659-- that can be caught and handled, 'kill' throws an unexposed exception type
660-- which cannot be handled explicitly (by type).
661kill :: ProcessId -> String -> Process ()
662-- NOTE: We send the message to our local node controller, which will then
663-- forward it to a remote node controller (if applicable). Sending it directly
664-- to a remote node controller means that that the message may overtake a
665-- 'monitor' or 'link' request.
666kill them reason = sendCtrlMsg Nothing (Kill them reason)
668-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
669-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
670-- manner can be handled using the 'catchExit' family of functions.
671exit :: Serializable a => ProcessId -> a -> Process ()
672-- NOTE: We send the message to our local node controller, which will then
673-- forward it to a remote node controller (if applicable). Sending it directly
674-- to a remote node controller means that that the message may overtake a
675-- 'monitor' or 'link' request.
676exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
678-- | Catches 'ProcessExitException'. The handler will not be applied unless its
679-- type matches the encoded data stored in the exception (see the /reason/
680-- argument given to the 'exit' primitive). If the handler cannot be applied,
681-- the exception will be re-thrown.
683-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
684-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
685catchExit :: forall a b . (Show a, Serializable a)
687 -> (ProcessId -> a -> Process b)
689catchExit act exitHandler = catch act handleExit
691 handleExit ex@(ProcessExitException from msg) =
692 if messageFingerprint msg == fingerprint (undefined :: a)
693 then exitHandler from decoded
694 else liftIO $ throwIO ex
697 -- Make sure the value is fully decoded so that we don't hang to
698 -- bytestrings if the caller doesn't use the value immediately
699 !decoded = decode (messageEncoding msg)
701-- | Lift 'Control.Exception.catches' (almost).
703-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
704-- message, a handler must accept inputs of the expected type. In order to
705-- handle a list of potentially different handlers (and therefore input types),
706-- a handler passed to 'catchesExit' must accept 'Message' and return
707-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
709-- See 'maybeHandleMessage' and 'Message' for more details.
710catchesExit :: Process b
711 -> [(ProcessId -> Message -> (Process (Maybe b)))]
713catchesExit act handlers = catch act ((flip handleExit) handlers)
715 handleExit :: ProcessExitException
716 -> [(ProcessId -> Message -> Process (Maybe b))]
718 handleExit ex [] = liftIO $ throwIO ex
719 handleExit ex@(ProcessExitException from msg) (h:hs) = do
722 Nothing -> handleExit ex hs
725-- | Our own process ID
726getSelfPid :: Process ProcessId
727getSelfPid = processId <$> ask
729-- | Get the node ID of our local node
730getSelfNode :: Process NodeId
731getSelfNode = localNodeId . processNode <$> ask
734-- | Get statistics about the specified node
735getNodeStats :: NodeId -> Process (Either DiedReason NodeStats)
737 selfNode <- getSelfNode
739 then Right `fmap` getLocalNodeStats -- optimisation
740 else getNodeStatsRemote nid
742 getNodeStatsRemote :: NodeId -> Process (Either DiedReason NodeStats)
743 getNodeStatsRemote nid = do
744 sendCtrlMsg (Just nid) $ GetNodeStats nid
745 bracket (monitorNode nid) unmonitor $ \mRef ->
746 receiveWait [ match (\(stats :: NodeStats) -> return $ Right stats)
747 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef)
748 (\(NodeMonitorNotification _ _ dr) -> return $ Left dr)
751-- | Get statistics about our local node
752getLocalNodeStats :: Process NodeStats
753getLocalNodeStats = do
755 sendCtrlMsg Nothing $ GetNodeStats self
756 receiveWait [ match (\(stats :: NodeStats) -> return stats) ]
758-- | Get information about the specified process
759getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
761 let them = processNodeId pid in do
763 dest <- mkNode them us
764 sendCtrlMsg dest $ GetInfo pid
766 match (\(p :: ProcessInfo) -> return $ Just p)
767 , match (\(_ :: ProcessInfoNone) -> return Nothing)
769 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
770 mkNode them us = case them == us of
771 True -> return Nothing
772 _ -> return $ Just them
774--------------------------------------------------------------------------------
775-- Monitoring and linking --
776--------------------------------------------------------------------------------
778-- | Link to a remote process (asynchronous)
780-- When process A links to process B (that is, process A calls
781-- @link pidB@) then an asynchronous exception will be thrown to process A
782-- when process B terminates (normally or abnormally), or when process A gets
783-- disconnected from process B. Although it is /technically/ possible to catch
784-- these exceptions, chances are if you find yourself trying to do so you should
785-- probably be using 'monitor' rather than 'link'. In particular, code such as
787-- > link pidB -- Link to process B
788-- > expect -- Wait for a message from process B
789-- > unlink pidB -- Unlink again
791-- doesn't quite do what one might expect: if process B sends a message to
792-- process A, and /subsequently terminates/, then process A might or might not
793-- be terminated too, depending on whether the exception is thrown before or
794-- after the 'unlink' (i.e., this code has a race condition).
796-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
797-- call to 'link' has no effect.
799-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
800-- Linking makes no distinction between normal and abnormal termination of
801-- the remote process.
802link :: ProcessId -> Process ()
803link = sendCtrlMsg Nothing . Link . ProcessIdentifier
805-- | Monitor another process (asynchronous)
807-- When process A monitors process B (that is, process A calls
808-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
809-- when process B terminates (normally or abnormally), or when process A gets
810-- disconnected from process B. You receive this message like any other (using
811-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
812-- 'DiedDisconnect', etc.).
814-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
815-- multiple monitors are set up, multiple notifications will be delivered
816-- and monitors can be disabled individually using 'unmonitor'.
817monitor :: ProcessId -> Process MonitorRef
818monitor = monitor' . ProcessIdentifier
820-- | Establishes temporary monitoring of another process.
822-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
823-- of @code@. Note: although monitoring is no longer active when
824-- @withMonitor@ returns, there might still be unreceived monitor
825-- messages in the queue.
827withMonitor :: ProcessId -> Process a -> Process a
828withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
829 -- unmonitor blocks waiting for the response, so there's a possibility
830 -- that an exception might interrupt withMonitor before the unmonitor
831 -- has completed. I think that's better than making the unmonitor
836-- This is synchronous in the sense that once it returns you are guaranteed
837-- that no exception will be raised if the remote process dies. However, it is
838-- asynchronous in the sense that we do not wait for a response from the remote
840unlink :: ProcessId -> Process ()
843 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
847-- | Remove a node link
849-- This has the same synchronous/asynchronous nature as 'unlink'.
850unlinkNode :: NodeId -> Process ()
853 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
857-- | Remove a channel (send port) link
859-- This has the same synchronous/asynchronous nature as 'unlink'.
860unlinkPort :: SendPort a -> Process ()
862 unlinkPortAsync sport
863 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
869-- This has the same synchronous/asynchronous nature as 'unlink'.
870unmonitor :: MonitorRef -> Process ()
873 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
877--------------------------------------------------------------------------------
878-- Exception handling --
879--------------------------------------------------------------------------------
881-- | Lift 'Control.Exception.catch'
882catch :: Exception e => Process a -> (e -> Process a) -> Process a
885 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
887-- | Lift 'Control.Exception.try'
888try :: Exception e => Process a -> Process (Either e a)
891 liftIO $ Ex.try (runLocalProcess lproc p)
893-- | Lift 'Control.Exception.mask'
894mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
897 liftIO $ Ex.mask $ \restore ->
898 runLocalProcess lproc (p (liftRestore restore))
900 liftRestore :: (forall a. IO a -> IO a)
901 -> (forall a. Process a -> Process a)
902 liftRestore restoreIO = \p2 -> do
904 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
906-- | Lift 'Control.Exception.onException'
907onException :: Process a -> Process b -> Process a
908onException p what = p `catch` \e -> do _ <- what
909 liftIO $ throwIO (e :: SomeException)
911-- | Lift 'Control.Exception.bracket'
912bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
913bracket before after thing =
914 mask $ \restore -> do
916 r <- restore (thing a) `onException` after a
920-- | Lift 'Control.Exception.bracket_'
921bracket_ :: Process a -> Process b -> Process c -> Process c
922bracket_ before after thing = bracket before (const after) (const thing)
924-- | Lift 'Control.Exception.finally'
925finally :: Process a -> Process b -> Process a
926finally a sequel = bracket_ (return ()) sequel a
928-- | You need this when using 'catches'
929data Handler a = forall e . Exception e => Handler (e -> Process a)
931instance Functor Handler where
932 fmap f (Handler h) = Handler (fmap f . h)
934-- | Lift 'Control.Exception.catches'
935catches :: Process a -> [Handler a] -> Process a
936catches proc handlers = proc `catch` catchesHandler handlers
938catchesHandler :: [Handler a] -> SomeException -> Process a
939catchesHandler handlers e = foldr tryHandler (throw e) handlers
940 where tryHandler (Handler handler) res
941 = case fromException e of
942 Just e' -> handler e'
945--------------------------------------------------------------------------------
947--------------------------------------------------------------------------------
949-- | Like 'expect' but with a timeout
950expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
951expectTimeout n = receiveTimeout n [match return]
953-- | Asynchronous version of 'spawn'
955-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
956spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
957spawnAsync nid proc = do
958 spawnRef <- getSpawnRef
959 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
962-- | Monitor a node (asynchronous)
963monitorNode :: NodeId -> Process MonitorRef
965 monitor' . NodeIdentifier
967-- | Monitor a typed channel (asynchronous)
968monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
969monitorPort (SendPort cid) =
970 monitor' (SendPortIdentifier cid)
972-- | Remove a monitor (asynchronous)
973unmonitorAsync :: MonitorRef -> Process ()
975 sendCtrlMsg Nothing . Unmonitor
977-- | Link to a node (asynchronous)
978linkNode :: NodeId -> Process ()
979linkNode = link' . NodeIdentifier
981-- | Link to a channel (asynchronous)
982linkPort :: SendPort a -> Process ()
983linkPort (SendPort cid) =
984 link' (SendPortIdentifier cid)
986-- | Remove a process link (asynchronous)
987unlinkAsync :: ProcessId -> Process ()
989 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
991-- | Remove a node link (asynchronous)
992unlinkNodeAsync :: NodeId -> Process ()
994 sendCtrlMsg Nothing . Unlink . NodeIdentifier
996-- | Remove a channel (send port) link (asynchronous)
997unlinkPortAsync :: SendPort a -> Process ()
998unlinkPortAsync (SendPort cid) =
999 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
1001--------------------------------------------------------------------------------
1003--------------------------------------------------------------------------------
1007-- @say message@ sends a message (time, pid of the current process, message)
1008-- to the process registered as 'logger'. By default, this process simply
1009-- sends the string to 'stderr'. Individual Cloud Haskell backends might
1010-- replace this with a different logger process, however.
1011say :: String -> Process ()
1013 now <- liftIO getCurrentTime
1015 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
1017--------------------------------------------------------------------------------
1019--------------------------------------------------------------------------------
1021-- | Register a process with the local registry (asynchronous).
1022-- This version will wait until a response is gotten from the
1023-- management process. The name must not already be registered.
1024-- The process need not be on this node.
1025-- A bad registration will result in a 'ProcessRegistrationException'
1027-- The process to be registered does not have to be local itself.
1028register :: String -> ProcessId -> Process ()
1029register = registerImpl False
1031-- | Like 'register', but will replace an existing registration.
1032-- The name must already be registered.
1033reregister :: String -> ProcessId -> Process ()
1034reregister = registerImpl True
1036registerImpl :: Bool -> String -> ProcessId -> Process ()
1037registerImpl force label pid = do
1038 mynid <- getSelfNode
1039 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
1040 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
1041 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
1044-- | Register a process with a remote registry (asynchronous).
1046-- The process to be registered does not have to live on the same remote node.
1047-- Reply wil come in the form of a 'RegisterReply' message
1049-- See comments in 'whereisRemoteAsync'
1050registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1051registerRemoteAsync nid label pid =
1052 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
1054reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
1055reregisterRemoteAsync nid label pid =
1056 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
1058-- | Remove a process from the local registry (asynchronous).
1059-- This version will wait until a response is gotten from the
1060-- management process. The name must already be registered.
1061unregister :: String -> Process ()
1062unregister label = do
1063 mynid <- getSelfNode
1064 sendCtrlMsg Nothing (Register label mynid Nothing False)
1065 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
1066 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
1069-- | Deal with the result from an attempted registration or unregistration
1070-- by throwing an exception if necessary
1071handleRegistrationReply :: String -> Bool -> Process ()
1072handleRegistrationReply label ok =
1074 liftIO $ throwIO $ ProcessRegistrationException label
1076-- | Remove a process from a remote registry (asynchronous).
1078-- Reply wil come in the form of a 'RegisterReply' message
1080-- See comments in 'whereisRemoteAsync'
1081unregisterRemoteAsync :: NodeId -> String -> Process ()
1082unregisterRemoteAsync nid label =
1083 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1085-- | Query the local process registry
1086whereis :: String -> Process (Maybe ProcessId)
1088 sendCtrlMsg Nothing (WhereIs label)
1089 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1090 (\(WhereIsReply _ mPid) -> return mPid)
1093-- | Query a remote process registry (asynchronous)
1095-- Reply will come in the form of a 'WhereIsReply' message.
1097-- There is currently no synchronous version of 'whereisRemoteAsync': if
1098-- you implement one yourself, be sure to take into account that the remote
1099-- node might die or get disconnect before it can respond (i.e. you should
1100-- use 'monitorNode' and take appropriate action when you receive a
1101-- 'NodeMonitorNotification').
1102whereisRemoteAsync :: NodeId -> String -> Process ()
1103whereisRemoteAsync nid label =
1104 sendCtrlMsg (Just nid) (WhereIs label)
1106-- | Named send to a process in the local registry (asynchronous)
1107nsend :: Serializable a => String -> a -> Process ()
1109 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1111-- | Named send to a process in the local registry (asynchronous).
1112-- This function makes /no/ attempt to serialize and (in the case when the
1113-- destination process resides on the same local node) therefore ensure that
1114-- the payload is fully evaluated before it is delivered.
1115unsafeNSend :: Serializable a => String -> a -> Process ()
1116unsafeNSend = Unsafe.nsend
1118-- | Named send to a process in a remote registry (asynchronous)
1119nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1120nsendRemote nid label msg =
1121 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1123--------------------------------------------------------------------------------
1125--------------------------------------------------------------------------------
1127-- | Resolve a static value
1128unStatic :: Typeable a => Static a -> Process a
1130 rtable <- remoteTable . processNode <$> ask
1131 case Static.unstatic rtable static of
1132 Left err -> fail $ "Could not resolve static value: " ++ err
1135-- | Resolve a closure
1136unClosure :: Typeable a => Closure a -> Process a
1137unClosure closure = do
1138 rtable <- remoteTable . processNode <$> ask
1139 case Static.unclosure rtable closure of
1140 Left err -> fail $ "Could not resolve closure: " ++ err
1143--------------------------------------------------------------------------------
1145--------------------------------------------------------------------------------
1147-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1148-- message passing. However, when network connections get disrupted this
1149-- illusion cannot always be maintained. Once a network connection breaks (even
1150-- temporarily) no further communication on that connection will be possible.
1151-- For example, if process A sends a message to process B, and A is then
1152-- notified (by monitor notification) that it got disconnected from B, A will
1153-- not be able to send any further messages to B, /unless/ A explicitly
1154-- indicates that it is acceptable to attempt to reconnect to B using the
1155-- Cloud Haskell 'reconnect' primitive.
1157-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1158-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1159-- then receives a monitor notification that its connection to B has been lost,
1160-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1161-- and m3 but not m2.
1163-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1164-- to attempt to reconnect on the next send/. In particular, if no further
1165-- communication attempts are made to B then A can use reconnect to clean up
1166-- its connection to B.
1167reconnect :: ProcessId -> Process ()
1170 node <- processNode <$> ask
1171 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1173-- | Reconnect to a sendport. See 'reconnect' for more information.
1174reconnectPort :: SendPort a -> Process ()
1175reconnectPort them = do
1177 node <- processNode <$> ask
1178 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1180--------------------------------------------------------------------------------
1181-- Auxiliary functions --
1182--------------------------------------------------------------------------------
1184sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1186 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1188sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1189sendChanLocal spId msg =
1190 -- we *must* fully serialize/encode the message here, because
1191 -- attempting to use `unsafeCoerce' in the node controller
1192 -- won't work since we know nothing about the required type
1193 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1195getMonitorRefFor :: Identifier -> Process MonitorRef
1196getMonitorRefFor ident = do
1198 liftIO $ modifyMVar (processState proc) $ \st -> do
1199 let counter = st ^. monitorCounter
1200 return ( monitorCounter ^: (+ 1) $ st
1201 , MonitorRef ident counter
1204getSpawnRef :: Process SpawnRef
1207 liftIO $ modifyMVar (processState proc) $ \st -> do
1208 let counter = st ^. spawnCounter
1209 return ( spawnCounter ^: (+ 1) $ st
1213-- | Monitor a process/node/channel
1214monitor' :: Identifier -> Process MonitorRef
1216 monitorRef <- getMonitorRefFor ident
1217 sendCtrlMsg Nothing $ Monitor monitorRef
1220-- | Link to a process/node/channel
1221link' :: Identifier -> Process ()
1222link' = 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