4e89b9d · distributed-process

git.iko.soy/distributed-process

on master help sign in

Merge pull request #190 from tweag/fixup-queue-size

4e89b9dFacundo Domínguez11 years ago

src/Control/Distributed/Process/Internal/CQueue.hs+13−14

compare with master

unifiedsplit

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

10 , enqueueSTM

11 , dequeue

12 , mkWeakCQueue

13− , queueSize

13+ , queueSize

14 ) where

16import Prelude hiding (length, reverse)

@@ -43,6 +43,7 @@ import Control.Distributed.Process.Internal.StrictList

43 , append

44 )

45import Data.Maybe (fromJust)

46+import Data.Traversable (traverse)

46import GHC.MVar (MVar(MVar))

47import GHC.MVar (MVar(MVar))

47import GHC.IO (IO(IO))

48import GHC.IO (IO(IO))

48import GHC.Prim (mkWeak#)

49import GHC.Prim (mkWeak#)

@@ -124,12 +125,9 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus

124 -- Decrement counter is smth is returned from the queue,

125 -- Decrement counter is smth is returned from the queue,

125 -- this is safe to use as method is called under a mask

126 -- this is safe to use as method is called under a mask

126 -- and there is no 'unmasked' operation inside

127 -- and there is no 'unmasked' operation inside

127− decrementJust f = do

128+ decrementJust f =

128− mx <- f

129+ traverse (either return (\x -> decrement >> return x)) =<< f

129− case mx of

130− Just{} -> atomically $ modifyTVar' size pred

130+ decrement = atomically $ modifyTVar' size pred

131− Nothing -> return ()

132− return mx

133

131

134 chunks = chunkMatches matchons

132 chunks = chunkMatches matchons

135

133

@@ -144,7 +142,7 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus

144 goCheck chunks arr'

142 goCheck chunks arr'

145

143

146 waitChans ports on_block =

144 waitChans ports on_block =

147− foldr orElse on_block (map (fmap Just) ports)

145+ foldr orElse on_block (map (fmap (Just . Left)) ports)

148

146

149 --

147 --

150 -- First check the MatchChunks against the messages already in the

148 -- First check the MatchChunks against the messages already in the

@@ -153,7 +151,7 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus

153 --

151 --

154 goCheck :: MatchChunks m a

152 goCheck :: MatchChunks m a

155 -> StrictList m -- messages to check, in this order

153 -> StrictList m -- messages to check, in this order

156− -> IO (Maybe a)

154+ -> IO (Maybe (Either a a))

157

155

158 goCheck [] old = goWait old

156 goCheck [] old = goWait old

159

157

@@ -170,7 +168,7 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus

170 -- of passing around restore and setting up exception handlers is

168 -- of passing around restore and setting up exception handlers is

171 -- high. So just don't use expensive matchIfs!

169 -- high. So just don't use expensive matchIfs!

172 case checkArrived matches old of

170 case checkArrived matches old of

173− (old', Just r) -> returnOld old' (Just r)

171+ (old', Just r) -> returnOld old' (Just (Right r))

174 (old', Nothing) -> goCheck rest old'

172 (old', Nothing) -> goCheck rest old'

175 -- use the result list, which is now left-biased

173 -- use the result list, which is now left-biased

176

174

@@ -199,6 +197,7 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus

199 -- Contents of 'arrived' from now on is (old ++ new), and

197 -- Contents of 'arrived' from now on is (old ++ new), and

200 -- messages that arrive are snocced onto new.

198 -- messages that arrive are snocced onto new.

201 --

199 --

200+ goWait :: StrictList m -> IO (Maybe (Either a a))

202 goWait old = do

201 goWait old = do

203 r <- waitIncoming `onException` putMVar arrived old

202 r <- waitIncoming `onException` putMVar arrived old

204 case r of

203 case r of

@@ -216,7 +215,7 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus

216 --

215 --

217 -- Right => message arrived on a channel first

216 -- Right => message arrived on a channel first

218 --

217 --

219− Right a -> returnOld old (Just a)

218+ Right a -> returnOld old (Just (Left a))

220

219

221 --

220 --

222 -- A message arrived in the process inbox; check the MatchChunks for

221 -- A message arrived in the process inbox; check the MatchChunks for

@@ -225,7 +224,7 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus

225 goCheck1 :: MatchChunks m a

224 goCheck1 :: MatchChunks m a

226 -> m -- single message to check

225 -> m -- single message to check

227 -> StrictList m -- old messages we have already checked

226 -> StrictList m -- old messages we have already checked

228− -> IO (Maybe a)

227+ -> IO (Maybe (Either a a))

229

228

230 goCheck1 [] m old = goWait (Snoc old m)

229 goCheck1 [] m old = goWait (Snoc old m)

231

230

@@ -238,10 +237,10 @@ dequeue (CQueue arrived incoming size) blockSpec matchons = mask_ $ decrementJus

238 goCheck1 (Left matches : rest) m old = do

237 goCheck1 (Left matches : rest) m old = do

239 case checkMatches matches m of

238 case checkMatches matches m of

240 Nothing -> goCheck1 rest m old

239 Nothing -> goCheck1 rest m old

241− Just p -> returnOld old (Just p)

240+ Just p -> returnOld old (Just (Right p))

242

241

243 -- a common pattern for putting back the arrived queue at the end

242 -- 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))

245 returnOld old r = do putMVar arrived old; return r

244 returnOld old r = do putMVar arrived old; return r

246

245

247 -- as a side-effect, this left-biases the list

246 -- as a side-effect, this left-biases the list