@@ -331,6 +331,18 @@ data SVarStats = SVarStats {
331-- This is essentially a 'Maybe Word' type
331-- This is essentially a 'Maybe Word' type
332data Limit = Unlimited | Limited Word deriving Show
332data Limit = Unlimited | Limited Word deriving Show
334+instance Eq Limit where
335+ Unlimited == Unlimited = True
336+ Unlimited == Limited _ = False
337+ Limited _ == Unlimited = False
338+ Limited x == Limited y = x == y
340+instance Ord Limit where
341+ Unlimited <= Unlimited = True
342+ Unlimited <= Limited _ = False
343+ Limited _ <= Unlimited = True
344+ Limited x <= Limited y = x <= y
334-- When to stop the composed stream.
346-- When to stop the composed stream.
335data SVarStopStyle =
347data SVarStopStyle =
336 StopNone -- stops only when all streams are finished
348 StopNone -- stops only when all streams are finished
@@ -375,8 +387,15 @@ data SVar t m a = SVar
375 , postProcess :: m Bool
387 , postProcess :: m Bool
377 -- Combined/aggregate parameters
389 -- Combined/aggregate parameters
390+ -- This is truncated to maxBufferLimit if set to more than that. Otherwise
391+ -- potentially each worker may yield one value to the buffer in the worst
392+ -- case exceeding the requested buffer size.
378 , maxWorkerLimit :: Limit
393 , maxWorkerLimit :: Limit
379 , maxBufferLimit :: Limit
394 , maxBufferLimit :: Limit
396+ -- [LOCKING] Read only access by consumer when dispatching a worker.
397+ -- Decremented by workers when picking work and undo decrement if the
398+ -- worker does not yield a value.
380 , remainingWork :: Maybe (IORef Count)
399 , remainingWork :: Maybe (IORef Count)
381 , yieldRateInfo :: Maybe YieldRateInfo
400 , yieldRateInfo :: Maybe YieldRateInfo
@@ -947,8 +966,6 @@ send sv msg = do
947 -- doorbell if something was added to the queue after it empties it.
966 -- doorbell if something was added to the queue after it empties it.
948 void $ tryPutMVar (outputDoorBell sv) ()
967 void $ tryPutMVar (outputDoorBell sv) ()
950− -- XXX we should reserve the buffer when we pick up the work from the
951− -- queue, instead of checking it here when it is too late.
952 let limit = maxBufferLimit sv
969 let limit = maxBufferLimit sv
953 case limit of
970 case limit of
954 Unlimited -> return True
971 Unlimited -> return True
@@ -2085,7 +2102,7 @@ getAheadSVar st f mrun = do
2085 { outputQueue = outQ
2102 { outputQueue = outQ
2086 , remainingWork = yl
2103 , remainingWork = yl
2087 , maxBufferLimit = getMaxBuffer st
2104 , maxBufferLimit = getMaxBuffer st
2088− , maxWorkerLimit = getMaxThreads st
2105+ , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
2089 , yieldRateInfo = rateInfo
2106 , yieldRateInfo = rateInfo
2090 , outputDoorBell = outQMv
2107 , outputDoorBell = outQMv
2091 , readOutputQ = readOutput sv
2108 , readOutputQ = readOutput sv