git.iko.soy/distributed-process
on 3178dec help sign in
distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs
1112 lines 42 kB history raw
1{-# LANGUAGE BangPatterns #-}
3-- | Cloud Haskell primitives
4--
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
8 ( -- * Basic messaging
9 send
10 , expect
11 -- * Channels
12 , newChan
13 , sendChan
14 , receiveChan
15 , mergePortsBiased
16 , mergePortsRR
17 -- * Advanced messaging
18 , Match
19 , receiveWait
20 , receiveTimeout
21 , match
22 , matchIf
23 , matchUnknown
24 , matchAny
25 , matchAnyIf
26 , matchChan
27 , matchMessage
28 , matchMessageIf
29 , wrapMessage
30 , unwrapMessage
31 , handleMessage
32 , forward
33 , delegate
34 , relay
35 , proxy
36 -- * Process management
37 , terminate
38 , ProcessTerminationException(..)
39 , die
40 , kill
41 , exit
42 , catchExit
43 , catchesExit
44 -- keep the exception constructor hidden, so that handling exit
45 -- reasons /must/ take place via the 'catchExit' family of primitives
46 , ProcessExitException()
47 , getSelfPid
48 , getSelfNode
49 , ProcessInfo(..)
50 , getProcessInfo
51 -- * Monitoring and linking
52 , link
53 , unlink
54 , monitor
55 , unmonitor
56 , withMonitor
57 -- * Logging
58 , say
59 -- * Registry
60 , register
61 , reregister
62 , unregister
63 , whereis
64 , nsend
65 , registerRemoteAsync
66 , reregisterRemoteAsync
67 , unregisterRemoteAsync
68 , whereisRemoteAsync
69 , nsendRemote
70 -- * Closures
71 , unClosure
72 , unStatic
73 -- * Exception handling
74 , catch
75 , Handler(..)
76 , catches
77 , try
78 , mask
79 , onException
80 , bracket
81 , bracket_
82 , finally
83 -- * Auxiliary API
84 , expectTimeout
85 , receiveChanTimeout
86 , spawnAsync
87 , linkNode
88 , linkPort
89 , unlinkNode
90 , unlinkPort
91 , monitorNode
92 , monitorPort
93 -- * Reconnecting
94 , reconnect
95 , reconnectPort
96 ) where
98#if ! MIN_VERSION_base(4,6,0)
99import Prelude hiding (catch)
100#endif
102import Data.Binary (decode)
103import Data.Time.Clock (getCurrentTime)
104import Data.Time.Format (formatTime)
105import System.Locale (defaultTimeLocale)
106import System.Timeout (timeout)
107import Control.Monad (when)
108import Control.Monad.Reader (ask)
109import Control.Monad.IO.Class (MonadIO, liftIO)
110import Control.Applicative ((<$>))
111import Control.Exception (Exception(..), throw, throwIO, SomeException)
112import qualified Control.Exception as Ex (catch, mask, try)
113import Control.Distributed.Process.Internal.StrictMVar
114 ( StrictMVar
115 , modifyMVar
116 , modifyMVar_
117 )
118import Control.Concurrent.Chan (writeChan)
119import Control.Concurrent.STM
120 ( STM
121 , TVar
122 , atomically
123 , orElse
124 , newTVar
125 , readTVar
126 , writeTVar
127 )
128import Control.Distributed.Process.Internal.CQueue
129 ( dequeue
130 , BlockSpec(..)
131 , MatchOn(..)
132 )
133import Control.Distributed.Process.Serializable (Serializable, fingerprint)
134import Data.Accessor ((^.), (^:), (^=))
135import Control.Distributed.Static
136 ( Static
137 , Closure
138 )
139import Data.Rank1Typeable (Typeable)
140import qualified Control.Distributed.Static as Static (unstatic, unclosure)
141import Control.Distributed.Process.Internal.Types
142 ( NodeId(..)
143 , ProcessId(..)
144 , LocalNode(..)
145 , LocalProcess(..)
146 , Process(..)
147 , Message(..)
148 , MonitorRef(..)
149 , SpawnRef(..)
150 , NCMsg(..)
151 , ProcessSignal(..)
152 , monitorCounter
153 , spawnCounter
154 , SendPort(..)
155 , ReceivePort(..)
156 , channelCounter
157 , typedChannelWithId
158 , TypedChannel(..)
159 , SendPortId(..)
160 , Identifier(..)
161 , ProcessExitException(..)
162 , DidUnmonitor(..)
163 , DidUnlinkProcess(..)
164 , DidUnlinkNode(..)
165 , DidUnlinkPort(..)
166 , WhereIsReply(..)
167 , RegisterReply(..)
168 , ProcessRegistrationException(..)
169 , ProcessInfo(..)
170 , ProcessInfoNone(..)
171 , createMessage
172 , createUnencodedMessage
173 , runLocalProcess
174 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)
175 , LocalProcessState
176 , LocalSendPortId
177 , messageToPayload
178 )
179import Control.Distributed.Process.Internal.Messaging
180 ( sendMessage
181 , sendBinary
182 , sendPayload
183 , disconnect
184 )
185import Control.Distributed.Process.Internal.Trace.Types
186 ( traceEvent
187 , TraceEvent(..)
188 )
189import Control.Distributed.Process.Internal.WeakTQueue
190 ( newTQueueIO
191 , readTQueue
192 , mkWeakTQueue
193 )
195import Unsafe.Coerce
197--------------------------------------------------------------------------------
198-- Basic messaging --
199--------------------------------------------------------------------------------
201-- | Send a message
202send :: Serializable a => ProcessId -> a -> Process ()
203-- This requires a lookup on every send. If we want to avoid that we need to
204-- modify serializable to allow for stateful (IO) deserialization.
205send them msg = do
206 proc <- ask
207 let us = processId proc
208 node = processNode proc
209 nodeId = localNodeId node
210 destNode = (processNodeId them) in do
211 case destNode == nodeId of
212 True -> sendLocal them msg
213 False -> liftIO $ sendMessage (processNode proc)
214 (ProcessIdentifier (processId proc))
215 (ProcessIdentifier them)
216 NoImplicitReconnect
217 msg
218 -- We do not fire the trace event until after the sending is complete;
219 -- In the remote case, 'sendMessage' can block in the networking stack.
220 liftIO $ traceEvent (localTracer node)
221 (TraceEvSent them us (createUnencodedMessage msg))
223-- | Wait for a message of a specific type
224expect :: forall a. Serializable a => Process a
225expect = receiveWait [match return]
227--------------------------------------------------------------------------------
228-- Channels --
229--------------------------------------------------------------------------------
231-- | Create a new typed channel
232newChan :: Serializable a => Process (SendPort a, ReceivePort a)
233newChan = do
234 proc <- ask
235 liftIO . modifyMVar (processState proc) $ \st -> do
236 let lcid = st ^. channelCounter
237 let cid = SendPortId { sendPortProcessId = processId proc
238 , sendPortLocalId = lcid
239 }
240 let sport = SendPort cid
241 chan <- liftIO newTQueueIO
242 chan' <- mkWeakTQueue chan $ finalizer (processState proc) lcid
243 let rport = ReceivePort $ readTQueue chan
244 let tch = TypedChannel chan'
245 return ( (channelCounter ^: (+ 1))
246 . (typedChannelWithId lcid ^= Just tch)
247 $ st
248 , (sport, rport)
249 )
250 where
251 finalizer :: StrictMVar LocalProcessState -> LocalSendPortId -> IO ()
252 finalizer st lcid = modifyMVar_ st $
253 return . (typedChannelWithId lcid ^= Nothing)
255-- | Send a message on a typed channel
256sendChan :: Serializable a => SendPort a -> a -> Process ()
257sendChan (SendPort cid) msg = do
258 proc <- ask
259 let node = localNodeId (processNode proc)
260 destNode = processNodeId (sendPortProcessId cid) in do
261 case destNode == node of
262 True -> sendChanLocal cid msg
263 False -> do
264 liftIO $ sendBinary (processNode proc)
265 (ProcessIdentifier (processId proc))
266 (SendPortIdentifier cid)
267 NoImplicitReconnect
268 msg
270-- | Wait for a message on a typed channel
271receiveChan :: Serializable a => ReceivePort a -> Process a
272receiveChan = liftIO . atomically . receiveSTM
274-- | Like 'receiveChan' but with a timeout. If the timeout is 0, do a
275-- non-blocking check for a message.
276receiveChanTimeout :: Serializable a => Int -> ReceivePort a -> Process (Maybe a)
277receiveChanTimeout 0 ch = liftIO . atomically $
278 (Just <$> receiveSTM ch) `orElse` return Nothing
279receiveChanTimeout n ch = liftIO . timeout n . atomically $
280 receiveSTM ch
282-- | Merge a list of typed channels.
283--
284-- The result port is left-biased: if there are messages available on more
285-- than one port, the first available message is returned.
286mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
287mergePortsBiased = return . ReceivePort. foldr1 orElse . map receiveSTM
289-- | Like 'mergePortsBiased', but with a round-robin scheduler (rather than
290-- left-biased)
291mergePortsRR :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
292mergePortsRR = \ps -> do
293 psVar <- liftIO . atomically $ newTVar (map receiveSTM ps)
294 return $ ReceivePort (rr psVar)
295 where
296 rotate :: [a] -> [a]
297 rotate [] = []
298 rotate (x:xs) = xs ++ [x]
300 rr :: TVar [STM a] -> STM a
301 rr psVar = do
302 ps <- readTVar psVar
303 a <- foldr1 orElse ps
304 writeTVar psVar (rotate ps)
305 return a
307--------------------------------------------------------------------------------
308-- Advanced messaging --
309--------------------------------------------------------------------------------
311-- | Opaque type used in 'receiveWait' and 'receiveTimeout'
312newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
314-- | Test the matches in order against each message in the queue
315receiveWait :: [Match b] -> Process b
316receiveWait ms = do
317 queue <- processQueue <$> ask
318 Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
319 proc
321-- | Like 'receiveWait' but with a timeout.
322--
323-- If the timeout is zero do a non-blocking check for matching messages. A
324-- non-zero timeout is applied only when waiting for incoming messages (that is,
325-- /after/ we have checked the messages that are already in the mailbox).
326receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
327receiveTimeout t ms = do
328 queue <- processQueue <$> ask
329 let blockSpec = if t == 0 then NonBlocking else Timeout t
330 mProc <- liftIO $ dequeue queue blockSpec (map unMatch ms)
331 case mProc of
332 Nothing -> return Nothing
333 Just proc -> Just <$> proc
335matchChan :: ReceivePort a -> (a -> Process b) -> Match b
336matchChan p fn = Match $ MatchChan (fmap fn (receiveSTM p))
338-- | Match against any message of the right type
339match :: forall a b. Serializable a => (a -> Process b) -> Match b
340match = matchIf (const True)
342-- | Match against any message of the right type that satisfies a predicate
343matchIf :: forall a b. Serializable a => (a -> Bool) -> (a -> Process b) -> Match b
344matchIf c p = Match $ MatchMsg $ \msg ->
345 case messageFingerprint msg == fingerprint (undefined :: a) of
346 False -> Nothing
347 True -> case msg of
348 (UnencodedMessage _ m) ->
349 let m' = unsafeCoerce m :: a in
350 case (c m') of
351 True -> Just (p m')
352 False -> Nothing
353 (EncodedMessage _ _) ->
354 if (c decoded) then Just (p decoded) else Nothing
355 where
356 decoded :: a
357 -- Make sure the value is fully decoded so that we don't hang to
358 -- bytestrings when the process calling 'matchIf' doesn't process
359 -- the values immediately
360 !decoded = decode (messageEncoding msg)
362-- | Match against any message, regardless of the underlying (contained) type
363matchMessage :: (Message -> Process Message) -> Match Message
364matchMessage p = Match $ MatchMsg $ \msg -> Just (p msg)
366-- | Match against any message (regardless of underlying type) that satisfies a predicate
367matchMessageIf :: (Message -> Bool) -> (Message -> Process Message) -> Match Message
368matchMessageIf c p = Match $ MatchMsg $ \msg ->
369 case (c msg) of
370 True -> Just (p msg)
371 False -> Nothing
373-- | Forward a raw 'Message' to the given 'ProcessId'.
374forward :: Message -> ProcessId -> Process ()
375forward msg them = do
376 proc <- ask
377 let node = processNode proc
378 us = processId proc
379 nid = localNodeId node
380 destNode = (processNodeId them) in do
381 case destNode == nid of
382 True -> sendCtrlMsg Nothing (LocalSend them msg)
383 False -> liftIO $ sendPayload (processNode proc)
384 (ProcessIdentifier (processId proc))
385 (ProcessIdentifier them)
386 NoImplicitReconnect
387 (messageToPayload msg)
388 -- We do not fire the trace event until after the sending is complete;
389 -- In the remote case, 'sendMessage' can block in the networking stack.
390 liftIO $ traceEvent (localTracer node)
391 (TraceEvSent them us (createUnencodedMessage msg))
394-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
395-- 'Serializable' - like the datum they contain - but also note, deserialising
396-- a 'Message' will yield a 'Message', not the type within it! To obtain the
397-- wrapped datum, use 'unwrapMessage' or 'handleMessage' with a specific type.
398--
399-- @
400-- do
401-- self <- getSelfPid
402-- send self (wrapMessage "blah")
403-- Nothing <- expectTimeout 1000000 :: Process (Maybe String)
404-- (Just m) <- expectTimeout 1000000 :: Process (Maybe Message)
405-- (Just "blah") <- unwrapMessage m :: Process (Maybe String)
406-- @
407--
408wrapMessage :: Serializable a => a -> Message
409wrapMessage = createUnencodedMessage -- see [note Serializable UnencodedMessage]
411-- [note Serializable UnencodedMessage]
412-- if we attempt to serialise an UnencodedMessage, it will be converted to an
413-- EncodedMessage first, so we're ok here. See 'messageToPayload' in
414-- Primitives.hs for the gory details
416-- | Attempt to unwrap a raw 'Message'.
417-- If the type of the decoded message payload matches the expected type, the
418-- value will be returned with @Just@, otherwise @Nothing@ indicates the types
419-- do not match.
420--
421-- This expression, for example, will evaluate to @Nothing@
422-- > unwrapMessage (wrapMessage "foobar") :: Process (Maybe Int)
423--
424-- Whereas this expression, will yield @Just "foo"@
425-- > unwrapMessage (wrapMessage "foo") :: Process (Maybe String)
426--
427unwrapMessage :: forall a. Serializable a => Message -> Process (Maybe a)
428unwrapMessage msg =
429 case messageFingerprint msg == fingerprint (undefined :: a) of
430 False -> return Nothing
431 True -> case msg of
432 (UnencodedMessage _ m) ->
433 let m' = unsafeCoerce m :: a
434 in return (Just m')
435 (EncodedMessage _ _) ->
436 return (Just (decoded))
437 where
438 decoded :: a -- note [decoding]
439 !decoded = decode (messageEncoding msg)
441-- | Attempt to handle a raw 'Message'.
442-- If the type of the message matches the type of the first argument to
443-- the supplied expression, then the message will be decoded and the expression
444-- evaluated against its value. If this runtime type checking fails however,
445-- @Nothing@ will be returned to indicate the fact. If the check succeeds and
446-- evaluation proceeds, the resulting value with be wrapped with @Just@.
447--
448-- Intended for use in `catchesExit` and `matchAny` primitives.
449--
450handleMessage :: forall a b. (Serializable a)
451 => Message -> (a -> Process b) -> Process (Maybe b)
452handleMessage msg proc = do
453 case messageFingerprint msg == fingerprint (undefined :: a) of
454 False -> return Nothing
455 True -> case msg of
456 (UnencodedMessage _ m) ->
457 let m' = unsafeCoerce m :: a -- note [decoding]
458 in do { r <- proc m'; return (Just r) }
459 (EncodedMessage _ _) ->
460 do { r <- proc (decoded :: a); return (Just r) }
461 where
462 decoded :: a -- note [decoding]
463 !decoded = decode (messageEncoding msg)
465-- | Match against an arbitrary message. 'matchAny' removes the first available
466-- message from the process mailbox. To handle arbitrary /raw/ messages once
467-- removed from the mailbox, see 'handleMessage' and 'unwrapMessage'.
468--
469matchAny :: forall b. (Message -> Process b) -> Match b
470matchAny p = Match $ MatchMsg $ \msg -> Just (p msg)
472-- | Match against an arbitrary message. Intended for use with 'handleMessage'
473-- and 'unwrapMessage', this function /only/ removes a message from the process
474-- mailbox, /if/ the supplied condition matches. The success (or failure) of
475-- runtime type checks deferred to @handleMessage@ and friends is irrelevant
476-- here, i.e., if the condition evaluates to @True@ then the message will
477-- be removed from the process mailbox and decoded, but that does /not/
478-- guarantee that an expression passed to @handleMessage@ will pass the
479-- runtime type checks and therefore be evaluated.
480--
481matchAnyIf :: forall a b. (Serializable a)
482 => (a -> Bool)
483 -> (Message -> Process b)
484 -> Match b
485matchAnyIf c p = Match $ MatchMsg $ \msg ->
486 case messageFingerprint msg == fingerprint (undefined :: a) of
487 True | check -> Just (p msg)
488 where
489 check :: Bool
490 !check =
491 case msg of
492 (EncodedMessage _ _) -> c decoded
493 (UnencodedMessage _ m') -> c (unsafeCoerce m')
495 decoded :: a -- note [decoding]
496 !decoded = decode (messageEncoding msg)
497 _ -> Nothing
499{- note [decoding]
500For an EncodedMessage, we need to ensure the value is fully decoded so that
501we don't hang to bytestrings if the calling process doesn't evaluate
502immediately. For UnencodedMessage we know (because the fingerprint comparison
503succeeds) that unsafeCoerce will not fail.
504-}
506-- | Remove any message from the queue
507matchUnknown :: Process b -> Match b
508matchUnknown p = Match $ MatchMsg (const (Just p))
510-- | Receives messages and forwards them to @pid@ if @p msg == True@.
511delegate :: ProcessId -> (Message -> Bool) -> Process ()
512delegate pid p = do
513 receiveWait [
514 matchAny (\m -> case (p m) of
515 True -> forward m pid
516 False -> return ())
517 ]
518 delegate pid p
520-- | A straight relay that simply forwards all messages to the supplied pid.
521relay :: ProcessId -> Process ()
522relay !pid = receiveWait [ matchAny (\m -> forward m pid) ] >> relay pid
524-- | Proxies @pid@ and forwards messages whenever @proc@ evaluates to @True@.
525-- Unlike 'delegate' the predicate @proc@ runs in the 'Process' monad, allowing
526-- for richer proxy behaviour. If @proc@ returns @False@ or the /runtime type check/
527-- fails, no action is taken and the proxy process will continue running.
528proxy :: Serializable a => ProcessId -> (a -> Process Bool) -> Process ()
529proxy pid proc = do
530 receiveWait [
531 matchAny (\m -> do
532 next <- handleMessage m proc
533 case next of
534 Just True -> forward m pid
535 Just False -> return () -- explicitly ignored
536 Nothing -> return ()) -- un-routable
537 ]
538 proxy pid proc
540--------------------------------------------------------------------------------
541-- Process management --
542--------------------------------------------------------------------------------
544-- | Thrown by 'terminate'
545data ProcessTerminationException = ProcessTerminationException
546 deriving (Show, Typeable)
548instance Exception ProcessTerminationException
550-- | Terminate immediately (throws a ProcessTerminationException)
551terminate :: Process a
552terminate = liftIO $ throwIO ProcessTerminationException
554-- [Issue #110]
555-- | Die immediately - throws a 'ProcessExitException' with the given @reason@.
556die :: Serializable a => a -> Process b
557die reason = do
558 pid <- getSelfPid
559 liftIO $ throwIO (ProcessExitException pid (createMessage reason))
561-- | Forceful request to kill a process. Where 'exit' provides an exception
562-- that can be caught and handled, 'kill' throws an unexposed exception type
563-- which cannot be handled explicitly (by type).
564kill :: ProcessId -> String -> Process ()
565-- NOTE: We send the message to our local node controller, which will then
566-- forward it to a remote node controller (if applicable). Sending it directly
567-- to a remote node controller means that that the message may overtake a
568-- 'monitor' or 'link' request.
569kill them reason = sendCtrlMsg Nothing (Kill them reason)
571-- | Graceful request to exit a process. Throws 'ProcessExitException' with the
572-- supplied @reason@ encoded as a message. Any /exit signal/ raised in this
573-- manner can be handled using the 'catchExit' family of functions.
574exit :: Serializable a => ProcessId -> a -> Process ()
575-- NOTE: We send the message to our local node controller, which will then
576-- forward it to a remote node controller (if applicable). Sending it directly
577-- to a remote node controller means that that the message may overtake a
578-- 'monitor' or 'link' request.
579exit them reason = sendCtrlMsg Nothing (Exit them (createMessage reason))
581-- | Catches 'ProcessExitException'. The handler will not be applied unless its
582-- type matches the encoded data stored in the exception (see the /reason/
583-- argument given to the 'exit' primitive). If the handler cannot be applied,
584-- the exception will be re-thrown.
585--
586-- To handle 'ProcessExitException' without regard for /reason/, see 'catch'.
587-- To handle multiple /reasons/ of differing types, see 'catchesExit'.
588catchExit :: forall a b . (Show a, Serializable a)
589 => Process b
590 -> (ProcessId -> a -> Process b)
591 -> Process b
592catchExit act exitHandler = catch act handleExit
593 where
594 handleExit ex@(ProcessExitException from msg) =
595 if messageFingerprint msg == fingerprint (undefined :: a)
596 then exitHandler from decoded
597 else liftIO $ throwIO ex
598 where
599 decoded :: a
600 -- Make sure the value is fully decoded so that we don't hang to
601 -- bytestrings if the caller doesn't use the value immediately
602 !decoded = decode (messageEncoding msg)
604-- | Lift 'Control.Exception.catches' (almost).
605--
606-- As 'ProcessExitException' stores the exit @reason@ as a typed, encoded
607-- message, a handler must accept an input of the expected type. In order to
608-- handle a list of potentially different handlers (and therefore input types),
609-- a handler passed to 'catchesExit' must accept 'AbstractMessage' and return
610-- @Maybe@ (i.e., @Just p@ if it handled the exit reason, otherwise @Nothing@).
611--
612-- See 'maybeHandleMessage' and 'AsbtractMessage' for more details.
613catchesExit :: Process b
614 -> [(ProcessId -> Message -> (Process (Maybe b)))]
615 -> Process b
616catchesExit act handlers = catch act ((flip handleExit) handlers)
617 where
618 handleExit :: ProcessExitException
619 -> [(ProcessId -> Message -> Process (Maybe b))]
620 -> Process b
621 handleExit ex [] = liftIO $ throwIO ex
622 handleExit ex@(ProcessExitException from msg) (h:hs) = do
623 r <- h from msg
624 case r of
625 Nothing -> handleExit ex hs
626 Just p -> return p
628-- | Our own process ID
629getSelfPid :: Process ProcessId
630getSelfPid = processId <$> ask
632-- | Get the node ID of our local node
633getSelfNode :: Process NodeId
634getSelfNode = localNodeId . processNode <$> ask
636-- | Get information about the specified process
637getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)
638getProcessInfo pid =
639 let them = processNodeId pid in do
640 us <- getSelfNode
641 dest <- mkNode them us
642 sendCtrlMsg dest $ GetInfo pid
643 receiveWait [
644 match (\(p :: ProcessInfo) -> return $ Just p)
645 , match (\(_ :: ProcessInfoNone) -> return Nothing)
646 ]
647 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)
648 mkNode them us = case them == us of
649 True -> return Nothing
650 _ -> return $ Just them
652--------------------------------------------------------------------------------
653-- Monitoring and linking --
654--------------------------------------------------------------------------------
656-- | Link to a remote process (asynchronous)
657--
658-- When process A links to process B (that is, process A calls
659-- @link pidB@) then an asynchronous exception will be thrown to process A
660-- when process B terminates (normally or abnormally), or when process A gets
661-- disconnected from process B. Although it is /technically/ possible to catch
662-- these exceptions, chances are if you find yourself trying to do so you should
663-- probably be using 'monitor' rather than 'link'. In particular, code such as
664--
665-- > link pidB -- Link to process B
666-- > expect -- Wait for a message from process B
667-- > unlink pidB -- Unlink again
668--
669-- doesn't quite do what one might expect: if process B sends a message to
670-- process A, and /subsequently terminates/, then process A might or might not
671-- be terminated too, depending on whether the exception is thrown before or
672-- after the 'unlink' (i.e., this code has a race condition).
673--
674-- Linking is all-or-nothing: A is either linked to B, or it's not. A second
675-- call to 'link' has no effect.
676--
677-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
678-- Linking makes no distinction between normal and abnormal termination of
679-- the remote process.
680link :: ProcessId -> Process ()
681link = sendCtrlMsg Nothing . Link . ProcessIdentifier
683-- | Monitor another process (asynchronous)
684--
685-- When process A monitors process B (that is, process A calls
686-- @monitor pidB@) then process A will receive a 'ProcessMonitorNotification'
687-- when process B terminates (normally or abnormally), or when process A gets
688-- disconnected from process B. You receive this message like any other (using
689-- 'expect'); the notification includes a reason ('DiedNormal', 'DiedException',
690-- 'DiedDisconnect', etc.).
691--
692-- Every call to 'monitor' returns a new monitor reference 'MonitorRef'; if
693-- multiple monitors are set up, multiple notifications will be delivered
694-- and monitors can be disabled individually using 'unmonitor'.
695monitor :: ProcessId -> Process MonitorRef
696monitor = monitor' . ProcessIdentifier
698-- | Establishes temporary monitoring of another process.
699--
700-- @withMonitor pid code@ sets up monitoring of @pid@ for the duration
701-- of @code@. Note: although monitoring is no longer active when
702-- @withMonitor@ returns, there might still be unreceived monitor
703-- messages in the queue.
704--
705withMonitor :: ProcessId -> Process a -> Process a
706withMonitor pid code = bracket (monitor pid) unmonitor (\_ -> code)
707 -- unmonitor blocks waiting for the response, so there's a possibility
708 -- that an exception might interrupt withMonitor before the unmonitor
709 -- has completed. I think that's better than making the unmonitor
710 -- uninterruptible.
712-- | Remove a link
713--
714-- This is synchronous in the sense that once it returns you are guaranteed
715-- that no exception will be raised if the remote process dies. However, it is
716-- asynchronous in the sense that we do not wait for a response from the remote
717-- node.
718unlink :: ProcessId -> Process ()
719unlink pid = do
720 unlinkAsync pid
721 receiveWait [ matchIf (\(DidUnlinkProcess pid') -> pid' == pid)
722 (\_ -> return ())
723 ]
725-- | Remove a node link
726--
727-- This has the same synchronous/asynchronous nature as 'unlink'.
728unlinkNode :: NodeId -> Process ()
729unlinkNode nid = do
730 unlinkNodeAsync nid
731 receiveWait [ matchIf (\(DidUnlinkNode nid') -> nid' == nid)
732 (\_ -> return ())
733 ]
735-- | Remove a channel (send port) link
736--
737-- This has the same synchronous/asynchronous nature as 'unlink'.
738unlinkPort :: SendPort a -> Process ()
739unlinkPort sport = do
740 unlinkPortAsync sport
741 receiveWait [ matchIf (\(DidUnlinkPort cid) -> cid == sendPortId sport)
742 (\_ -> return ())
743 ]
745-- | Remove a monitor
746--
747-- This has the same synchronous/asynchronous nature as 'unlink'.
748unmonitor :: MonitorRef -> Process ()
749unmonitor ref = do
750 unmonitorAsync ref
751 receiveWait [ matchIf (\(DidUnmonitor ref') -> ref' == ref)
752 (\_ -> return ())
753 ]
755--------------------------------------------------------------------------------
756-- Exception handling --
757--------------------------------------------------------------------------------
759-- | Lift 'Control.Exception.catch'
760catch :: Exception e => Process a -> (e -> Process a) -> Process a
761catch p h = do
762 lproc <- ask
763 liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)
765-- | Lift 'Control.Exception.try'
766try :: Exception e => Process a -> Process (Either e a)
767try p = do
768 lproc <- ask
769 liftIO $ Ex.try (runLocalProcess lproc p)
771-- | Lift 'Control.Exception.mask'
772mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b
773mask p = do
774 lproc <- ask
775 liftIO $ Ex.mask $ \restore ->
776 runLocalProcess lproc (p (liftRestore restore))
777 where
778 liftRestore :: (forall a. IO a -> IO a)
779 -> (forall a. Process a -> Process a)
780 liftRestore restoreIO = \p2 -> do
781 ourLocalProc <- ask
782 liftIO $ restoreIO $ runLocalProcess ourLocalProc p2
784-- | Lift 'Control.Exception.onException'
785onException :: Process a -> Process b -> Process a
786onException p what = p `catch` \e -> do _ <- what
787 liftIO $ throwIO (e :: SomeException)
789-- | Lift 'Control.Exception.bracket'
790bracket :: Process a -> (a -> Process b) -> (a -> Process c) -> Process c
791bracket before after thing =
792 mask $ \restore -> do
793 a <- before
794 r <- restore (thing a) `onException` after a
795 _ <- after a
796 return r
798-- | Lift 'Control.Exception.bracket_'
799bracket_ :: Process a -> Process b -> Process c -> Process c
800bracket_ before after thing = bracket before (const after) (const thing)
802-- | Lift 'Control.Exception.finally'
803finally :: Process a -> Process b -> Process a
804finally a sequel = bracket_ (return ()) sequel a
806-- | You need this when using 'catches'
807data Handler a = forall e . Exception e => Handler (e -> Process a)
809instance Functor Handler where
810 fmap f (Handler h) = Handler (fmap f . h)
812-- | Lift 'Control.Exception.catches'
813catches :: Process a -> [Handler a] -> Process a
814catches proc handlers = proc `catch` catchesHandler handlers
816catchesHandler :: [Handler a] -> SomeException -> Process a
817catchesHandler handlers e = foldr tryHandler (throw e) handlers
818 where tryHandler (Handler handler) res
819 = case fromException e of
820 Just e' -> handler e'
821 Nothing -> res
823--------------------------------------------------------------------------------
824-- Auxiliary API --
825--------------------------------------------------------------------------------
827-- | Like 'expect' but with a timeout
828expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)
829expectTimeout n = receiveTimeout n [match return]
831-- | Asynchronous version of 'spawn'
832--
833-- ('spawn' is defined in terms of 'spawnAsync' and 'expect')
834spawnAsync :: NodeId -> Closure (Process ()) -> Process SpawnRef
835spawnAsync nid proc = do
836 spawnRef <- getSpawnRef
837 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
838 return spawnRef
840-- | Monitor a node (asynchronous)
841monitorNode :: NodeId -> Process MonitorRef
842monitorNode =
843 monitor' . NodeIdentifier
845-- | Monitor a typed channel (asynchronous)
846monitorPort :: forall a. Serializable a => SendPort a -> Process MonitorRef
847monitorPort (SendPort cid) =
848 monitor' (SendPortIdentifier cid)
850-- | Remove a monitor (asynchronous)
851unmonitorAsync :: MonitorRef -> Process ()
852unmonitorAsync =
853 sendCtrlMsg Nothing . Unmonitor
855-- | Link to a node (asynchronous)
856linkNode :: NodeId -> Process ()
857linkNode = link' . NodeIdentifier
859-- | Link to a channel (asynchronous)
860linkPort :: SendPort a -> Process ()
861linkPort (SendPort cid) =
862 link' (SendPortIdentifier cid)
864-- | Remove a process link (asynchronous)
865unlinkAsync :: ProcessId -> Process ()
866unlinkAsync =
867 sendCtrlMsg Nothing . Unlink . ProcessIdentifier
869-- | Remove a node link (asynchronous)
870unlinkNodeAsync :: NodeId -> Process ()
871unlinkNodeAsync =
872 sendCtrlMsg Nothing . Unlink . NodeIdentifier
874-- | Remove a channel (send port) link (asynchronous)
875unlinkPortAsync :: SendPort a -> Process ()
876unlinkPortAsync (SendPort cid) =
877 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid
879--------------------------------------------------------------------------------
880-- Logging --
881--------------------------------------------------------------------------------
883-- | Log a string
884--
885-- @say message@ sends a message (time, pid of the current process, message)
886-- to the process registered as 'logger'. By default, this process simply
887-- sends the string to 'stderr'. Individual Cloud Haskell backends might
888-- replace this with a different logger process, however.
889say :: String -> Process ()
890say string = do
891 now <- liftIO getCurrentTime
892 us <- getSelfPid
893 nsend "logger" (formatTime defaultTimeLocale "%c" now, us, string)
895--------------------------------------------------------------------------------
896-- Registry --
897--------------------------------------------------------------------------------
899-- | Register a process with the local registry (asynchronous).
900-- This version will wait until a response is gotten from the
901-- management process. The name must not already be registered.
902-- The process need not be on this node.
903-- A bad registration will result in a 'ProcessRegistrationException'
904--
905-- The process to be registered does not have to be local itself.
906register :: String -> ProcessId -> Process ()
907register = registerImpl False
909-- | Like 'register', but will replace an existing registration.
910-- The name must already be registered.
911reregister :: String -> ProcessId -> Process ()
912reregister = registerImpl True
914registerImpl :: Bool -> String -> ProcessId -> Process ()
915registerImpl force label pid = do
916 mynid <- getSelfNode
917 sendCtrlMsg Nothing (Register label mynid (Just pid) force)
918 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
919 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
920 ]
922-- | Register a process with a remote registry (asynchronous).
923--
924-- The process to be registered does not have to live on the same remote node.
925-- Reply wil come in the form of a 'RegisterReply' message
926--
927-- See comments in 'whereisRemoteAsync'
928registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
929registerRemoteAsync nid label pid =
930 sendCtrlMsg (Just nid) (Register label nid (Just pid) False)
932reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
933reregisterRemoteAsync nid label pid =
934 sendCtrlMsg (Just nid) (Register label nid (Just pid) True)
936-- | Remove a process from the local registry (asynchronous).
937-- This version will wait until a response is gotten from the
938-- management process. The name must already be registered.
939unregister :: String -> Process ()
940unregister label = do
941 mynid <- getSelfNode
942 sendCtrlMsg Nothing (Register label mynid Nothing False)
943 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
944 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
945 ]
947-- | Deal with the result from an attempted registration or unregistration
948-- by throwing an exception if necessary
949handleRegistrationReply :: String -> Bool -> Process ()
950handleRegistrationReply label ok =
951 when (not ok) $
952 liftIO $ throwIO $ ProcessRegistrationException label
954-- | Remove a process from a remote registry (asynchronous).
955--
956-- Reply wil come in the form of a 'RegisterReply' message
957--
958-- See comments in 'whereisRemoteAsync'
959unregisterRemoteAsync :: NodeId -> String -> Process ()
960unregisterRemoteAsync nid label =
961 sendCtrlMsg (Just nid) (Register label nid Nothing False)
963-- | Query the local process registry
964whereis :: String -> Process (Maybe ProcessId)
965whereis label = do
966 sendCtrlMsg Nothing (WhereIs label)
967 receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')
968 (\(WhereIsReply _ mPid) -> return mPid)
969 ]
971-- | Query a remote process registry (asynchronous)
972--
973-- Reply will come in the form of a 'WhereIsReply' message.
974--
975-- There is currently no synchronous version of 'whereisRemoteAsync': if
976-- you implement one yourself, be sure to take into account that the remote
977-- node might die or get disconnect before it can respond (i.e. you should
978-- use 'monitorNode' and take appropriate action when you receive a
979-- 'NodeMonitorNotification').
980whereisRemoteAsync :: NodeId -> String -> Process ()
981whereisRemoteAsync nid label =
982 sendCtrlMsg (Just nid) (WhereIs label)
984-- | Named send to a process in the local registry (asynchronous)
985nsend :: Serializable a => String -> a -> Process ()
986nsend label msg =
987 sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
989-- | Named send to a process in a remote registry (asynchronous)
990nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
991nsendRemote nid label msg =
992 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
994--------------------------------------------------------------------------------
995-- Closures --
996--------------------------------------------------------------------------------
998-- | Resolve a static value
999unStatic :: Typeable a => Static a -> Process a
1000unStatic static = do
1001 rtable <- remoteTable . processNode <$> ask
1002 case Static.unstatic rtable static of
1003 Left err -> fail $ "Could not resolve static value: " ++ err
1004 Right x -> return x
1006-- | Resolve a closure
1007unClosure :: Typeable a => Closure a -> Process a
1008unClosure closure = do
1009 rtable <- remoteTable . processNode <$> ask
1010 case Static.unclosure rtable closure of
1011 Left err -> fail $ "Could not resolve closure: " ++ err
1012 Right x -> return x
1014--------------------------------------------------------------------------------
1015-- Reconnecting --
1016--------------------------------------------------------------------------------
1018-- | Cloud Haskell provides the illusion of connection-less, reliable, ordered
1019-- message passing. However, when network connections get disrupted this
1020-- illusion cannot always be maintained. Once a network connection breaks (even
1021-- temporarily) no further communication on that connection will be possible.
1022-- For example, if process A sends a message to process B, and A is then
1023-- notified (by monitor notification) that it got disconnected from B, A will
1024-- not be able to send any further messages to B, /unless/ A explicitly
1025-- indicates that it is acceptable to attempt to reconnect to B using the
1026-- Cloud Haskell 'reconnect' primitive.
1028-- Importantly, when A calls 'reconnect' it acknowledges that some messages to
1029-- B might have been lost. For instance, if A sends messages m1 and m2 to B,
1030-- then receives a monitor notification that its connection to B has been lost,
1031-- calls 'reconnect' and then sends m3, it is possible that B will receive m1
1032-- and m3 but not m2.
1034-- Note that 'reconnect' does not mean /reconnect now/ but rather /it is okay
1035-- to attempt to reconnect on the next send/. In particular, if no further
1036-- communication attempts are made to B then A can use reconnect to clean up
1037-- its connection to B.
1038reconnect :: ProcessId -> Process ()
1039reconnect them = do
1040 us <- getSelfPid
1041 node <- processNode <$> ask
1042 liftIO $ disconnect node (ProcessIdentifier us) (ProcessIdentifier them)
1044-- | Reconnect to a sendport. See 'reconnect' for more information.
1045reconnectPort :: SendPort a -> Process ()
1046reconnectPort them = do
1047 us <- getSelfPid
1048 node <- processNode <$> ask
1049 liftIO $ disconnect node (ProcessIdentifier us) (SendPortIdentifier (sendPortId them))
1051--------------------------------------------------------------------------------
1052-- Auxiliary functions --
1053--------------------------------------------------------------------------------
1055sendLocal :: (Serializable a) => ProcessId -> a -> Process ()
1056sendLocal to msg =
1057 sendCtrlMsg Nothing $ LocalSend to (createUnencodedMessage msg)
1059sendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
1060sendChanLocal spId msg =
1061 -- we *must* fully serialize/encode the message here, because
1062 -- attempting to use `unsafeCoerce' in the node controller
1063 -- won't work since we know nothing about the required type
1064 sendCtrlMsg Nothing $ LocalPortSend spId (createUnencodedMessage msg)
1066getMonitorRefFor :: Identifier -> Process MonitorRef
1067getMonitorRefFor ident = do
1068 proc <- ask
1069 liftIO $ modifyMVar (processState proc) $ \st -> do
1070 let counter = st ^. monitorCounter
1071 return ( monitorCounter ^: (+ 1) $ st
1072 , MonitorRef ident counter
1075getSpawnRef :: Process SpawnRef
1076getSpawnRef = do
1077 proc <- ask
1078 liftIO $ modifyMVar (processState proc) $ \st -> do
1079 let counter = st ^. spawnCounter
1080 return ( spawnCounter ^: (+ 1) $ st
1081 , SpawnRef counter
1084-- | Monitor a process/node/channel
1085monitor' :: Identifier -> Process MonitorRef
1086monitor' ident = do
1087 monitorRef <- getMonitorRefFor ident
1088 sendCtrlMsg Nothing $ Monitor monitorRef
1089 return monitorRef
1091-- | Link to a process/node/channel
1092link' :: Identifier -> Process ()
1093link' = sendCtrlMsg Nothing . Link
1095-- Send a control message
1096sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node
1097 -> ProcessSignal -- ^ Message to send
1098 -> Process ()
1099sendCtrlMsg mNid signal = do
1100 proc <- ask
1101 let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc)
1102 , ctrlMsgSignal = signal
1104 case mNid of
1105 Nothing -> do
1106 liftIO $ writeChan (localCtrlChan (processNode proc)) msg
1107 Just nid ->
1108 liftIO $ sendBinary (processNode proc)
1109 (ProcessIdentifier (processId proc))
1110 (NodeIdentifier nid)
1111 WithImplicitReconnect
1112 msg