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
43 -- * Process management
45 , ProcessTerminationException(..)
51 -- keep the exception constructor hidden, so that handling exit
52 -- reasons /must/ take place via the 'catchExit' family of primitives
53 , ProcessExitException()
58 -- * Monitoring and linking
73 , reregisterRemoteAsync
74 , unregisterRemoteAsync
80 -- * Exception handling
103 -- * Internal Exports
107#if ! MIN_VERSION_base(4,6,0)
108import Prelude hiding (catch)
111import Data.Binary (decode)
112import Data.Time.Clock (getCurrentTime)
113import Data.Time.Format (formatTime)
114import System.Locale (defaultTimeLocale)
115import System.Timeout (timeout)
116import Control.Monad (when)
117import Control.Monad.Reader (ask)
118import Control.Monad.IO.Class (MonadIO, liftIO)
119import Control.Applicative ((<$>))
120import Control.Exception (Exception(..), throw, throwIO, SomeException)
121import qualified Control.Exception as Ex (catch, mask, try)
122import Control.Distributed.Process.Internal.StrictMVar
127import Control.Concurrent.STM
136import Control.Distributed.Process.Internal.CQueue
141import Control.Distributed.Process.Serializable (Serializable, fingerprint)
142import Data.Accessor ((^.), (^:), (^=))
143import Control.Distributed.Static
147import Data.Rank1Typeable (Typeable)
148import qualified Control.Distributed.Static as Static (unstatic, unclosure)
149import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
150import Control.Distributed.Process.Internal.Types
169 , ProcessExitException(..)
171 , DidUnlinkProcess(..)
176 , ProcessRegistrationException(..)
178 , ProcessInfoNone(..)
181 , createUnencodedMessage
183 , ImplicitReconnect( NoImplicitReconnect)
188import Control.Distributed.Process.Internal.Messaging
195import Control.Distributed.Process.Management.Types
198import Control.Distributed.Process.Management.Trace.Types
201import Control.Distributed.Process.Internal.WeakTQueue
209--------------------------------------------------------------------------------
211--------------------------------------------------------------------------------
214send :: Serializable a => ProcessId -> a -> Process ()
215-- This requires a lookup on every send. If we want to avoid that we need to
216-- modify serializable to allow for stateful (IO) deserialization.
219 let us = processId proc
220 node = processNode proc
221 nodeId = localNodeId node
222 destNode = (processNodeId them) in do
223 case destNode == nodeId of
224 True -> sendLocal them msg
225 False -> liftIO $ sendMessage (processNode proc)
226 (ProcessIdentifier (processId proc))
227 (ProcessIdentifier them)
230 -- We do not fire the trace event until after the sending is complete;
231 -- In the remote case, 'sendMessage' can block in the networking stack.
232 liftIO $ traceEvent (localEventBus node)
233 (MxSent them us (createUnencodedMessage msg))
235-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
236-- and (in the case when the destination process resides on the same local
237-- node) therefore ensure that the payload is fully evaluated before it is
239unsafeSend :: Serializable a => ProcessId -> a -> Process ()
240unsafeSend = Unsafe.send
242-- | Wait for a message of a specific type
243expect :: forall a. Serializable a => Process a
244expect = receiveWait [match return]
246--------------------------------------------------------------------------------
248--------------------------------------------------------------------------------
250-- | Create a new typed channel
251newChan :: Serializable a => Process (SendPort a, ReceivePort a)
254 liftIO . modifyMVar (processState proc) $ \st -> do
255 let lcid = st ^. channelCounter
256 let cid = SendPortId { sendPortProcessId = processId proc
257 , sendPortLocalId = lcid
259 let sport = SendPort cid
260 chan <- liftIO newTQueueIO
261 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
262 let rport = ReceivePort $ readTQueue chan
263 let tch = TypedChannel chan'
264 return ( (channelCounter ^: (+ 1))
265 . (typedChannelWithId lcid ^= Just tch)
270 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
271 finalizer st lcid = modifyMVar_ st $
272 return . (typedChannelWithId lcid ^= Nothing)
274-- | Send a message on a typed channel
275sendChan :: Serializable a => SendPort a -> a -> Process ()
276sendChan (SendPort cid) msg = do
278 let node = localNodeId (processNode proc)
279 destNode = processNodeId (sendPortProcessId cid) in do
280 case destNode == node of
281 True -> sendChanLocal cid msg
283 liftIO $ sendBinary (processNode proc)
284 (ProcessIdentifier (processId proc))
285 (SendPortIdentifier cid)
289-- | Send a message on a typed channel. This function makes /no/ attempt to
290-- serialize and (in the case when the @ReceivePort@ resides on the same local
291-- node) therefore ensure that the payload is fully evaluated before it is
293unsafeSendChan :: Serializable a => SendPort a -> a -> Process ()
294unsafeSendChan = Unsafe.sendChan
296-- | Wait for a message on a typed channel
297receiveChan :: Serializable a => ReceivePort a -> Process a
298receiveChan = liftIO . atomically . receiveSTM
300-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
301-- non-blocking check for a message.
302receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
303receiveChanTimeout 0 ch = liftIO . atomically $
304 (Just <$> receiveSTM ch) `orElse` return Nothing
305receiveChanTimeout n ch = liftIO . timeout n . atomically $
308-- | Merge a list of typed channels.
310-- The result port is left-biased: if there are messages available on more
311-- than one port, the first available message is returned.
312mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
313mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
315-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
317mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
318mergePortsRR = \ps -> do
319 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
320 return $ ReceivePort (rr psVar)
324 rotate (x:xs) = xs ++ [x]
326 rr :: TVar [STM a] -> STM a
329 a <- foldr1 orElse ps
330 writeTVar psVar (rotate ps)
333--------------------------------------------------------------------------------
334-- Advanced messaging --
335--------------------------------------------------------------------------------
337-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
338newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
340-- | Test the matches in order against each message in the queue
341receiveWait :: [Match b] -> Process b
343 queue <- processQueue <$> ask
344 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
347-- | Like 'receiveWait' but with a timeout.
349-- If the timeout is zero do a non-blocking check for matching messages. A
350-- non-zero timeout is applied only when waiting for incoming messages (that is,
351-- /after/ we have checked the messages that are already in the mailbox).
352receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
353receiveTimeout t ms = do
354 queue <- processQueue <$> ask
355 let blockSpec = if t == 0 then NonBlocking else Timeout t
356 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
358 Nothing -> return Nothing
359 Just proc -> Just <$> proc
361matchChan :: ReceivePort a -> (a -> Process b) -> Match b
362matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
364-- | Match against any message of the right type
365match :: forall a b. Serializable a => (a -> Process b) -> Match b
366match = matchIf (const True)
368-- | Match against any message of the right type that satisfies a predicate
369matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
370matchIf c p = Match $ MatchMsg $ \msg ->
371 case messageFingerprint msg == fingerprint (undefined :: a) of
374 (UnencodedMessage _ m) ->
375 let m' = unsafeCoerce m :: a in
379 (EncodedMessage _ _) ->
380 if (c decoded) then Just (p decoded) else Nothing
383 -- Make sure the value is fully decoded so that we don't hang to
384 -- bytestrings when the process calling 'matchIf' doesn't process
385 -- the values immediately
386 !decoded = decode (messageEncoding msg)
388-- | Match against any message, regardless of the underlying (contained) type
389matchMessage :: (Message -> Process Message) -> Match Message
390matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
392-- | Match against any message (regardless of underlying type) that satisfies a predicate
393matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
394matchMessageIf c p = Match $ MatchMsg $ \msg ->
399-- | Forward a raw 'Message' to the given 'ProcessId'.
400forward :: Message -> ProcessId -> Process ()
403 let node = processNode proc
405 nid = localNodeId node
406 destNode = (processNodeId them) in do
407 case destNode == nid of
408 True -> sendCtrlMsg Nothing (LocalSend them msg)
409 False -> liftIO $ sendPayload (processNode proc)
410 (ProcessIdentifier (processId proc))
411 (ProcessIdentifier them)
413 (messageToPayload msg)
414 -- We do not fire the trace event until after the sending is complete;
415 -- In the remote case, 'sendMessage' can block in the networking stack.
416 liftIO $ traceEvent (localEventBus node)
420-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
421-- 'Serializable' - like the datum they contain - but also note, deserialising
422-- such a 'Message' will yield a 'Message', not the type within it! To obtain the
423-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
428-- send self (wrapMessage "blah")
429-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
430-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
431-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
434wrapMessage :: Serializable a => a -> Message
435wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
437-- | This is the /unsafe/ variant of 'wrapMessage'. See
438-- "Control.Distributed.Process.UnsafePrimitives" for details.
439unsafeWrapMessage :: Serializable a => a -> Message
440unsafeWrapMessage = Unsafe.wrapMessage
442-- [note Serializable UnencodedMessage]
443-- if we attempt to serialise an UnencodedMessage, it will be converted to an
444-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
445-- Primitives.hs for the gory details
447-- | Attempt to unwrap a raw 'Message'.
448-- If the type of the decoded message payload matches the expected type, the
449-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
452-- This expression, for example, will evaluate to @Nothing@
453-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
455-- Whereas this expression, will yield @Just "foo"@
456-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
458unwrapMessage :: forall m a. (Monad m, Serializable a) => Message -> m (Maybe a)
460 case messageFingerprint msg == fingerprint (undefined :: a) of
461 False -> return Nothing :: m (Maybe a)
463 (UnencodedMessage _ ms) ->
464 let ms' = unsafeCoerce ms :: a
466 (EncodedMessage _ _) ->
467 return (Just (decoded))
469 decoded :: a -- note [decoding]
470 !decoded = decode (messageEncoding msg)
472-- | Attempt to handle a raw 'Message'.
473-- If the type of the message matches the type of the first argument to
474-- the supplied expression, then the message will be decoded and the expression
475-- evaluated against its value. If this runtime type checking fails however,
476-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
477-- evaluation proceeds, the resulting value with be wrapped with @Just@.
479-- Intended for use in `catchesExit` and `matchAny` primitives.
481handleMessage :: forall m a b. (Monad m, Serializable a)
482 => Message -> (a -> m b) -> m (Maybe b)
483handleMessage msg proc = handleMessageIf msg (const True) proc
485handleMessageIf :: forall m a b . (Monad m, Serializable a)
490handleMessageIf msg c proc = do
491 case messageFingerprint msg == fingerprint (undefined :: a) of
492 False -> return Nothing :: m (Maybe b)
494 (UnencodedMessage _ ms) ->
495 let ms' = unsafeCoerce ms :: a in
497 True -> do { r <- proc ms'; return (Just r) }
498 False -> return Nothing :: m (Maybe b)
499 (EncodedMessage _ _) ->
501 True -> do { r <- proc (decoded :: a); return (Just r) }
502 False -> return Nothing :: m (Maybe b)
504 decoded :: a -- note [decoding]
505 !decoded = decode (messageEncoding msg)
507-- | Match against an arbitrary message. 'matchAny' removes the first available
508-- message from the process mailbox. To handle arbitrary /raw/ messages once
509-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
511matchAny :: forall b. (Message -> Process b) -> Match b
512matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
514-- | Match against an arbitrary message. Intended for use with 'handleMessage'
515-- and 'unwrapMessage', this function /only/ removes a message from the process
516-- mailbox, /if/ the supplied condition matches. The success (or failure) of
517-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
518-- here, i.e., if the condition evaluates to @True@ then the message will
519-- be removed from the process mailbox and decoded, but that does /not/
520-- guarantee that an expression passed to @handleMessage@ will pass the
521-- runtime type checks and therefore be evaluated.
523matchAnyIf :: forall a b. (Serializable a)
525 -> (Message -> Process b)
527matchAnyIf c p = Match $ MatchMsg $ \msg ->
528 case messageFingerprint msg == fingerprint (undefined :: a) of
529 True | check -> Just (p msg)
534 (EncodedMessage _ _) -> c decoded
535 (UnencodedMessage _ m') -> c (unsafeCoerce m')
537 decoded :: a -- note [decoding]
538 !decoded = decode (messageEncoding msg)
542For an EncodedMessage, we need to ensure the value is fully decoded so that
543we don't hang to bytestrings if the calling process doesn't evaluate
544immediately. For UnencodedMessage we know (because the fingerprint comparison
545succeeds) that unsafeCoerce will not fail.
548-- | Remove any message from the queue
549matchUnknown :: Process b -> Match b
550matchUnknown p = Match $ MatchMsg (const (Just p))
552-- | Receives messages and forwards them to @pid@ if @p msg == True@.
553delegate :: ProcessId -> (Message -> Bool) -> Process ()
556 matchAny (\m -> case (p m) of
557 True -> forward m pid
562-- | A straight relay that forwards all messages to the supplied pid.
563relay :: ProcessId -> Process ()
564relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
566-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
567-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
568-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
569-- fails, no action is taken and the proxy process will continue running.
570proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
574 next <- handleMessage m proc
576 Just True -> forward m pid
577 Just False -> return () -- explicitly ignored
578 Nothing -> return ()) -- un-routable
582--------------------------------------------------------------------------------
583-- Process management --
584--------------------------------------------------------------------------------
586-- | Thrown by 'terminate'
587data ProcessTerminationException = ProcessTerminationException
588 deriving (Show, Typeable)
590instance Exception ProcessTerminationException
592-- | Terminate immediately (throws a ProcessTerminationException)
593terminate :: Process a
594terminate = liftIO $ throwIO ProcessTerminationException
597-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
598die :: Serializable a => a -> Process b
600-- NOTE: We do /not/ want to use UnencodedMessage here, as the exception
601-- could be decoded by a handler passed to 'catchExit', re-thrown or even
602-- passed to another process via the tracing mechanism.
604 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
606-- | Forceful request to kill a process. Where 'exit' provides an exception
607-- that can be caught and handled, 'kill' throws an unexposed exception type
608-- which cannot be handled explicitly (by type).
609kill :: ProcessId -> String -> Process ()
610-- NOTE: We send the message to our local node controller, which will then
611-- forward it to a remote node controller (if applicable). Sending it directly
612-- to a remote node controller means that that the message may overtake a
613-- 'monitor' or 'link' request.
614kill them reason = sendCtrlMsg Nothing (Kill them reason)
616-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
617-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
618-- manner can be handled using the 'catchExit' family of functions.
619exit :: Serializable a => ProcessId -> a -> Process ()
620-- NOTE: We send the message to our local node controller, which will then
621-- forward it to a remote node controller (if applicable). Sending it directly
622-- to a remote node controller means that that the message may overtake a
623-- 'monitor' or 'link' request.
624exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
626-- | Catches 'ProcessExitException'. The handler will not be applied unless its
627-- type matches the encoded data stored in the exception (see the /reason/
628-- argument given to the 'exit' primitive). If the handler cannot be applied,
629-- the exception will be re-thrown.
631-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
632-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
633catchExit :: forall a b . (Show a, Serializable a)
635 -> (ProcessId -> a -> Process b)
637catchExit act exitHandler = catch act handleExit
639 handleExit ex@(ProcessExitException from msg) =
640 if messageFingerprint msg == fingerprint (undefined :: a)
641 then exitHandler from decoded
642 else liftIO $ throwIO ex
645 -- Make sure the value is fully decoded so that we don't hang to
646 -- bytestrings if the caller doesn't use the value immediately
647 !decoded = decode (messageEncoding msg)
649-- | Lift 'Control.Exception.catches' (almost).
651-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
652-- message, a handler must accept inputs of the expected type. In order to
653-- handle a list of potentially different handlers (and therefore input types),
654-- a handler passed to 'catchesExit' must accept 'Message' and return
655-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
657-- See 'maybeHandleMessage' and 'Message' for more details.
658catchesExit :: Process b
659 -> [(ProcessId -> Message -> (Process (Maybe b)))]
661catchesExit act handlers = catch act ((flip handleExit) handlers)
663 handleExit :: ProcessExitException
664 -> [(ProcessId -> Message -> Process (Maybe b))]
666 handleExit ex [] = liftIO $ throwIO ex
667 handleExit ex@(ProcessExitException from msg) (h:hs) = do
670 Nothing -> handleExit ex hs
673-- | Our own process ID
674getSelfPid :: Process ProcessId
675getSelfPid = processId <$> ask
677-- | Get the node ID of our local node
678getSelfNode :: Process NodeId
679getSelfNode = localNodeId . processNode <$> ask
681-- | Get information about the specified process
682getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
684 let them = processNodeId pid in do
686 dest <- mkNode them us
687 sendCtrlMsg dest $ GetInfo pid
689 match (\(p :: ProcessInfo) -> return $ Just p)
690 , match (\(_ :: ProcessInfoNone) -> return Nothing)
692 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
693 mkNode them us = case them == us of
694 True -> return Nothing
695 _ -> return $ Just them
697--------------------------------------------------------------------------------
698-- Monitoring and linking --
699--------------------------------------------------------------------------------
701-- | Link to a remote process (asynchronous)
703-- When process A links to process B (that is, process A calls
704-- @link pidB@) then an asynchronous exception will be thrown to process A
705-- when process B terminates (normally or abnormally), or when process A gets
706-- disconnected from process B. Although it is /technically/ possible to catch
707-- these exceptions, chances are if you find yourself trying to do so you should
708-- probably be using 'monitor' rather than 'link'. In particular, code such as
710-- > link pidB -- Link to process B
711-- > expect -- Wait for a message from process B
712-- > unlink pidB -- Unlink again
714-- doesn't quite do what one might expect: if process B sends a message to
715-- process A, and /subsequently terminates/, then process A might or might not
716-- be terminated too, depending on whether the exception is thrown before or
717-- after the 'unlink' (i.e., this code has a race condition).
719-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
720-- call to 'link' has no effect.
722-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
723-- Linking makes no distinction between normal and abnormal termination of
724-- the remote process.
725link :: ProcessId -> Process ()
726link = sendCtrlMsg Nothing . Link . ProcessIdentifier
728-- | Monitor another process (asynchronous)
730-- When process A monitors process B (that is, process A calls
731-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
732-- when process B terminates (normally or abnormally), or when process A gets
733-- disconnected from process B. You receive this message like any other (using
734-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
735-- 'DiedDisconnect', etc.).
737-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
738-- multiple monitors are set up, multiple notifications will be delivered
739-- and monitors can be disabled individually using 'unmonitor'.
740monitor :: ProcessId -> Process MonitorRef
741monitor = monitor' . ProcessIdentifier
743-- | Establishes temporary monitoring of another process.
745-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
746-- of @code@. Note: although monitoring is no longer active when
747-- @withMonitor@ returns, there might still be unreceived monitor
748-- messages in the queue.
750withMonitor :: ProcessId -> Process a -> Process a
751withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
752 -- unmonitor blocks waiting for the response, so there's a possibility
753 -- that an exception might interrupt withMonitor before the unmonitor
754 -- has completed. I think that's better than making the unmonitor
759-- This is synchronous in the sense that once it returns you are guaranteed
760-- that no exception will be raised if the remote process dies. However, it is
761-- asynchronous in the sense that we do not wait for a response from the remote
763unlink :: ProcessId -> Process ()
766 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
770-- | Remove a node link
772-- This has the same synchronous/asynchronous nature as 'unlink'.
773unlinkNode :: NodeId -> Process ()
776 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
780-- | Remove a channel (send port) link
782-- This has the same synchronous/asynchronous nature as 'unlink'.
783unlinkPort :: SendPort a -> Process ()
785 unlinkPortAsync sport
786 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
792-- This has the same synchronous/asynchronous nature as 'unlink'.
793unmonitor :: MonitorRef -> Process ()
796 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
800--------------------------------------------------------------------------------
801-- Exception handling --
802--------------------------------------------------------------------------------
804-- | Lift 'Control.Exception.catch'
805catch :: Exception e => Process a -> (e -> Process a) -> Process a
808 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
810-- | Lift 'Control.Exception.try'
811try :: Exception e => Process a -> Process (Either e a)
814 liftIO $ Ex.try (runLocalProcess lproc p)
816-- | Lift 'Control.Exception.mask'
817mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
820 liftIO $ Ex.mask $ \restore ->
821 runLocalProcess lproc (p (liftRestore restore))
823 liftRestore :: (forall a. IO a -> IO a)
824 -> (forall a. Process a -> Process a)
825 liftRestore restoreIO = \p2 -> do
827 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
829-- | Lift 'Control.Exception.onException'
830onException :: Process a -> Process b -> Process a
831onException p what = p `catch` \e -> do _ <- what
832 liftIO $ throwIO (e :: SomeException)
834-- | Lift 'Control.Exception.bracket'
835bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
836bracket before after thing =
837 mask $ \restore -> do
839 r <- restore (thing a) `onException` after a
843-- | Lift 'Control.Exception.bracket_'
844bracket_ :: Process a -> Process b -> Process c -> Process c
845bracket_ before after thing = bracket before (const after) (const thing)
847-- | Lift 'Control.Exception.finally'
848finally :: Process a -> Process b -> Process a
849finally a sequel = bracket_ (return ()) sequel a
851-- | You need this when using 'catches'
852data Handler a = forall e . Exception e => Handler (e -> Process a)
854instance Functor Handler where
855 fmap f (Handler h) = Handler (fmap f . h)
857-- | Lift 'Control.Exception.catches'
858catches :: Process a -> [Handler a] -> Process a
859catches proc handlers = proc `catch` catchesHandler handlers
861catchesHandler :: [Handler a] -> SomeException -> Process a
862catchesHandler handlers e = foldr tryHandler (throw e) handlers
863 where tryHandler (Handler handler) res
864 = case fromException e of
865 Just e' -> handler e'
868--------------------------------------------------------------------------------
870--------------------------------------------------------------------------------
872-- | Like 'expect' but with a timeout
873expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
874expectTimeout n = receiveTimeout n [match return]
876-- | Asynchronous version of 'spawn'
878-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
879spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
880spawnAsync nid proc = do
881 spawnRef <- getSpawnRef
882 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
885-- | Monitor a node (asynchronous)
886monitorNode :: NodeId -> Process MonitorRef
888 monitor' . NodeIdentifier
890-- | Monitor a typed channel (asynchronous)
891monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
892monitorPort (SendPort cid) =
893 monitor' (SendPortIdentifier cid)
895-- | Remove a monitor (asynchronous)
896unmonitorAsync :: MonitorRef -> Process ()
898 sendCtrlMsg Nothing . Unmonitor
900-- | Link to a node (asynchronous)
901linkNode :: NodeId -> Process ()
902linkNode = link' . NodeIdentifier
904-- | Link to a channel (asynchronous)
905linkPort :: SendPort a -> Process ()
906linkPort (SendPort cid) =
907 link' (SendPortIdentifier cid)
909-- | Remove a process link (asynchronous)
910unlinkAsync :: ProcessId -> Process ()
912 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
914-- | Remove a node link (asynchronous)
915unlinkNodeAsync :: NodeId -> Process ()
917 sendCtrlMsg Nothing . Unlink . NodeIdentifier
919-- | Remove a channel (send port) link (asynchronous)
920unlinkPortAsync :: SendPort a -> Process ()
921unlinkPortAsync (SendPort cid) =
922 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
924--------------------------------------------------------------------------------
926--------------------------------------------------------------------------------
930-- @say message@ sends a message (time, pid of the current process, message)
931-- to the process registered as 'logger'. By default, this process simply
932-- sends the string to 'stderr'. Individual Cloud Haskell backends might
933-- replace this with a different logger process, however.
934say :: String -> Process ()
936 now <- liftIO getCurrentTime
938 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
940--------------------------------------------------------------------------------
942--------------------------------------------------------------------------------
944-- | Register a process with the local registry (asynchronous).
945-- This version will wait until a response is gotten from the
946-- management process. The name must not already be registered.
947-- The process need not be on this node.
948-- A bad registration will result in a 'ProcessRegistrationException'
950-- The process to be registered does not have to be local itself.
951register :: String -> ProcessId -> Process ()
952register = registerImpl False
954-- | Like 'register', but will replace an existing registration.
955-- The name must already be registered.
956reregister :: String -> ProcessId -> Process ()
957reregister = registerImpl True
959registerImpl :: Bool -> String -> ProcessId -> Process ()
960registerImpl force label pid = do
962 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
963 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
964 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
967-- | Register a process with a remote registry (asynchronous).
969-- The process to be registered does not have to live on the same remote node.
970-- Reply wil come in the form of a 'RegisterReply' message
972-- See comments in 'whereisRemoteAsync'
973registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
974registerRemoteAsync nid label pid =
975 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
977reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
978reregisterRemoteAsync nid label pid =
979 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
981-- | Remove a process from the local registry (asynchronous).
982-- This version will wait until a response is gotten from the
983-- management process. The name must already be registered.
984unregister :: String -> Process ()
987 sendCtrlMsg Nothing (Register label mynid Nothing False)
988 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
989 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
992-- | Deal with the result from an attempted registration or unregistration
993-- by throwing an exception if necessary
994handleRegistrationReply :: String -> Bool -> Process ()
995handleRegistrationReply label ok =
997 liftIO $ throwIO $ ProcessRegistrationException label
999-- | Remove a process from a remote registry (asynchronous).
1001-- Reply wil come in the form of a 'RegisterReply' message
1003-- See comments in 'whereisRemoteAsync'
1004unregisterRemoteAsync :: NodeId -> String -> Process ()
1005unregisterRemoteAsync nid label =
1006 sendCtrlMsg (Just nid) (Register label nid Nothing False)
1008-- | Query the local process registry
1009whereis :: String -> Process (Maybe ProcessId)
1011 sendCtrlMsg Nothing (WhereIs label)
1012 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
1013 (\(WhereIsReply _ mPid) -> return mPid)
1016-- | Query a remote process registry (asynchronous)
1018-- Reply will come in the form of a 'WhereIsReply' message.
1020-- There is currently no synchronous version of 'whereisRemoteAsync': if
1021-- you implement one yourself, be sure to take into account that the remote
1022-- node might die or get disconnect before it can respond (i.e. you should
1023-- use 'monitorNode' and take appropriate action when you receive a
1024-- 'NodeMonitorNotification').
1025whereisRemoteAsync :: NodeId -> String -> Process ()
1026whereisRemoteAsync nid label =
1027 sendCtrlMsg (Just nid) (WhereIs label)
1029-- | Named send to a process in the local registry (asynchronous)
1030nsend :: Serializable a => String -> a -> Process ()
1032 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1034-- | Named send to a process in the local registry (asynchronous).
1035-- This function makes /no/ attempt to serialize and (in the case when the
1036-- destination process resides on the same local node) therefore ensure that
1037-- the payload is fully evaluated before it is delivered.
1038unsafeNSend :: Serializable a => String -> a -> Process ()
1039unsafeNSend = Unsafe.nsend
1041-- | Named send to a process in a remote registry (asynchronous)
1042nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
1043nsendRemote nid label msg =
1044 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1046--------------------------------------------------------------------------------
1048--------------------------------------------------------------------------------
1050-- | Resolve a static value
1051unStatic :: Typeable a => Static a -> Process a
1053 rtable <- remoteTable . processNode <$> ask
1054 case Static.unstatic rtable static of
1055 Left err -> fail $ "Could not resolve static value: " ++ err
1058-- | Resolve a closure
1059unClosure :: Typeable a => Closure a -> Process a
1060unClosure closure = do
1061 rtable <- remoteTable . processNode <$> ask
1062 case Static.unclosure rtable closure of
1063 Left err -> fail $ "Could not resolve closure: " ++ err
1066--------------------------------------------------------------------------------
1068--------------------------------------------------------------------------------
1070-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1071-- message passing. However, when network connections get disrupted this
1072-- illusion cannot always be maintained. Once a network connection breaks (even
1073-- temporarily) no further communication on that connection will be possible.
1074-- For example, if process A sends a message to process B, and A is then
1075-- notified (by monitor notification) that it got disconnected from B, A will
1076-- not be able to send any further messages to B, /unless/ A explicitly
1077-- indicates that it is acceptable to attempt to reconnect to B using the
1078-- Cloud Haskell 'reconnect' primitive.
1080-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1081-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1082-- then receives a monitor notification that its connection to B has been lost,
1083-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1084-- and m3 but not m2.
1086-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1087-- to attempt to reconnect on the next send/. In particular, if no further
1088-- communication attempts are made to B then A can use reconnect to clean up
1089-- its connection to B.
1090reconnect :: ProcessId -> Process ()
1093 node <- processNode <$> ask
1094 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1096-- | Reconnect to a sendport. See 'reconnect' for more information.
1097reconnectPort :: SendPort a -> Process ()
1098reconnectPort them = do
1100 node <- processNode <$> ask
1101 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1103--------------------------------------------------------------------------------
1104-- Auxiliary functions --
1105--------------------------------------------------------------------------------
1107sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1109 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1111sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1112sendChanLocal spId msg =
1113 -- we *must* fully serialize/encode the message here, because
1114 -- attempting to use `unsafeCoerce' in the node controller
1115 -- won't work since we know nothing about the required type
1116 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1118getMonitorRefFor :: Identifier -> Process MonitorRef
1119getMonitorRefFor ident = do
1121 liftIO $ modifyMVar (processState proc) $ \st -> do
1122 let counter = st ^. monitorCounter
1123 return ( monitorCounter ^: (+ 1) $ st
1124 , MonitorRef ident counter
1127getSpawnRef :: Process SpawnRef
1130 liftIO $ modifyMVar (processState proc) $ \st -> do
1131 let counter = st ^. spawnCounter
1132 return ( spawnCounter ^: (+ 1) $ st
1136-- | Monitor a process/node/channel
1137monitor' :: Identifier -> Process MonitorRef
1139 monitorRef <- getMonitorRefFor ident
1140 sendCtrlMsg Nothing $ Monitor monitorRef
1143-- | Link to a process/node/channel
1144link' :: Identifier -> Process ()
1145link' = 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