git.iko.soy/distributed-process
on master help sign in
Change timeout behaviour of receiveTimeout
efb4614Edsko de Vries
distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs+11−12
compare with master
unifiedsplit
@@ -7,7 +7,6 @@ module Control.Distributed.Process.Internal.CQueue
7 , dequeue
7 , dequeue
8 ) where
8 ) where
9
9
10−import Control.Monad (join)
11import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)
10import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)
12import Control.Concurrent.STM
11import Control.Concurrent.STM
13 ( atomically
12 ( atomically
@@ -21,8 +20,7 @@ import Control.Applicative ((<$>), (<*>))
21import Control.Exception (mask, onException)
20import Control.Exception (mask, onException)
22import System.Timeout (timeout)
21import System.Timeout (timeout)
23
22
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
26data CQueue a = CQueue (MVar [a]) -- Arrived
24data CQueue a = CQueue (MVar [a]) -- Arrived
27 (TChan a) -- Incoming
25 (TChan a) -- Incoming
28
26
@@ -37,23 +35,22 @@ data BlockSpec =
37 | Blocking
35 | Blocking
38 | Timeout Int
36 | Timeout Int
39
37
38+-- | Dequeue an element
39+--
40−-- TODO: Timeout only on checking incoming messages, not arrived
40+-- The timeout (if any) is applied only to waiting for incoming messages, not
41−-- (otherwise we might easily get into an infinite loop)
41+-- to checking messages that have already arrived
42dequeue :: forall a b.
42dequeue :: forall a b.
43 CQueue a -- ^ Queue
43 CQueue a -- ^ Queue
44 -> BlockSpec -- ^ Blocking behaviour
44 -> BlockSpec -- ^ Blocking behaviour
45 -> [a -> Maybe b] -- ^ List of matches
45 -> [a -> Maybe b] -- ^ List of matches
46 -> IO (Maybe b) -- ^ 'Nothing' only on timeout
46 -> IO (Maybe b) -- ^ 'Nothing' only on timeout
47−dequeue (CQueue arrived incoming) blockSpec matches =
47+dequeue (CQueue arrived incoming) blockSpec matches = go
48− case blockSpec of
49− Timeout t -> join <$> timeout t go
50− _ -> go
51 where
48 where
52 go :: IO (Maybe b)
49 go :: IO (Maybe b)
53 go = mask $ \restore -> do
50 go = mask $ \restore -> do
54 arr <- takeMVar arrived
51 arr <- takeMVar arrived
55− -- We first check the arrived messages. If we timeout during this search,
52+ -- We first check the arrived messages. If we get interrupted during this
56− -- we just put the MVar back (since we haven't touched the Chan yet)
53+ -- search, we just put the MVar back (we haven't read from the Chan yet)
57 (arr', mb) <- onException (restore (checkArrived [] arr))
54 (arr', mb) <- onException (restore (checkArrived [] arr))
58 (putMVar arrived arr)
55 (putMVar arrived arr)
59 case (mb, blockSpec) of
56 case (mb, blockSpec) of
@@ -62,8 +59,10 @@ dequeue (CQueue arrived incoming) blockSpec matches =
62 return (Just b)
59 return (Just b)
63 (Nothing, NonBlocking) ->
60 (Nothing, NonBlocking) ->
64 checkNonBlocking arr'
61 checkNonBlocking arr'
65− (Nothing, _) ->
62+ (Nothing, Blocking) ->
66 Just <$> checkBlocking arr'
63 Just <$> checkBlocking arr'
64+ (Nothing, Timeout n) ->
65+ timeout n $ checkBlocking arr'
67
66
68 -- We reverse the accumulator on return only if we find a match
67 -- We reverse the accumulator on return only if we find a match
69 checkArrived :: [a] -> [a] -> IO ([a], Maybe b)
68 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
181
181
182-- | Like 'receiveWait' but with a timeout.
182-- | Like 'receiveWait' but with a timeout.
183--
183--
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).
185receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
187receiveTimeout :: Int -> [Match b] -> Process (Maybe b)
186receiveTimeout t ms = do
188receiveTimeout t ms = do
187 queue <- processQueue <$> ask
189 queue <- processQueue <$> ask