@@ -69,14 +69,64 @@ pushWorker yieldMax sv = do
70 exception = handleChildException (outputQueue sv) (outputDoorBell sv)
70 exception = handleChildException (outputQueue sv) (outputDoorBell sv)
72−-- Returns:
72+-- | Determine the maximum number of workers required based on 'maxWorkerLimit'
73−-- True: can dispatch more
73+-- and 'remainingWork'.
74−-- False: cannot dispatch any more
74+{-# INLINE getEffectiveWorkerLimit #-}
75−dispatchWorker :: (MonadIO m, MonadBaseControl IO m) => Count -> Channel m a -> m Bool
75+getEffectiveWorkerLimit :: MonadIO m => Channel m a -> m Limit
76−dispatchWorker yieldCount sv = do
76+getEffectiveWorkerLimit sv = do
77 let workerLimit = maxWorkerLimit sv
77 let workerLimit = maxWorkerLimit sv
78+ case remainingWork sv of
79+ Nothing -> return workerLimit
81+ n <- liftIO $ readIORef ref
82+ case yieldRateInfo sv of
83+ Just _ -> return workerLimit
87+ Unlimited -> Limited (fromIntegral n)
88+ Limited lim -> Limited $ min lim (fromIntegral n)
90+-- | Determine whether the active threads are more than the max threads we are
91+-- allowed to dispatch.
92+{-# INLINE checkMaxThreads #-}
93+checkMaxThreads :: MonadIO m => Int -> Channel m a -> m Bool
94+checkMaxThreads active sv = do
95+ -- Note that we may deadlock if the previous workers (tasks in the
96+ -- stream) wait/depend on the future workers (tasks in the stream)
97+ -- executing. In that case we should either configure the maxWorker
98+ -- count to higher or use parallel style instead of ahead or async
100+ limit <- getEffectiveWorkerLimit sv
104+ -- Note that the use of remainingWork and workerCount is not
105+ -- atomic and the counts may even have changed between reading
106+ -- and using them here, so this is just approximate logic and
107+ -- we cannot rely on it for correctness. We may actually
108+ -- dispatch more workers than required.
109+ Limited lim -> fromIntegral lim > active
111+-- | Determine whether we would exceed max buffer if we dispatch more workers
112+-- based on the current outputQueue size and active workers.
113+{-# INLINE checkMaxBuffer #-}
114+checkMaxBuffer :: MonadIO m => Int -> Channel m a -> m Bool
115+checkMaxBuffer active sv = do
116+ let limit = maxBufferLimit sv
118+ Unlimited -> return True
120+ (_, n) <- liftIO $ readIORef (outputQueue sv)
121+ return $ fromIntegral lim > n + active
123+dispatchWorker :: (MonadIO m, MonadBaseControl IO m) =>
124+ Count -> Channel m a -> m Bool
125+dispatchWorker yieldCount sv = do
78 -- XXX in case of Ahead streams we should not send more than one worker
126 -- XXX in case of Ahead streams we should not send more than one worker
79 -- when the work queue is done but heap is not done.
127 -- when the work queue is done but heap is not done.
128+ -- XXX Should we have a single abstraction for checking q and
129+ -- work instead checking the two separately?
80 done <- liftIO $ isWorkDone sv
130 done <- liftIO $ isWorkDone sv
81 -- Note, "done" may not mean that the work is actually finished if there
131 -- Note, "done" may not mean that the work is actually finished if there
82 -- are workers active, because there may be a worker which has not yet
132 -- are workers active, because there may be a worker which has not yet
@@ -84,43 +134,26 @@ dispatchWorker yieldCount sv = do
84 if not done
134 if not done
86 qDone <- liftIO $ isQueueDone sv
136 qDone <- liftIO $ isQueueDone sv
87− -- This count may not be accurate as it is decremented by the workers
137+ -- This count may be more until the sendStop events are processed.
88− -- and we have no synchronization with that decrement.
89 active <- liftIO $ readIORef $ workerCount sv
138 active <- liftIO $ readIORef $ workerCount sv
139+ when (active < 0) $ error "dispatchWorker active negative"
90 if not qDone
140 if not qDone
92− -- Note that we may deadlock if the previous workers (tasks in the
93− -- stream) wait/depend on the future workers (tasks in the stream)
94− -- executing. In that case we should either configure the maxWorker
95− -- count to higher or use parallel style instead of ahead or async
97− limit <- case remainingWork sv of
98− Nothing -> return workerLimit
100− n <- liftIO $ readIORef ref
101− case yieldRateInfo sv of
102− Just _ -> return workerLimit
106− Unlimited -> Limited (fromIntegral n)
107− Limited lim -> Limited $ min lim (fromIntegral n)
109 -- XXX for ahead streams shall we take the heap yields into account
142 -- XXX for ahead streams shall we take the heap yields into account
110 -- for controlling the dispatch? We should not dispatch if the heap
143 -- for controlling the dispatch? We should not dispatch if the heap
111 -- has already got the limit covered.
144 -- has already got the limit covered.
145+ r <- checkMaxThreads active sv
148+ r1 <- checkMaxBuffer active sv
112− let dispatch = pushWorker yieldCount sv >> return True
150+ then pushWorker yieldCount sv >> return True
113− in case limit of
151+ else return False
114− Unlimited -> dispatch
115− -- Note that the use of remainingWork and workerCount is not
116− -- atomic and the counts may even have changed between reading
117− -- and using them here, so this is just approximate logic and
118− -- we cannot rely on it for correctness. We may actually
119− -- dispatch more workers than required.
120− Limited lim | lim > fromIntegral active -> dispatch
121− _ -> return False
152+ else return False
123− when (active <= 0) $ pushWorker 0 sv
154+ when (active <= 0) $ do
155+ r <- liftIO $ isWorkDone sv
156+ when (not r) $ pushWorker 0 sv
124 return False
157 return False
125 else return False
158 else return False