@@ -1,8 +1,9 @@
1-- | Concurrent queue for single reader, single writer
1-- | Concurrent queue for single reader, single writer
2−{-# LANGUAGE MagicHash, UnboxedTuples #-}
2+{-# LANGUAGE MagicHash, UnboxedTuples, PatternGuards #-}
3module Control.Distributed.Process.Internal.CQueue
3module Control.Distributed.Process.Internal.CQueue
5 , BlockSpec(..)
5 , BlockSpec(..)
6 , newCQueue
7 , newCQueue
@@ -12,14 +13,17 @@ module Control.Distributed.Process.Internal.CQueue
12import Prelude hiding (length, reverse)
13import Prelude hiding (length, reverse)
13import Control.Concurrent.STM
14import Control.Concurrent.STM
14 ( atomically
15 ( atomically
16 , newTChan
19 , newTChan
17 , writeTChan
20 , writeTChan
18 , readTChan
21 , readTChan
19− , tryReadTChan
22+ , orElse
21import Control.Applicative ((<$>), (<*>))
25import Control.Applicative ((<$>), (<*>))
22−import Control.Exception (mask, onException)
26+import Control.Exception (mask_, onException)
23import System.Timeout (timeout)
27import System.Timeout (timeout)
24import Control.Distributed.Process.Internal.StrictMVar
28import Control.Distributed.Process.Internal.StrictMVar
25 ( StrictMVar(StrictMVar)
29 ( StrictMVar(StrictMVar)
@@ -29,9 +33,9 @@ import Control.Distributed.Process.Internal.StrictMVar
30import Control.Distributed.Process.Internal.StrictList
34import Control.Distributed.Process.Internal.StrictList
31 ( StrictList(..)
35 ( StrictList(..)
37+import qualified Control.Distributed.Process.Internal.StrictList as S
38+import Data.Maybe (fromJust)
35import GHC.MVar (MVar(MVar))
39import GHC.MVar (MVar(MVar))
36import GHC.IO (IO(IO))
40import GHC.IO (IO(IO))
37import GHC.Prim (mkWeak#)
41import GHC.Prim (mkWeak#)
@@ -55,71 +59,176 @@ data BlockSpec =
55 | Blocking
59 | Blocking
56 | Timeout Int
60 | Timeout Int
63+ = MatchMsg (m -> Maybe a)
66+type MatchChunks m a = [Either [m -> Maybe a] [STM a]]
68+chunkMatches :: [MatchOn m a] -> MatchChunks m a
70+chunkMatches (MatchMsg m : ms) = Left (m : chk) : chunkMatches rest
71+ where (chk, rest) = spanMatchMsg ms
72+chunkMatches (MatchChan r : ms) = Right (r : chk) : chunkMatches rest
73+ where (chk, rest) = spanMatchChan ms
75+spanMatchMsg :: [MatchOn m a] -> ([m -> Maybe a], [MatchOn m a])
76+spanMatchMsg [] = ([],[])
78+ | MatchMsg msg <- m = (msg:msgs, rest)
79+ | otherwise = ([], m:ms)
80+ where !(msgs,rest) = spanMatchMsg ms
82+spanMatchChan :: [MatchOn m a] -> ([STM a], [MatchOn m a])
83+spanMatchChan [] = ([],[])
84+spanMatchChan (m : ms)
85+ | MatchChan stm <- m = (stm:stms, rest)
86+ | otherwise = ([], m:ms)
87+ where !(stms,rest) = spanMatchChan ms
58-- | Dequeue an element
89-- | Dequeue an element
60-- The timeout (if any) is applied only to waiting for incoming messages, not
91-- The timeout (if any) is applied only to waiting for incoming messages, not
61-- to checking messages that have already arrived
92-- to checking messages that have already arrived
62−dequeue :: forall a b.
93+dequeue :: forall m a.
63− CQueue a -- ^ Queue
94+ CQueue m -- ^ Queue
64 -> BlockSpec -- ^ Blocking behaviour
95 -> BlockSpec -- ^ Blocking behaviour
65− -> [a -> Maybe b] -- ^ List of matches
96+ -> [MatchOn m a] -- ^ List of matches
66− -> IO (Maybe b) -- ^ 'Nothing' only on timeout
97+ -> IO (Maybe a) -- ^ 'Nothing' only on timeout
67−dequeue (CQueue arrived incoming) blockSpec matches = go
98+dequeue (CQueue arrived incoming) blockSpec matchons =
100+ Timeout n -> timeout n $ fmap fromJust go
103+ [Right ports] -> -- channels only, this is easy:
105+ NonBlocking -> atomically $ waitChans ports (return Nothing)
106+ _ -> atomically $ waitChans ports retry
107+ -- no onException needed
69− go :: IO (Maybe b)
110+ chunks = chunkMatches matchons
70− go = mask $ \restore -> do
111+
71− arr <- takeMVar arrived
113+ arr <- takeMVar arrived
72− -- We first check the arrived messages. If we get interrupted during this
114+ let grabNew xs = do
73− -- search, we just put the MVar back (we haven't read from the Chan yet)
115+ r <- atomically $ tryReadTChan incoming
74− (arr', mb) <- onException (restore (checkArrived Nil arr))
116+ case r of
75− (putMVar arrived arr)
117+ Nothing -> return xs
118+ Just x -> grabNew (Snoc xs x)
122+ waitChans ports on_block =
123+ foldr orElse on_block (map (fmap Just) ports)
126+ -- First check the MatchChunks against the messages already in the
127+ -- mailbox. For channel matches, we do a non-blocking check at
130+ goCheck :: MatchChunks m a
131+ -> StrictList m -- messages to check, in this order
134+ goCheck [] old = goWait old
136+ goCheck (Right ports : rest) old = do
137+ r <- atomically $ waitChans ports (return Nothing) -- does not block
139+ Just _ -> returnOld old r
140+ Nothing -> goCheck rest old
142+ goCheck (Left matches : rest) old = do
143+ -- checkArrived might in principle take arbitrary time, so
144+ -- we ought to call restore and use an exception handler. However,
145+ -- the check is usually fast (just a comparison), and the overhead
146+ -- of passing around restore and setting up exception handlers is
147+ -- high. So just don't use expensive matchIfs!
148+ case checkArrived matches old of
149+ (old', Just r) -> returnOld old' (Just r)
150+ (old', Nothing) -> goCheck rest old'
151+ -- use the result list, which is now left-biased
154+ -- Construct an STM transaction that looks at the relevant channels
155+ -- in the correct order.
157+ mkSTM :: MatchChunks m a -> STM (Either m a)
159+ mkSTM (Left _ : rest)
160+ = fmap Left (readTChan incoming) `orElse` mkSTM rest
161+ mkSTM (Right ports : rest)
162+ = foldr orElse (mkSTM rest) (map (fmap Right) ports)
164+ waitIncoming :: IO (Maybe (Either m a))
76− case (mb, blockSpec) of
165+ waitIncoming = case blockSpec of
77− (Just b, _) -> do
166+ NonBlocking -> atomically $ fmap Just stm `orElse` return Nothing
78− putMVar arrived arr'
167+ _ -> atomically $ fmap Just stm
79− return (Just b)
168+ where
80− (Nothing, NonBlocking) ->
169+ stm = mkSTM chunks
81− checkNonBlocking arr'
170+
82− (Nothing, Blocking) ->
171+ --
83− Just <$> checkBlocking arr'
172+ -- The initial pass didn't find a message, so now we go into blocking
84− (Nothing, Timeout n) ->
173+ -- mode.
85− timeout n $ checkBlocking arr'
174+ --
86−
175+ -- Contents of 'arrived' from now on is (old ++ new), and
87− -- We reverse the accumulator on return only if we find a match
176+ -- messages that arrive are snocced onto new.
88− checkArrived :: StrictList a -> StrictList a -> IO (StrictList a, Maybe b)
177+ --
89− checkArrived acc Nil = return (acc, Nothing)
178+ goWait old = do
90− checkArrived acc (Cons x xs) =
179+ r <- waitIncoming `onException` putMVar arrived old
181+ -- Nothing => non-blocking and no message
182+ Nothing -> returnOld old Nothing
185+ -- Left => message arrived in the process mailbox. We now have to
186+ -- run through the MatchChunks checking each one, because we might
187+ -- have a situation where the first chunk fails to match and the
188+ -- second chunk is a channel match and there *is* a message in the
189+ -- channel. In that case the channel wins.
191+ Left m -> goCheck1 chunks m old
193+ -- Right => message arrived on a channel first
195+ Right a -> returnOld old (Just a)
198+ -- A message arrived in the process inbox; check the MatchChunks for
201+ goCheck1 :: MatchChunks m a
202+ -> m -- single message to check
203+ -> StrictList m -- old messages we have already checked
206+ goCheck1 [] m old = goWait (Snoc old m)
208+ goCheck1 (Right ports : rest) m old = do
209+ r <- atomically $ waitChans ports (return Nothing) -- does not block
91− case check x of
210+ case r of
92− Just y -> return (reverse' acc xs, Just y)
211+ Nothing -> goCheck1 rest m old
93− Nothing -> checkArrived (Cons x acc) xs
212+ Just _ -> returnOld (Snoc old m) r
95− -- If we call checkBlocking there may or may not be a timeout
214+ goCheck1 (Left matches : rest) m old = do
96− checkBlocking :: StrictList a -> IO b
215+ case checkMatches matches m of
97− checkBlocking acc = do
216+ Nothing -> goCheck1 rest m old
98− -- readTChan is a blocking call, and hence is interruptable. If it is
217+ Just p -> returnOld old (Just p)
99− -- interrupted, we put the value of the accumulator in 'arrived'
218+
100− -- (as opposed to the original value), so that no messages get lost
219+ -- a common pattern for putting back the arrived queue at the end
101− -- (hence the low-level structure using mask rather than modifyMVar)
220+ returnOld :: StrictList m -> Maybe a -> IO (Maybe a)
102− x <- onException (atomically $ readTChan incoming)
221+ returnOld old r = do putMVar arrived old; return r
103− (putMVar arrived $ reverse acc)
222+
104− case check x of
223+ -- as a side-effect, this left-biases the list
105− Nothing -> checkBlocking (Cons x acc)
224+ checkArrived :: [m -> Maybe a] -> StrictList m -> (StrictList m, Maybe a)
106− Just y -> putMVar arrived (reverse acc) >> return y
225+ checkArrived matches list = S.foldr f (Nil, Nothing) list
108− -- checkNonBlocking is only called if there is no timeout
227+ f x (rest,r)
109− checkNonBlocking :: StrictList a -> IO (Maybe b)
228+ | Just y <- checkMatches matches x = (rest, Just y)
110− checkNonBlocking acc = do
229+ | otherwise = (Cons x rest, r)
111− -- tryReadTChan is *not* interruptible
230+
112− mx <- atomically $ tryReadTChan incoming
114− Nothing -> putMVar arrived (reverse acc) >> return Nothing
115− Just x -> case check x of
116− Nothing -> checkNonBlocking (Cons x acc)
117− Just y -> putMVar arrived (reverse acc) >> return (Just y)
119− check :: a -> Maybe b
120− check = checkMatches matches
122− checkMatches :: [a -> Maybe b] -> a -> Maybe b
231+ checkMatches :: [m -> Maybe a] -> m -> Maybe a
123 checkMatches [] _ = Nothing
232 checkMatches [] _ = Nothing
124 checkMatches (m:ms) a = case m a of Nothing -> checkMatches ms a
233 checkMatches (m:ms) a = case m a of Nothing -> checkMatches ms a
125 Just b -> Just b
234 Just b -> Just b