efb4614 · distributed-process

git.iko.soy/distributed-process

Change timeout behaviour of receiveTimeout

Apply the timeout only when checking for incoming messages, not when checking the messages already in the mailbox.

commitefb46143ce26862d7ddea822c4701e87e0564c7a authorEdsko de Vries <edsko@edsko.net> dateFri 29 Jun 2012, 09:36 (14 years ago) parentc72345c

2 files changed+14−13

compare with master

unifiedsplit

distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs+11−12 distributed-process/src/Control/Distributed/Process/Internal/Process/Primitives.hs+3−1

distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs+11−12

@@ -7,7 +7,6 @@ module Control.Distributed.Process.Internal.CQueue

77 , dequeue

88 ) where

10−import Control.Monad (join)

1110import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)

1211import Control.Concurrent.STM

1312 ( atomically

@@ -21,8 +20,7 @@ import Control.Applicative ((<$>), (<*>))

2120import Control.Exception (mask, onException)

2221import System.Timeout (timeout)

2322

24−-- TODO: The only reason for TChan rather than Chan here is to that we have

25−-- a non-blocking read

23+-- We use a TCHan rather than a Chan so that we have a non-blocking read

2624data CQueue a = CQueue (MVar [a]) -- Arrived

2725 (TChan a) -- Incoming

2826

@@ -37,23 +35,22 @@ data BlockSpec =

3735 | Blocking

3836 | Timeout Int

3937

40−-- TODO: Timeout only on checking incoming messages, not arrived

41−-- (otherwise we might easily get into an infinite loop)

38+-- | Dequeue an element

39+--

40+-- The timeout (if any) is applied only to waiting for incoming messages, not

41+-- to checking messages that have already arrived

4242dequeue :: forall a b.

4343 CQueue a -- ^ Queue

4444 -> BlockSpec -- ^ Blocking behaviour

4545 -> [a -> Maybe b] -- ^ List of matches

4646 -> IO (Maybe b) -- ^ 'Nothing' only on timeout

47−dequeue (CQueue arrived incoming) blockSpec matches =

48− case blockSpec of

49− Timeout t -> join <$> timeout t go

50− _ -> go

47+dequeue (CQueue arrived incoming) blockSpec matches = go

5148 where

5249 go :: IO (Maybe b)

5350 go = mask $ \restore -> do

5451 arr <- takeMVar arrived

55− -- We first check the arrived messages. If we timeout during this search,

56− -- we just put the MVar back (since we haven't touched the Chan yet)

52+ -- We first check the arrived messages. If we get interrupted during this

53+ -- search, we just put the MVar back (we haven't read from the Chan yet)

5754 (arr', mb) <- onException (restore (checkArrived [] arr))

5855 (putMVar arrived arr)

5956 case (mb, blockSpec) of

@@ -62,8 +59,10 @@ dequeue (CQueue arrived incoming) blockSpec matches =

6259 return (Just b)

6360 (Nothing, NonBlocking) ->

6461 checkNonBlocking arr'

65− (Nothing, _) ->

62+ (Nothing, Blocking) ->

6663 Just <$> checkBlocking arr'

64+ (Nothing, Timeout n) ->

65+ timeout n $ checkBlocking arr'

6766

6867 -- We reverse the accumulator on return only if we find a match

6968 checkArrived :: [a] -> [a] -> IO ([a], Maybe b)

distributed-process/src/Control/Distributed/Process/Internal/Process/Primitives.hs+3−1

@@ -181,7 +181,9 @@ receiveWait ms = do

181181

182182-- | Like 'receiveWait' but with a timeout.

183183--

184−-- If the timeout is zero do a non-blocking check for matching messages.

184+-- If the timeout is zero do a non-blocking check for matching messages. A

185+-- non-zero timeout is applied only when waiting for incoming messages (that is,

186+-- /after/ we have checked the messages that are already in the mailbox).

185187receiveTimeout :: Int -> [Match b] -> Process (Maybe b)

186188receiveTimeout t ms = do

187189 queue <- processQueue <$> ask