git.iko.soy/streamly
on ea703d4 help sign in
streamly/src/Streamly/Internal/Data/Stream/Concurrent/Channel/Dispatcher.hs
417 lines 18 kB history raw
1-- |
2-- Module : Streamly.Internal.Data.Stream.Concurrent.Channel.Dispatcher
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
7-- Portability : GHC
8--
9--
10module Streamly.Internal.Data.Stream.Concurrent.Channel.Dispatcher
11 (
12 -- * Dispatching
13 pushWorker
14 , dispatchWorker
15 , dispatchWorkerPaced
16 , sendWorkerWait
17 , startChannel
18 , sendWorkerDelay
19 , sendWorkerDelayPaced
20 )
21where
23import Control.Concurrent (takeMVar, threadDelay)
24import Control.Exception (assert)
25import Control.Monad (when, void)
26import Control.Monad.IO.Class (MonadIO(liftIO))
27import Data.Maybe (fromJust, fromMaybe)
28import Data.IORef (modifyIORef, newIORef, readIORef, writeIORef)
29import Streamly.Internal.Control.Concurrent (MonadRunInIO)
30import Streamly.Internal.Control.ForkLifted (doFork)
31import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS_, storeLoadBarrier)
32import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
33import Streamly.Internal.Data.Time.Units
34 (MicroSecond64(..), diffAbsTime64, fromRelTime64, toRelTime64)
36import Streamly.Internal.Data.Stream.Concurrent.Channel.Type
37import Streamly.Internal.Data.Stream.Channel.Dispatcher
38import Streamly.Internal.Data.Stream.Channel.Types
39import Streamly.Internal.Data.Stream.Channel.Worker
41-------------------------------------------------------------------------------
42-- Dispatching workers
43-------------------------------------------------------------------------------
45{-# NOINLINE pushWorker #-}
46pushWorker :: MonadRunInIO m => Count -> Channel m a -> m ()
47pushWorker yieldMax sv = do
48 liftIO $ atomicModifyIORefCAS_ (workerCount sv) $ \n -> n + 1
49 when (svarInspectMode sv)
50 $ recordMaxWorkers (workerCount sv) (svarStats sv)
51 -- This allocation matters when significant number of workers are being
52 -- sent. We allocate it only when needed.
53 winfo <-
54 case yieldRateInfo sv of
55 Nothing -> return Nothing
56 Just _ -> liftIO $ do
57 cntRef <- newIORef 0
58 t <- getTime Monotonic
59 lat <- newIORef (0, t)
60 return $ Just WorkerInfo
61 { workerYieldMax = yieldMax
62 , workerYieldCount = cntRef
63 , workerLatencyStart = lat
64 }
65 -- In case of lazy dispatch we dispatch workers only from the consumer
66 -- thread. In that case it is ok to use addThread here as it is guaranteed
67 -- that the thread will be added to the workerSet before the thread STOP
68 -- event is processed, because we do both of these actions in the same
69 -- consumer thread. However, in case of eager dispatch we may dispatch
70 -- workers from workers, in which case the thread Stop even may get
71 -- processed before the addThread occurs, so in that case we have to use
72 -- modifyThread which performs a toggle rather than adding or deleting.
73 --
74 -- XXX We can use addThread or modThread based on eager flag.
75 doFork (workLoop sv winfo) (svarMrun sv) exception >>= modThread
77 where
79 modThread = modifyThread (workerThreads sv) (outputDoorBell sv)
80 exception = handleChildException (outputQueue sv) (outputDoorBell sv)
82-- | Determine the maximum number of workers required based on 'maxWorkerLimit'
83-- and 'remainingWork'.
84{-# INLINE getEffectiveWorkerLimit #-}
85getEffectiveWorkerLimit :: MonadIO m => Channel m a -> m Limit
86getEffectiveWorkerLimit sv = do
87 let workerLimit = maxWorkerLimit sv
88 case remainingWork sv of
89 Nothing -> return workerLimit
90 Just ref -> do
91 n <- liftIO $ readIORef ref
92 case yieldRateInfo sv of
93 Just _ -> return workerLimit
94 Nothing ->
95 return $
96 case workerLimit of
97 Unlimited -> Limited (fromIntegral n)
98 Limited lim -> Limited $ min lim (fromIntegral n)
100-- | Determine whether the active threads are more than the max threads we are
101-- allowed to dispatch.
102{-# INLINE checkMaxThreads #-}
103checkMaxThreads :: MonadIO m => Int -> Channel m a -> m Bool
104checkMaxThreads active sv = do
105 -- Note that we may deadlock if the previous workers (tasks in the
106 -- stream) wait/depend on the future workers (tasks in the stream)
107 -- executing. In that case we should either configure the maxWorker
108 -- count to higher or use parallel style instead of ahead or async
109 -- style.
110 limit <- getEffectiveWorkerLimit sv
111 return
112 $ case limit of
113 Unlimited -> True
114 -- Note that the use of remainingWork and workerCount is not
115 -- atomic and the counts may even have changed between reading
116 -- and using them here, so this is just approximate logic and
117 -- we cannot rely on it for correctness. We may actually
118 -- dispatch more workers than required.
119 Limited lim -> fromIntegral lim > active
121-- | Determine whether we would exceed max buffer if we dispatch more workers
122-- based on the current outputQueue size and active workers.
123{-# INLINE checkMaxBuffer #-}
124checkMaxBuffer :: MonadIO m => Int -> Channel m a -> m Bool
125checkMaxBuffer active sv = do
126 let limit = maxBufferLimit sv
127 case limit of
128 Unlimited -> return True
129 Limited lim -> do
130 (_, n) <- liftIO $ readIORef (outputQueue sv)
131 return $ fromIntegral lim > n + active
133dispatchWorker :: MonadRunInIO m =>
134 Count -> Channel m a -> m Bool
135dispatchWorker yieldCount sv = do
136 -- XXX in case of Ahead streams we should not send more than one worker
137 -- when the work queue is done but heap is not done.
138 -- XXX Should we have a single abstraction for checking q and
139 -- work instead checking the two separately?
140 done <- liftIO $ isWorkDone sv
141 -- Note, "done" may not mean that the work is actually finished if there
142 -- are workers active, because there may be a worker which has not yet
143 -- queued the leftover work.
144 if not done
145 then do
146 qDone <- liftIO $ isQueueDone sv
147 -- This count may be more until the sendStop events are processed.
148 active <- liftIO $ readIORef $ workerCount sv
149 when (active < 0) $ error "dispatchWorker active negative"
150 if not qDone
151 then do
152 -- XXX for ahead streams shall we take the heap yields into account
153 -- for controlling the dispatch? We should not dispatch if the heap
154 -- has already got the limit covered.
155 r <- checkMaxThreads active sv
156 if r
157 then do
158 r1 <- checkMaxBuffer active sv
159 if r1
160 then pushWorker yieldCount sv >> return True
161 else return False
162 else return False
163 else do
164 when (active <= 0) $ do
165 r <- liftIO $ isWorkDone sv
166 when (not r) $ pushWorker 0 sv
167 return False
168 else return False
170-- XXX in case of ahead style stream we need to take the heap size into account
171-- because we return the workers on the basis of that which causes a condition
172-- where we keep dispatching and they keep returning. So we must have exactly
173-- the same logic for not dispatching and for returning.
174--
175-- Returns:
176-- True: can dispatch more
177-- False: full, no more dispatches
178dispatchWorkerPaced :: MonadRunInIO m =>
179 Channel m a -> m Bool
180dispatchWorkerPaced sv = do
181 let yinfo = fromJust $ yieldRateInfo sv
182 (svarYields, svarElapsed, wLatency) <- do
183 now <- liftIO $ getTime Monotonic
184 (yieldCount, baseTime, lat) <-
185 liftIO
186 $ collectLatency
187 (svarInspectMode sv) (svarStats sv) yinfo False
188 let elapsed = fromRelTime64 $ diffAbsTime64 now baseTime
189 let latency =
190 if lat == 0
191 then fromMaybe lat (workerBootstrapLatency yinfo)
192 else lat
194 return (yieldCount, elapsed, latency)
196 if wLatency == 0
197 -- Need to measure the latency with a single worker before we can perform
198 -- any computation.
199 then return False
200 else do
201 let workerLimit = maxWorkerLimit sv
202 let targetLat = svarLatencyTarget yinfo
203 let range = svarLatencyRange yinfo
204 gainLoss <- liftIO $ readIORef (svarGainedLostYields yinfo)
205 let work = estimateWorkers workerLimit svarYields gainLoss svarElapsed
206 wLatency targetLat range
208 -- XXX we need to take yieldLimit into account here. If we are at the
209 -- end of the limit as well as the time, we should not be sleeping.
210 -- If we are not actually planning to dispatch any more workers we need
211 -- to take that in account.
212 case work of
213 BlockWait s -> do
214 assert (s >= 0) (return ())
215 -- XXX note that when we return from here we will block waiting
216 -- for the result from the existing worker. If that takes too
217 -- long we won't be able to send another worker until the
218 -- result arrives.
219 --
220 -- Sleep only if there are no active workers, otherwise we will
221 -- defer the output of those. Note we cannot use workerCount
222 -- here as it is not a reliable way to ensure there are
223 -- definitely no active workers. When workerCount is 0 we may
224 -- still have a Stop event waiting in the outputQueue.
225 done <- allThreadsDone (workerThreads sv)
226 when done $ void $ do
227 let us = fromRelTime64 (toRelTime64 s) :: MicroSecond64
228 liftIO $ threadDelay (fromIntegral us)
229 dispatchWorker 1 sv
230 return False
231 PartialWorker yields -> do
232 assert (yields > 0) (return ())
233 updateGainedLostYields yinfo yields
235 done <- allThreadsDone (workerThreads sv)
236 when done $ void $ dispatchWorker yields sv
237 return False
238 ManyWorkers netWorkers yields -> do
239 assert (netWorkers >= 1) (return ())
240 assert (yields >= 0) (return ())
241 updateGainedLostYields yinfo yields
243 let periodRef = workerPollingInterval yinfo
244 ycnt = max 1 $ yields `div` fromIntegral netWorkers
245 period = min ycnt (fromIntegral magicMaxBuffer)
247 old <- liftIO $ readIORef periodRef
248 when (period < old) $
249 liftIO $ writeIORef periodRef period
251 cnt <- liftIO $ readIORef $ workerCount sv
252 if cnt < netWorkers
253 then do
254 let total = netWorkers - cnt
255 batch = max 1 $ fromIntegral $
256 minThreadDelay `div` targetLat
257 -- XXX stagger the workers over a period?
258 -- XXX cannot sleep, as that would mean we cannot process
259 -- the outputs. need to try a different mechanism to
260 -- stagger.
261 -- when (total > batch) $
262 -- liftIO $ threadDelay $ nanoToMicroSecs minThreadDelay
263 dispatchN (min total batch)
264 else return False
266 where
268 updateGainedLostYields yinfo yields = do
269 let buf = fromIntegral $ svarRateBuffer yinfo
270 when (yields /= 0 && abs yields > buf) $ do
271 let delta =
272 if yields > 0
273 then yields - buf
274 else yields + buf
275 liftIO $ modifyIORef (svarGainedLostYields yinfo) (+ delta)
277 dispatchN n =
278 if n == 0
279 then return True
280 else do
281 r <- dispatchWorker 0 sv
282 if r
283 then dispatchN (n - 1)
284 else return False
286{-# NOINLINE sendWorkerWait #-}
287sendWorkerWait
288 :: MonadIO m
289 => Bool
290 -> (Channel m a -> IO ())
291 -> (Channel m a -> m Bool)
292 -> Channel m a
293 -> m ()
294sendWorkerWait eagerEval delay dispatch sv = go
296 where
298 go = do
300 -- Note that we are guaranteed to have at least one outstanding worker
301 -- when we enter this function. So if we sleep we are guaranteed to be
302 -- woken up by an outputDoorBell, when the worker exits.
304 liftIO $ delay sv
305 (_, n) <- liftIO $ readIORef (outputQueue sv)
306 when (n <= 0 || eagerEval) $ do
307 -- The queue may be empty temporarily if the worker has dequeued
308 -- the work item but has not enqueued the remaining part yet. For
309 -- the same reason, a worker may come back if it tries to dequeue
310 -- and finds the queue empty, even though the whole work has not
311 -- finished yet.
313 -- If we find that the queue is empty, but it may be empty
314 -- temporarily, when we checked it. If that's the case we might
315 -- sleep indefinitely unless the active workers produce some
316 -- output. We may deadlock specially if the otuput from the active
317 -- workers depends on the future workers that we may never send.
318 -- So in case the queue was temporarily empty set a flag to inform
319 -- the enqueue to send us a doorbell.
321 -- Note that this is just a best effort mechanism to avoid a
322 -- deadlock. Deadlocks may still happen if for some weird reason
323 -- the consuming computation shares an MVar or some other resource
324 -- with the producing computation and gets blocked on that resource
325 -- and therefore cannot do any pushworker to add more threads to
326 -- the producer. In such cases the programmer should use a parallel
327 -- style so that all the producers are scheduled immediately and
328 -- unconditionally. We can also use a separate monitor thread to
329 -- push workers instead of pushing them from the consumer, but then
330 -- we are no longer using pull based concurrency rate adaptation.
331 --
332 -- XXX update this in the tutorial.
333 --
334 -- Having pending active workers does not mean that we are
335 -- guaranteed to be woken up if we sleep. In case of Ahead streams,
336 -- there may be queued items in the heap even though the
337 -- outputQueue is empty, and we may have active workers which are
338 -- deadlocked on those items to be processed by the consumer. We
339 -- should either guarantee that any worker, before returning,
340 -- clears the heap or we send a worker to clear it. Normally we
341 -- always send a worker if no output is seen, but if the thread
342 -- limit is reached or we are using pacing then we may not send a
343 -- worker. See the concurrentApplication test in the tests, that
344 -- test case requires at least one yield from the producer to not
345 -- deadlock, if the last workers output is stuck in the heap then
346 -- this test fails. This problem can be extended to n threads when
347 -- the consumer may depend on the evaluation of next n items in the
348 -- producer stream.
350 -- register for the outputDoorBell before we check the queue so
351 -- that if we sleep because the queue was empty we are guaranteed
352 -- to get a doorbell on the next enqueue.
354 liftIO $ atomicModifyIORefCAS_ (doorBellOnWorkQ sv) $ const True
355 liftIO storeLoadBarrier
356 canDoMore <- dispatch sv
358 -- XXX test for the case when we miss sending a worker when the
359 -- worker count is more than 1500.
360 --
361 -- XXX Assert here that if the heap is not empty then there is at
362 -- least one outstanding worker. Otherwise we could be sleeping
363 -- forever.
365 if canDoMore
366 then go
367 else do
368 liftIO
369 $ withDiagMVar
370 (svarInspectMode sv)
371 (dumpSVar sv)
372 "sendWorkerWait: nothing to do"
373 $ takeMVar (outputDoorBell sv)
374 (_, len) <- liftIO $ readIORef (outputQueue sv)
375 if len <= 0
376 then go
377 else
378 liftIO
379 $ atomicModifyIORefCAS_ (doorBellOnWorkQ sv)
380 $ const False
382-- | Start the evaluation of the channel's work queue by kicking off a worker.
383-- Note: Work queue must not be empty otherwise the worker will exit without
384-- doing anything.
385startChannel :: MonadRunInIO m =>
386 Channel m a -> m ()
387startChannel chan = do
388 case yieldRateInfo chan of
389 Nothing -> pushWorker 0 chan
390 Just yinfo ->
391 if svarLatencyTarget yinfo == maxBound
392 then liftIO $ threadDelay maxBound
393 else pushWorker 1 chan
395sendWorkerDelayPaced :: Channel m a -> IO ()
396sendWorkerDelayPaced _ = return ()
398sendWorkerDelay :: Channel m a -> IO ()
399sendWorkerDelay _sv =
400 -- XXX we need a better way to handle this than hardcoded delays. The
401 -- delays may be different for different systems.
402 -- If there is a usecase where this is required we can create a combinator
403 -- to set it as a config in the state.
404 {-
405 do
406 ncpu <- getNumCapabilities
407 if ncpu <= 1
408 then
409 if (svarStyle sv == AheadVar)
410 then threadDelay 100
411 else threadDelay 25
412 else
413 if (svarStyle sv == AheadVar)
414 then threadDelay 100
415 else threadDelay 10
416 -}
417 return ()