1-- | Cloud Haskell primitives
3-- We define these in a separate module so that we don't have to rely on
4-- the closure combinators
5module Control.Distributed.Process.Internal.Primitives
15 -- * Advanced messaging
31 -- * Process management
33 , ProcessTerminationException(..)
39 -- keep the exception constructor hidden, so that handling exit
40 -- reasons /must/ take place via the 'catchExit' family of primitives
41 , ProcessExitException()
47 -- * Monitoring and linking
62 , reregisterRemoteAsync
63 , unregisterRemoteAsync
69 -- * Exception handling
92 -- * Tracing/Debugging
96#if ! MIN_VERSION_base(4,6,0)
97import Prelude hiding (catch)
100import Control.Monad (when)
101import Control.Monad.Reader (ask)
102import Control.Monad.IO.Class (MonadIO, liftIO)
103import Control.Applicative ((<$>))
104import Control.Exception (Exception(..), throw, throwIO, SomeException)
105import qualified Control.Exception as Ex (catch, mask, try)
106import Control.Distributed.Process.Internal.StrictMVar
112import Control.Concurrent.Chan (writeChan)
113import Control.Concurrent.STM
122import Control.Distributed.Process.Internal.CQueue
128import Control.Distributed.Process.Serializable (Serializable, fingerprint)
129import Control.Distributed.Static (Closure, Static)
130import qualified Control.Distributed.Static as Static (unstatic, unclosure)
131import Control.Distributed.Process.Internal.Types
152 , ProcessExitException(..)
155 , DidUnlinkProcess(..)
160 , ProcessRegistrationException(..)
162 , ProcessInfoNone(..)
164 , createUnencodedMessage
166 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
167 , LocalProcessState(..)
171import Control.Distributed.Process.Internal.Messaging
177import qualified Control.Distributed.Process.Internal.Trace as Trace
178import Control.Distributed.Process.Internal.WeakTQueue
185import Data.Accessor ((^.), (^:), (^=))
186import Data.Binary (decode)
187import Data.Foldable (forM_)
188import Data.Rank1Typeable (Typeable)
189import Data.Time.Clock (getCurrentTime)
190import Data.Time.Format (formatTime)
191import System.Locale (defaultTimeLocale)
192import System.Mem.Weak (deRefWeak)
193import System.Timeout (timeout)
196--------------------------------------------------------------------------------
198--------------------------------------------------------------------------------
201send :: Serializable a => ProcessId -> a -> Process ()
202-- This requires a lookup on every send. If we want to avoid that we need to
203-- modify serializable to allow for stateful (IO) deserialization
206 let node = localNodeId (processNode proc)
207 destNode = (processNodeId them) in do
208 case destNode == node of
209 True -> sendLocal them msg
210 False -> liftIO $ sendMessage (processNode proc)
211 (ProcessIdentifier (processId proc))
212 (ProcessIdentifier them)
216-- | Wait for a message of a specific type
217expect :: forall a. Serializable a => Process a
218expect = receiveWait [match return]
220--------------------------------------------------------------------------------
222--------------------------------------------------------------------------------
224-- | Create a new typed channel
225newChan :: Serializable a => Process (SendPort a, ReceivePort a)
228 liftIO . modifyMVar (processState proc) $ \st -> do
229 let lcid = st ^. channelCounter
230 let cid = SendPortId { sendPortProcessId = processId proc
231 , sendPortLocalId = lcid
233 let sport = SendPort cid
234 chan <- liftIO newTQueueIO
235 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
236 let rport = ReceivePort $ readTQueue chan
237 let tch = TypedChannel chan'
238 return ( (channelCounter ^: (+ 1))
239 . (typedChannelWithId lcid ^= Just tch)
244 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
245 finalizer st lcid = modifyMVar_ st $
246 return . (typedChannelWithId lcid ^= Nothing)
248-- | Send a message on a typed channel
249sendChan :: Serializable a => SendPort a -> a -> Process ()
250sendChan (SendPort cid) msg = do
252 let node = localNodeId (processNode proc)
253 destNode = processNodeId (sendPortProcessId cid) in do
254 case destNode == node of
255 True -> sendChanLocal cid msg
257 liftIO $ sendBinary (processNode proc)
258 (ProcessIdentifier (processId proc))
259 (SendPortIdentifier cid)
263-- | Wait for a message on a typed channel
264receiveChan :: Serializable a => ReceivePort a -> Process a
265receiveChan = liftIO . atomically . receiveSTM
267-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
268-- non-blocking check for a message.
269receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
270receiveChanTimeout 0 ch = liftIO . atomically $
271 (Just <$> receiveSTM ch) `orElse` return Nothing
272receiveChanTimeout n ch = liftIO . timeout n . atomically $
275-- | Merge a list of typed channels.
277-- The result port is left-biased: if there are messages available on more
278-- than one port, the first available message is returned.
279mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
280mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
282-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
284mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
285mergePortsRR = \ps -> do
286 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
287 return $ ReceivePort (rr psVar)
291 rotate (x:xs) = xs ++ [x]
293 rr :: TVar [STM a] -> STM a
296 a <- foldr1 orElse ps
297 writeTVar psVar (rotate ps)
300--------------------------------------------------------------------------------
301-- Advanced messaging --
302--------------------------------------------------------------------------------
304-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
305newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
307-- | Test the matches in order against each message in the queue
308receiveWait :: [Match b] -> Process b
310 queue <- processQueue <$> ask
311 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
314-- | Like 'receiveWait' but with a timeout.
316-- If the timeout is zero do a non-blocking check for matching messages. A
317-- non-zero timeout is applied only when waiting for incoming messages (that is,
318-- /after/ we have checked the messages that are already in the mailbox).
319receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
320receiveTimeout t ms = do
321 queue <- processQueue <$> ask
322 let blockSpec = if t == 0 then NonBlocking else Timeout t
323 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
325 Nothing -> return Nothing
326 Just proc -> Just <$> proc
328matchChan :: ReceivePort a -> (a -> Process b) -> Match b
329matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
331-- | Match against any message of the right type
332match :: forall a b. Serializable a => (a -> Process b) -> Match b
333match = matchIf (const True)
335-- | Match against any message of the right type that satisfies a predicate
336matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
337matchIf c p = Match $ MatchMsg $ \msg ->
338 case messageFingerprint msg == fingerprint (undefined :: a) of
341 (UnencodedMessage _ m) ->
342 let m' = unsafeCoerce m :: a in
346 (EncodedMessage _ _) ->
347 if (c decoded) then Just (p decoded) else Nothing
350 -- Make sure the value is fully decoded so that we don't hang to
351 -- bytestrings when the process calling 'matchIf' doesn't process
352 -- the values immediately
353 !decoded = decode (messageEncoding msg)
355-- | Match against any message, regardless of the underlying (contained) type
356matchMessage :: (Message -> Process Message) -> Match Message
357matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
359-- | Match against any message (regardless of underlying type) that satisfies a predicate
360matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
361matchMessageIf c p = Match $ MatchMsg $ \msg ->
366-- | Forward a raw 'Message' to the given 'ProcessId'.
367forward :: Message -> ProcessId -> Process ()
370 let node = localNodeId (processNode proc)
371 destNode = (processNodeId them) in do
372 case destNode == node of
373 True -> sendCtrlMsg Nothing (LocalSend them msg)
374 False -> liftIO $ sendPayload (processNode proc)
375 (ProcessIdentifier (processId proc))
376 (ProcessIdentifier them)
378 (messageToPayload msg)
381-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
382-- 'Serializable' - like the datum they contain - but remember that deserializing
383-- a 'Message' will yield a 'Message', not the type within it! To obtain the
384-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
389-- send self (wrapMessage "blah")
390-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
391-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
392-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
395wrapMessage :: Serializable a => a -> Message
396wrapMessage = createMessage
398-- | Attempt to unwrap a raw 'Message'.
399-- If the type of the decoded message payload matches the expected type, the
400-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
403-- This expression, for example, will evaluate to @Nothing@
404-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
406-- Whereas this expression, will yield @Just "foo"@
407-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
409unwrapMessage :: forall a. Serializable a => Message -> Process (Maybe a)
411 case messageFingerprint msg == fingerprint (undefined :: a) of
412 False -> return Nothing
414 (UnencodedMessage _ m) ->
415 let m' = unsafeCoerce m :: a
417 (EncodedMessage _ _) ->
418 return (Just (decoded))
420 decoded :: a -- note [decoding]
421 !decoded = decode (messageEncoding msg)
423-- | Attempt to handle a raw 'Message'.
424-- If the type of the message matches the type of the first argument to
425-- the supplied expression, then the message will be decoded and the expression
426-- evaluated against its value. If this runtime type checking fails however,
427-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
428-- evaluation proceeds, the resulting value with be wrapped with @Just@.
430-- Intended for use in `catchesExit` and `matchAny` primitives.
432handleMessage :: forall a b. (Serializable a)
433 => Message -> (a -> Process b) -> Process (Maybe b)
434handleMessage msg proc = do
435 case messageFingerprint msg == fingerprint (undefined :: a) of
436 False -> return Nothing
438 (UnencodedMessage _ m) ->
439 let m' = unsafeCoerce m :: a -- note [decoding]
440 in do { r <- proc m'; return (Just r) }
441 (EncodedMessage _ _) ->
442 do { r <- proc (decoded :: a); return (Just r) }
444 decoded :: a -- note [decoding]
445 !decoded = decode (messageEncoding msg)
447-- | Match against an arbitrary message. 'matchAny' removes the first available
448-- message from the process mailbox. To handle arbitrary /raw/ messages once
449-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
451matchAny :: forall b. (Message -> Process b) -> Match b
452matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
454-- | Match against an arbitrary message. Intended for use with 'handleMessage'
455-- and 'unwrapMessage', this function /only/ removes a message from the process
456-- mailbox, /if/ the supplied condition matches. The success (or failure) of
457-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
458-- here, i.e., if the condition evaluates to @True@ then the message will
459-- be removed from the process mailbox and decoded, but that does /not/
460-- guarantee that an expression passed to @handleMessage@ will pass the
461-- runtime type checks and therefore be evaluated.
463matchAnyIf :: forall a b. (Serializable a)
465 -> (Message -> Process b)
467matchAnyIf c p = Match $ MatchMsg $ \msg ->
468 case messageFingerprint msg == fingerprint (undefined :: a) of
469 True | check -> Just (p msg)
474 (EncodedMessage _ _) -> c decoded
475 (UnencodedMessage _ m') -> c (unsafeCoerce m')
477 decoded :: a -- note [decoding]
478 !decoded = decode (messageEncoding msg)
482For an EncodedMessage, we need to ensure the value is fully decoded so that
483we don't hang to bytestrings if the calling process doesn't evaluate
484immediately. For UnencodedMessage we know (because the fingerprint comparison
485succeeds) that unsafeCoerce will not fail.
488-- | Remove any message from the queue
489matchUnknown :: Process b -> Match b
490matchUnknown p = Match $ MatchMsg (const (Just p))
492--------------------------------------------------------------------------------
493-- Process management --
494--------------------------------------------------------------------------------
496-- | Thrown by 'terminate'
497data ProcessTerminationException = ProcessTerminationException
498 deriving (Show, Typeable)
500instance Exception ProcessTerminationException
502-- | Terminate immediately (throws a ProcessTerminationException)
503terminate :: Process a
504terminate = liftIO $ throwIO ProcessTerminationException
507-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
508die :: Serializable a => a -> Process b
511 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
513-- | Forceful request to kill a process. Where 'exit' provides an exception
514-- that can be caught and handled, 'kill' throws an unexposed exception type
515-- which cannot be handled explicitly (by type).
516kill :: ProcessId -> String -> Process ()
517-- NOTE: We send the message to our local node controller, which will then
518-- forward it to a remote node controller (if applicable). Sending it directly
519-- to a remote node controller means that that the message may overtake a
520-- 'monitor' or 'link' request.
521kill them reason = sendCtrlMsg Nothing (Kill them reason)
523-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
524-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
525-- manner can be handled using the 'catchExit' family of functions.
526exit :: Serializable a => ProcessId -> a -> Process ()
527-- NOTE: We send the message to our local node controller, which will then
528-- forward it to a remote node controller (if applicable). Sending it directly
529-- to a remote node controller means that that the message may overtake a
530-- 'monitor' or 'link' request.
531exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
533-- | Catches 'ProcessExitException'. The handler will not be applied unless its
534-- type matches the encoded data stored in the exception (see the /reason/
535-- argument given to the 'exit' primitive). If the handler cannot be applied,
536-- the exception will be re-thrown.
538-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
539-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
540catchExit :: forall a b . (Show a, Serializable a)
542 -> (ProcessId -> a -> Process b)
544catchExit act exitHandler = catch act handleExit
546 handleExit ex@(ProcessExitException from msg) =
547 if messageFingerprint msg == fingerprint (undefined :: a)
548 then exitHandler from decoded
549 else liftIO $ throwIO ex
552 -- Make sure the value is fully decoded so that we don't hang to
553 -- bytestrings if the caller doesn't use the value immediately
554 !decoded = decode (messageEncoding msg)
556-- | Lift 'Control.Exception.catches' (almost).
558-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
559-- message, a handler must accept an input of the expected type. In order to
560-- handle a list of potentially different handlers (and therefore input types),
561-- a handler passed to 'catchesExit' must accept 'AbstractMessage' and return
562-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
564-- See 'maybeHandleMessage' and 'AsbtractMessage' for more details.
565catchesExit :: Process b
566 -> [(ProcessId -> Message -> (Process (Maybe b)))]
568catchesExit act handlers = catch act ((flip handleExit) handlers)
570 handleExit :: ProcessExitException
571 -> [(ProcessId -> Message -> Process (Maybe b))]
573 handleExit ex [] = liftIO $ throwIO ex
574 handleExit ex@(ProcessExitException from msg) (h:hs) = do
577 Nothing -> handleExit ex hs
580-- | Our own process ID
581getSelfPid :: Process ProcessId
582getSelfPid = processId <$> ask
584-- | Get the node ID of our local node
585getSelfNode :: Process NodeId
586getSelfNode = localNodeId . processNode <$> ask
588-- | Get information about the specified process
589getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
591 let them = processNodeId pid in do
593 dest <- mkNode them us
594 sendCtrlMsg dest $ GetInfo pid
596 match (\(p :: ProcessInfo) -> return $ Just p)
597 , match (\(_ :: ProcessInfoNone) -> return Nothing)
599 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
600 mkNode them us = case them == us of
601 True -> return Nothing
602 _ -> return $ Just them
604--------------------------------------------------------------------------------
605-- Monitoring and linking --
606--------------------------------------------------------------------------------
608-- | Link to a remote process (asynchronous)
610-- When process A links to process B (that is, process A calls
611-- @link pidB@) then an asynchronous exception will be thrown to process A
612-- when process B terminates (normally or abnormally), or when process A gets
613-- disconnected from process B. Although it is /technically/ possible to catch
614-- these exceptions, chances are if you find yourself trying to do so you should
615-- probably be using 'monitor' rather than 'link'. In particular, code such as
617-- > link pidB -- Link to process B
618-- > expect -- Wait for a message from process B
619-- > unlink pidB -- Unlink again
621-- doesn't quite do what one might expect: if process B sends a message to
622-- process A, and /subsequently terminates/, then process A might or might not
623-- be terminated too, depending on whether the exception is thrown before or
624-- after the 'unlink' (i.e., this code has a race condition).
626-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
627-- call to 'link' has no effect.
629-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
630-- Linking makes no distinction between normal and abnormal termination of
631-- the remote process.
632link :: ProcessId -> Process ()
633link = sendCtrlMsg Nothing . Link . ProcessIdentifier
635-- | Monitor another process (asynchronous)
637-- When process A monitors process B (that is, process A calls
638-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
639-- when process B terminates (normally or abnormally), or when process A gets
640-- disconnected from process B. You receive this message like any other (using
641-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
642-- 'DiedDisconnect', etc.).
644-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
645-- multiple monitors are set up, multiple notifications will be delivered
646-- and monitors can be disabled individually using 'unmonitor'.
647monitor :: ProcessId -> Process MonitorRef
648monitor = monitor' . ProcessIdentifier
650-- | Establishes temporary monitoring of another process.
652-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
653-- of @code@. Note: although monitoring is no longer active when
654-- @withMonitor@ returns, there might still be unreceived monitor
655-- messages in the queue.
657withMonitor :: ProcessId -> Process a -> Process a
658withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
659 -- unmonitor blocks waiting for the response, so there's a possibility
660 -- that an exception might interrupt withMonitor before the unmonitor
661 -- has completed. I think that's better than making the unmonitor
666-- This is synchronous in the sense that once it returns you are guaranteed
667-- that no exception will be raised if the remote process dies. However, it is
668-- asynchronous in the sense that we do not wait for a response from the remote
670unlink :: ProcessId -> Process ()
673 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
677-- | Remove a node link
679-- This has the same synchronous/asynchronous nature as 'unlink'.
680unlinkNode :: NodeId -> Process ()
683 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
687-- | Remove a channel (send port) link
689-- This has the same synchronous/asynchronous nature as 'unlink'.
690unlinkPort :: SendPort a -> Process ()
692 unlinkPortAsync sport
693 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
699-- This has the same synchronous/asynchronous nature as 'unlink'.
700unmonitor :: MonitorRef -> Process ()
703 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
707--------------------------------------------------------------------------------
708-- Exception handling --
709--------------------------------------------------------------------------------
711-- | Lift 'Control.Exception.catch'
712catch :: Exception e => Process a -> (e -> Process a) -> Process a
715 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
717-- | Lift 'Control.Exception.try'
718try :: Exception e => Process a -> Process (Either e a)
721 liftIO $ Ex.try (runLocalProcess lproc p)
723-- | Lift 'Control.Exception.mask'
724mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
727 liftIO $ Ex.mask $ \restore ->
728 runLocalProcess lproc (p (liftRestore lproc restore))
730 liftRestore :: LocalProcess -> (forall a. IO a -> IO a) -> (forall a. Process a -> Process a)
731 liftRestore lproc restoreIO = liftIO . restoreIO . runLocalProcess lproc
733-- | Lift 'Control.Exception.onException'
734onException :: Process a -> Process b -> Process a
735onException p what = p `catch` \e -> do _ <- what
736 liftIO $ throwIO (e :: SomeException)
738-- | Lift 'Control.Exception.bracket'
739bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
740bracket before after thing =
741 mask $ \restore -> do
743 r <- restore (thing a) `onException` after a
747-- | Lift 'Control.Exception.bracket_'
748bracket_ :: Process a -> Process b -> Process c -> Process c
749bracket_ before after thing = bracket before (const after) (const thing)
751-- | Lift 'Control.Exception.finally'
752finally :: Process a -> Process b -> Process a
753finally a sequel = bracket_ (return ()) sequel a
755-- | You need this when using 'catches'
756data Handler a = forall e . Exception e => Handler (e -> Process a)
758instance Functor Handler where
759 fmap f (Handler h) = Handler (fmap f . h)
761-- | Lift 'Control.Exception.catches'
762catches :: Process a -> [Handler a] -> Process a
763catches proc handlers = proc `catch` catchesHandler handlers
765catchesHandler :: [Handler a] -> SomeException -> Process a
766catchesHandler handlers e = foldr tryHandler (throw e) handlers
767 where tryHandler (Handler handler) res
768 = case fromException e of
769 Just e' -> handler e'
772--------------------------------------------------------------------------------
774--------------------------------------------------------------------------------
776-- | Like 'expect' but with a timeout
777expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
778expectTimeout n = receiveTimeout n [match return]
780-- | Asynchronous version of 'spawn'
782-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
783spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
784spawnAsync nid proc = do
785 spawnRef <- getSpawnRef
786 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
789-- | Monitor a node (asynchronous)
790monitorNode :: NodeId -> Process MonitorRef
792 monitor' . NodeIdentifier
794-- | Monitor a typed channel (asynchronous)
795monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
796monitorPort (SendPort cid) =
797 monitor' (SendPortIdentifier cid)
799-- | Remove a monitor (asynchronous)
800unmonitorAsync :: MonitorRef -> Process ()
802 sendCtrlMsg Nothing . Unmonitor
804-- | Link to a node (asynchronous)
805linkNode :: NodeId -> Process ()
806linkNode = link' . NodeIdentifier
808-- | Link to a channel (asynchronous)
809linkPort :: SendPort a -> Process ()
810linkPort (SendPort cid) =
811 link' (SendPortIdentifier cid)
813-- | Remove a process link (asynchronous)
814unlinkAsync :: ProcessId -> Process ()
816 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
818-- | Remove a node link (asynchronous)
819unlinkNodeAsync :: NodeId -> Process ()
821 sendCtrlMsg Nothing . Unlink . NodeIdentifier
823-- | Remove a channel (send port) link (asynchronous)
824unlinkPortAsync :: SendPort a -> Process ()
825unlinkPortAsync (SendPort cid) =
826 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
828--------------------------------------------------------------------------------
830--------------------------------------------------------------------------------
834-- @say message@ sends a message (time, pid of the current process, message)
835-- to the process registered as 'logger'. By default, this process simply
836-- sends the string to 'stderr'. Individual Cloud Haskell backends might
837-- replace this with a different logger process, however.
838say :: String -> Process ()
840 now <- liftIO getCurrentTime
842 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
844--------------------------------------------------------------------------------
846--------------------------------------------------------------------------------
848-- | Register a process with the local registry (asynchronous).
849-- This version will wait until a response is gotten from the
850-- management process. The name must not already be registered.
851-- The process need not be on this node.
852-- A bad registration will result in a 'ProcessRegistrationException'
854-- The process to be registered does not have to be local itself.
855register :: String -> ProcessId -> Process ()
856register = registerImpl False
858-- | Like 'register', but will replace an existing registration.
859-- The name must already be registered.
860reregister :: String -> ProcessId -> Process ()
861reregister = registerImpl True
863registerImpl :: Bool -> String -> ProcessId -> Process ()
864registerImpl force label pid = do
866 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
867 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
868 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
871-- | Register a process with a remote registry (asynchronous).
873-- The process to be registered does not have to live on the same remote node.
874-- Reply wil come in the form of a 'RegisterReply' message
876-- See comments in 'whereisRemoteAsync'
877registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
878registerRemoteAsync nid label pid =
879 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
881reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
882reregisterRemoteAsync nid label pid =
883 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
885-- | Remove a process from the local registry (asynchronous).
886-- This version will wait until a response is gotten from the
887-- management process. The name must already be registered.
888unregister :: String -> Process ()
891 sendCtrlMsg Nothing (Register label mynid Nothing False)
892 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
893 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
896-- | Deal with the result from an attempted registration or unregistration
897-- by throwing an exception if necessary
898handleRegistrationReply :: String -> Bool -> Process ()
899handleRegistrationReply label ok =
901 liftIO $ throwIO $ ProcessRegistrationException label
903-- | Remove a process from a remote registry (asynchronous).
905-- Reply wil come in the form of a 'RegisterReply' message
907-- See comments in 'whereisRemoteAsync'
908unregisterRemoteAsync :: NodeId -> String -> Process ()
909unregisterRemoteAsync nid label =
910 sendCtrlMsg (Just nid) (Register label nid Nothing False)
912-- | Query the local process registry
913whereis :: String -> Process (Maybe ProcessId)
915 sendCtrlMsg Nothing (WhereIs label)
916 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
917 (\(WhereIsReply _ mPid) -> return mPid)
920-- | Query a remote process registry (asynchronous)
922-- Reply will come in the form of a 'WhereIsReply' message.
924-- There is currently no synchronous version of 'whereisRemoteAsync': if
925-- you implement one yourself, be sure to take into account that the remote
926-- node might die or get disconnect before it can respond (i.e. you should
927-- use 'monitorNode' and take appropriate action when you receive a
928-- 'NodeMonitorNotification').
929whereisRemoteAsync :: NodeId -> String -> Process ()
930whereisRemoteAsync nid label =
931 sendCtrlMsg (Just nid) (WhereIs label)
933-- | Named send to a process in the local registry (asynchronous)
934nsend :: Serializable a => String -> a -> Process ()
936 sendCtrlMsg Nothing (NamedSend label (createMessage msg))
938-- | Named send to a process in a remote registry (asynchronous)
939nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
940nsendRemote nid label msg =
941 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
943--------------------------------------------------------------------------------
945--------------------------------------------------------------------------------
947-- | Resolve a static value
948unStatic :: Typeable a => Static a -> Process a
950 rtable <- remoteTable . processNode <$> ask
951 case Static.unstatic rtable static of
952 Left err -> fail $ "Could not resolve static value: " ++ err
955-- | Resolve a closure
956unClosure :: Typeable a => Closure a -> Process a
957unClosure closure = do
958 rtable <- remoteTable . processNode <$> ask
959 case Static.unclosure rtable closure of
960 Left err -> fail $ "Could not resolve closure: " ++ err
963--------------------------------------------------------------------------------
965--------------------------------------------------------------------------------
967-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
968-- message passing. However, when network connections get disrupted this
969-- illusion cannot always be maintained. Once a network connection breaks (even
970-- temporarily) no further communication on that connection will be possible.
971-- For example, if process A sends a message to process B, and A is then
972-- notified (by monitor notification) that it got disconnected from B, A will
973-- not be able to send any further messages to B, /unless/ A explicitly
974-- indicates that it is acceptable to attempt to reconnect to B using the
975-- Cloud Haskell 'reconnect' primitive.
977-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
978-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
979-- then receives a monitor notification that its connection to B has been lost,
980-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
983-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
984-- to attempt to reconnect on the next send/. In particular, if no further
985-- communication attempts are made to B then A can use reconnect to clean up
986-- its connection to B.
987reconnect :: ProcessId -> Process ()
990 node <- processNode <$> ask
991 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
993-- | Reconnect to a sendport. See 'reconnect' for more information.
994reconnectPort :: SendPort a -> Process ()
995reconnectPort them = do
997 node <- processNode <$> ask
998 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1000--------------------------------------------------------------------------------
1001-- Debugging/Tracing --
1002--------------------------------------------------------------------------------
1004-- | Send a message to the internal (system) trace facility. If tracing is
1005-- enabled, this will create a custom trace event. Note that several Cloud Haskell
1006-- sub-systems also generate trace events for informational/debugging purposes,
1007-- thus traces generated this way will not be the only output seen.
1009-- Just as with the "Debug.Trace" module, this is a debugging/tracing facility
1010-- for use in development, and should not be used in a production setting -
1011-- which is why the default behaviour is to trace to the GHC eventlog. For a
1012-- general purpose logging facility, you should consider 'say'.
1014-- Trace events can be written to the GHC event log, a text file, or to the
1015-- standard system logger process (see 'say'). The default behaviour for writing
1016-- to the eventlog requires specific intervention to work, without which traces
1017-- are silently dropped/ignored and no output will be generated.
1018-- The GHC eventlog documentation provides information about enabling, viewing
1019-- and working with event traces: <http://hackage.haskell.org/trac/ghc/wiki/EventLog>.
1021-- When a new local node is started, the contents of the environment variable
1022-- @DISTRIBUTED_PROCESS_TRACE_FILE@ are checked for a valid file path. If this
1023-- exists and the file can be opened for writing, all trace output will be directed
1024-- thence. If the environment variable is empty, the path invalid, or the file
1025-- unavailable for writing - e.g., because another node has already started
1026-- tracing to it - then the @DISTRIBUTED_PROCESS_TRACE_CONSOLE@ environment
1027-- variable is checked for /any/ non-empty value. If this is set, then all trace
1028-- output will be directed to the system logger process. If neither evironment
1029-- variable provides a valid trace configuration, all internal traces are written
1030-- to "Debug.Trace.traceEventIO", which writes to the GHC eventlog.
1032-- Users of the /simplelocalnet/ Cloud Haskell backend should also note that
1033-- because the trace file option only supports trace output from a single node
1034-- (so as to avoid interleaving), a file trace configured for the master node will
1035-- prevent slaves from tracing to the file and they will fall back to using the
1036-- console/'say' or eventlog instead.
1038trace :: String -> Process ()
1040 node <- processNode <$> ask
1041 liftIO $ Trace.trace (localTracer node) s
1043--------------------------------------------------------------------------------
1044-- Messages to local processes --
1045--------------------------------------------------------------------------------
1047deliverToLocalProcess :: ProcessId -> Message -> Process ()
1048deliverToLocalProcess pid msg = do
1049 withLocalProc pid $ \p -> enqueue (processQueue p) msg
1051deliverToLocalPort :: SendPortId -> Message -> Process ()
1052deliverToLocalPort spId msg =
1053 let pid = sendPortProcessId spId
1054 cid = sendPortLocalId spId
1056 withLocalProc pid $ \proc -> do
1057 mChan <- withMVar (processState proc) $ return . (^. typedChannelWithId cid)
1059 -- in the unlikely event we know nothing about this channel id,
1060 -- see [note: missing recipients]
1061 Nothing -> return ()
1062 Just (TypedChannel chan') -> do
1063 -- If ch is Nothing, the process has given up the read end of
1064 -- the channel and we simply ignore the incoming message - this
1065 ch <- deRefWeak chan'
1066 forM_ ch $ \chan -> deliverChan msg chan
1067 where deliverChan :: forall a . Message -> TQueue a -> IO ()
1068 deliverChan (UnencodedMessage _ raw) chan' =
1069 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1070 deliverChan (EncodedMessage _ _) _ =
1071 -- this will /never/ happen
1072 error "invalid local channel delivery"
1074withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> Process ()
1075withLocalProc pid p = do
1076 node <- processNode <$> ask
1078 -- see [note: missing recipients]
1079 let lpid = processLocalId pid
1080 mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
1083-- [note: missing recipients]
1084-- By [Unified: table 6, rule missing_process] messages to dead processes
1085-- can silently be dropped
1087--------------------------------------------------------------------------------
1088-- Auxiliary functions --
1089--------------------------------------------------------------------------------
1091sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1093 -- sendCtrlMsg Nothing $ LocalSend pid (createUnencodedMessage msg)
1094 deliverToLocalProcess pid (createUnencodedMessage msg)
1096sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1097sendChanLocal spId msg =
1098 -- sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1099 deliverToLocalPort spId (createUnencodedMessage msg)
1101getMonitorRefFor :: Identifier -> Process MonitorRef
1102getMonitorRefFor ident = do
1104 liftIO $ modifyMVar (processState proc) $ \st -> do
1105 let counter = st ^. monitorCounter
1106 return ( monitorCounter ^: (+ 1) $ st
1107 , MonitorRef ident counter
1110getSpawnRef :: Process SpawnRef
1113 liftIO $ modifyMVar (processState proc) $ \st -> do
1114 let counter = st ^. spawnCounter
1115 return ( spawnCounter ^: (+ 1) $ st
1119-- | Monitor a process/node/channel
1120monitor' :: Identifier -> Process MonitorRef
1122 monitorRef <- getMonitorRefFor ident
1123 sendCtrlMsg Nothing $ Monitor monitorRef
1126-- | Link to a process/node/channel
1127link' :: Identifier -> Process ()
1128link' = sendCtrlMsg Nothing . Link
1130-- Send a control message
1131sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node
1132 -> ProcessSignal -- ^ Message to send
1134sendCtrlMsg mNid signal = do
1136 let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc)
1137 , ctrlMsgSignal = signal
1141 liftIO $ writeChan (localCtrlChan (processNode proc)) msg
1143 liftIO $ sendBinary (processNode proc)
1144 (ProcessIdentifier (processId proc))
1145 (NodeIdentifier nid)
1146 WithImplicitReconnect
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