@@ -10,7 +10,7 @@ module Control.Distributed.Process.Internal.CQueue
1010 , enqueueSTM
1111 , dequeue
1212 , mkWeakCQueue
13− , queueSize
13+ , queueSize
1414 ) where
1515
1616import Prelude hiding (length, reverse)
@@ -43,6 +43,7 @@ import Control.Distributed.Process.Internal.StrictList
4343 , append
4444 )
4545import Data.Maybe (fromJust)
46+import Data.Traversable (traverse)
4647import GHC.MVar (MVar(MVar))
4748import GHC.IO (IO(IO))
4849import GHC.Prim (mkWeak#)
@@ -124,12 +125,9 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus
124125 -- Decrement counter is smth is returned from the queue,
125126 -- this is safe to use as method is called under a mask
126127 -- and there is no 'unmasked' operation inside
127− decrementJust f = do
128− mx <- f
129− case mx of
130− Just{} -> atomically $ modifyTVar' size pred
131− Nothing -> return ()
132− return mx
128+ decrementJust f =
129+ traverse (either return (\x -> decrement >> return x)) =<< f
130+ decrement = atomically $ modifyTVar' size pred
133131
134132 chunks = chunkMatches matchons
135133
@@ -144,7 +142,7 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus
144142 goCheck chunks arr'
145143
146144 waitChans ports on_block =
147− foldr orElse on_block (map (fmap Just) ports)
145+ foldr orElse on_block (map (fmap (Just . Left)) ports)
148146
149147 --
150148 -- First check the MatchChunks against the messages already in the
@@ -153,7 +151,7 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus
153151 --
154152 goCheck :: MatchChunks m a
155153 -> StrictList m -- messages to check, in this order
156− -> IO (Maybe a)
154+ -> IO (Maybe (Either a a))
157155
158156 goCheck [] old = goWait old
159157
@@ -170,7 +168,7 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus
170168 -- of passing around restore and setting up exception handlers is
171169 -- high. So just don't use expensive matchIfs!
172170 case checkArrived matches old of
173− (old', Just r) -> returnOld old' (Just r)
171+ (old', Just r) -> returnOld old' (Just (Right r))
174172 (old', Nothing) -> goCheck rest old'
175173 -- use the result list, which is now left-biased
176174
@@ -199,6 +197,7 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus
199197 -- Contents of 'arrived' from now on is (old ++ new), and
200198 -- messages that arrive are snocced onto new.
201199 --
200+ goWait :: StrictList m -> IO (Maybe (Either a a))
202201 goWait old = do
203202 r <- waitIncoming `onException` putMVar arrived old
204203 case r of
@@ -216,7 +215,7 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus
216215 --
217216 -- Right => message arrived on a channel first
218217 --
219− Right a -> returnOld old (Just a)
218+ Right a -> returnOld old (Just (Left a))
220219
221220 --
222221 -- A message arrived in the process inbox; check the MatchChunks for
@@ -225,7 +224,7 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus
225224 goCheck1 :: MatchChunks m a
226225 -> m -- single message to check
227226 -> StrictList m -- old messages we have already checked
228− -> IO (Maybe a)
227+ -> IO (Maybe (Either a a))
229228
230229 goCheck1 [] m old = goWait (Snoc old m)
231230
@@ -238,10 +237,10 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus
238237 goCheck1 (Left matches : rest) m old = do
239238 case checkMatches matches m of
240239 Nothing -> goCheck1 rest m old
241− Just p -> returnOld old (Just p)
240+ Just p -> returnOld old (Just (Right p))
242241
243242 -- a common pattern for putting back the arrived queue at the end
244− returnOld :: StrictList m -> Maybe a -> IO (Maybe a)
243+ returnOld :: StrictList m -> Maybe (Either a a) -> IO (Maybe (Either a a))
245244 returnOld old r = do putMVar arrived old; return r
246245
247246 -- as a side-effect, this left-biases the list