2-- Module : Streamly.Internal.Data.SVar
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
10module Streamly.Internal.Data.SVar
12 module Streamly.Internal.Data.SVar.Type
13 , module Streamly.Internal.Data.SVar.Worker
14 , module Streamly.Internal.Data.SVar.Dispatch
15 , module Streamly.Internal.Data.SVar.Pull
29 , HeapDequeueResult(..)
42import Control.Concurrent (myThreadId, takeMVar)
43import Control.Concurrent.MVar (newEmptyMVar, tryPutMVar, tryTakeMVar, newMVar)
44import Control.Exception (assert)
45import Control.Monad (void)
46import Control.Monad.IO.Class (MonadIO(liftIO))
47import Data.Heap (Heap, Entry(..))
48import Data.IORef (newIORef, readIORef)
49import Data.IORef (IORef, atomicModifyIORef)
50import Streamly.Internal.Control.Concurrent
51 (MonadAsync, askRunInIO, RunInIO)
52import Streamly.Internal.Data.Atomics
53 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier)
54import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
55import Streamly.Internal.Data.Time.Units (NanoSecond64(..))
57import qualified Data.Heap as H
58import qualified Data.Set as S
60import Streamly.Internal.Data.SVar.Dispatch
61import Streamly.Internal.Data.SVar.Pull
62import Streamly.Internal.Data.SVar.Type
63import Streamly.Internal.Data.SVar.Worker
65-------------------------------------------------------------------------------
67-------------------------------------------------------------------------------
69getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
70getYieldRateInfo st = do
71 -- convert rate in Hertz to latency in Nanoseconds
72 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
73 case getStreamRate st of
74 Just (Rate low goal high buf) ->
75 let l = rateToLatency goal
76 minl = rateToLatency high
77 maxl = rateToLatency low
78 in mkYieldRateInfo l (LatencyRange minl maxl) buf
79 Nothing -> return Nothing
83 mkYieldRateInfo latency latRange buf = do
84 measured <- newIORef 0
85 wcur <- newIORef (0,0,0)
86 wcol <- newIORef (0,0,0)
87 now <- getTime Monotonic
88 wlong <- newIORef (0,now)
90 gainLoss <- newIORef (Count 0)
92 return $ Just YieldRateInfo
93 { svarLatencyTarget = latency
94 , svarLatencyRange = latRange
95 , svarRateBuffer = buf
96 , svarGainedLostYields = gainLoss
97 , workerBootstrapLatency = getStreamLatency st
98 , workerPollingInterval = period
99 , workerMeasuredLatency = measured
100 , workerPendingLatency = wcur
101 , workerCollectedLatency = wcol
102 , svarAllTimeLatency = wlong
105newSVarStats :: IO SVarStats
112 avgLat <- newIORef (0, NanoSecond64 0)
113 maxLat <- newIORef (NanoSecond64 0)
114 minLat <- newIORef (NanoSecond64 0)
115 stpTime <- newIORef Nothing
118 { totalDispatches = disp
119 , maxWorkers = maxWrk
120 , maxOutQSize = maxOq
121 , maxHeapSize = maxHs
122 , maxWorkQSize = maxWq
123 , avgWorkerLatency = avgLat
124 , minWorkerLatency = minLat
125 , maxWorkerLatency = maxLat
126 , svarStopTime = stpTime
129-------------------------------------------------------------------------------
131-------------------------------------------------------------------------------
133-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
134-- but always serve them in the same order as they appear in the stream. To
135-- implement lookahead streams efficiently we assign a sequence number to each
136-- task when the task is picked up for execution. When the task finishes, the
137-- output is tagged with the same sequence number and we rearrange the outputs
138-- in sequence based on that number.
140-- To explain the mechanism imagine that the current task at the head of the
141-- stream has a "token" to yield to the outputQueue. The ownership of the token
142-- is determined by the current sequence number is maintained in outputHeap.
143-- Sequence number is assigned when a task is queued. When a thread dequeues a
144-- task it picks up the sequence number as well and when the output is ready it
145-- uses the sequence number to queue the output to the outputQueue.
147-- The thread with current sequence number sends the output directly to the
148-- outputQueue. Other threads push the output to the outputHeap. When the task
149-- being queued on the heap is a stream of many elements we evaluate only the
150-- first element and keep the rest of the unevaluated computation in the heap.
151-- When such a task gets the "token" for outputQueue it evaluates and directly
152-- yields all the elements to the outputQueue without checking for the
155-- Note that no two outputs in the heap can have the same sequence numbers and
156-- therefore we do not need a stable heap. We have also separated the buffer
157-- for the current task (outputQueue) and the pending tasks (outputHeap) so
158-- that the pending tasks cannot interfere with the current task. Note that for
159-- a single task just the outputQueue is enough and for the case of many
160-- threads just a heap is good enough. However we balance between these two
161-- cases, so that both are efficient.
163-- For bigger streams it may make sense to have separate buffers for each
164-- stream. However, for singleton streams this may become inefficient. However,
165-- if we do not have separate buffers, then the streams that come later in
166-- sequence may hog the buffer, hindering the streams that are ahead. For this
167-- reason we have a single element buffer limitation for the streams being
168-- executed in advance.
170-- This scheme works pretty efficiently with less than 40% extra overhead
171-- compared to the Async streams where we do not have any kind of sequencing of
172-- the outputs. It is especially devised so that we are most efficient when we
173-- have short tasks and need just a single thread. Also when a thread yields
174-- many items it can hold lockfree access to the outputQueue and do it
177-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
178-- that they do not hog the resources and hinder the progress of the threads in
181-- XXX Left associated ahead expressions are expensive. We start a new SVar for
182-- each left associative expression. The queue is used only for right
183-- associated expression, we queue the right expression and execute the left.
184-- Thererefore the queue never has more than one item in it.
186-- XXX we can fix this. When we queue more than one item on the queue we can
187-- mark the previously queued item as not-runnable. The not-runnable item is
188-- not dequeued until the already running one has finished and at that time we
189-- would also know the exact sequence number of the already queued item.
191-- we can even run the already queued items but they will have to be sorted in
192-- layers in the heap. We can use a list of heaps for that.
193{-# INLINE enqueueAhead #-}
194enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> (RunInIO m, t m a) -> IO ()
195enqueueAhead sv q m = do
196 atomicModifyIORefCAS_ q $ \ case
197 ([], n) -> ([snd m], n + 1) -- increment sequence
198 _ -> error "enqueueAhead: queue is not empty"
201-- enqueue without incrementing the sequence number
202{-# INLINE reEnqueueAhead #-}
203reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
204reEnqueueAhead sv q m = do
205 atomicModifyIORefCAS_ q $ \ case
206 ([], n) -> ([m], n) -- DO NOT increment sequence
207 _ -> error "reEnqueueAhead: queue is not empty"
210-- Normally the thread that has the token should never go away. The token gets
211-- handed over to another thread, but someone or the other has the token at any
212-- point of time. But if the task that has the token finds that the outputQueue
213-- is full, in that case it can go away without even handing over the token to
214-- another thread. In that case it sets the nextSequence number in the heap its
215-- own sequence number before going away. To handle this case, any task that
216-- does not have the token tries to dequeue from the heap first before
217-- dequeuing from the work queue. If it finds that the task at the top of the
218-- heap is the one that owns the current sequence number then it grabs the
219-- token and starts with that.
221-- XXX instead of queueing just the head element and the remaining computation
222-- on the heap, evaluate as many as we can and place them on the heap. But we
223-- need to give higher priority to the lower sequence numbers so that lower
224-- priority tasks do not fill up the heap making higher priority tasks block
225-- due to full heap. Maybe we can have a weighted space for them in the heap.
226-- The weight is inversely proportional to the sequence number.
228-- XXX review for livelock
230{-# INLINE queueEmptyAhead #-}
231queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
232queueEmptyAhead q = liftIO $ do
233 (xs, _) <- readIORef q
236{-# INLINE dequeueAhead #-}
237dequeueAhead :: MonadIO m
238 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
239dequeueAhead q = liftIO $
240 atomicModifyIORefCAS q $ \case
241 ([], n) -> (([], n), Nothing)
242 (x : [], n) -> (([], n), Just (x, n))
243 _ -> error "more than one item on queue"
245-------------------------------------------------------------------------------
247-------------------------------------------------------------------------------
249withIORef :: IORef a -> (a -> IO b) -> IO b
250withIORef ref f = readIORef ref >>= f
252atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()
253atomicModifyIORef_ ref f =
254 atomicModifyIORef ref $ \x -> (f x, ())
256data HeapDequeueResult t m a =
259 | Ready (Entry Int (AheadHeapEntry t m a))
261{-# INLINE dequeueFromHeap #-}
263 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
264 -> IO (HeapDequeueResult t m a)
265dequeueFromHeap hpVar =
266 atomicModifyIORef hpVar $ \pair@(hp, snum) ->
268 Nothing -> (pair, Clearing)
272 Just (ent@(Entry seqNo _ev), hp') ->
274 then ((hp', Nothing), Ready ent)
275 else assert (seqNo >= n) (pair, Waiting n)
276 Nothing -> (pair, Waiting n)
278{-# INLINE dequeueFromHeapSeq #-}
280 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
282 -> IO (HeapDequeueResult t m a)
283dequeueFromHeapSeq hpVar i =
284 atomicModifyIORef hpVar $ \(hp, snum) ->
289 Just (ent@(Entry seqNo _ev), hp') ->
291 then ((hp', Nothing), Ready ent)
292 else assert (seqNo >= i) ((hp, Just i), Waiting i)
293 Nothing -> ((hp, Just i), Waiting i)
294 Just _ -> error "dequeueFromHeapSeq: unreachable"
296heapIsSane :: Maybe Int -> Int -> Bool
297heapIsSane snum seqNo =
302{-# INLINE requeueOnHeapTop #-}
304 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
305 -> Entry Int (AheadHeapEntry t m a)
308requeueOnHeapTop hpVar ent seqNo =
309 atomicModifyIORef_ hpVar $ \(hp, snum) ->
310 assert (heapIsSane snum seqNo) (H.insert ent hp, Just seqNo)
312{-# INLINE updateHeapSeq #-}
314 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
317updateHeapSeq hpVar seqNo =
318 atomicModifyIORef_ hpVar $ \(hp, snum) ->
319 assert (heapIsSane snum seqNo) (hp, Just seqNo)
321-- XXX remove polymorphism in t, inline f
322getAheadSVar :: MonadAsync m
324 -> ( IORef ([t m a], Int)
325 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
332getAheadSVar st f mrun = do
333 outQ <- newIORef ([], 0)
334 -- the second component of the tuple is "Nothing" when heap is being
335 -- cleared, "Just n" when we are expecting sequence number n to arrive
336 -- before we can start clearing the heap.
337 outH <- newIORef (H.empty, Just 0)
338 outQMv <- newEmptyMVar
340 wfw <- newIORef False
341 running <- newIORef S.empty
342 -- Sequence number is incremented whenever something is queued, therefore,
343 -- first sequence number would be 0
344 q <- newIORef ([], -1)
345 stopMVar <- newMVar ()
346 yl <- case getYieldLimit st of
347 Nothing -> return Nothing
348 Just x -> Just <$> newIORef x
349 rateInfo <- getYieldRateInfo st
351 stats <- newSVarStats
354 let getSVar sv readOutput postProc = SVar
356 , outputQueueFromConsumer = undefined
358 , maxBufferLimit = getMaxBuffer st
359 , pushBufferSpace = undefined
360 , pushBufferPolicy = undefined
361 , pushBufferMVar = undefined
362 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
363 , yieldRateInfo = rateInfo
364 , outputDoorBell = outQMv
365 , outputDoorBellFromConsumer = undefined
366 , readOutputQ = readOutput sv
367 , postProcess = postProc sv
368 , workerThreads = running
369 , workLoop = f q outH st{streamVar = Just sv} sv
370 , enqueue = enqueueAhead sv q
371 , isWorkDone = isWorkDoneAhead sv q outH
372 , isQueueDone = isQueueDoneAhead sv q
374 , svarStyle = AheadVar
375 , svarStopStyle = StopNone
376 , svarStopBy = undefined
378 , workerCount = active
379 , accountThread = delThread sv
380 , workerStopMVar = stopMVar
382 , svarInspectMode = getInspectMode st
390 case getStreamRate st of
391 Nothing -> getSVar sv readOutputQBounded postProcessBounded
392 Just _ -> getSVar sv readOutputQPaced postProcessPaced
397 {-# INLINE isQueueDoneAhead #-}
398 isQueueDoneAhead sv q = do
399 queueDone <- checkEmpty q
401 case remainingWork sv of
405 Nothing -> return False
406 -- XXX note that yieldsDone can only be authoritative only when there
407 -- are no workers running. If there are active workers they can
408 -- later increment the yield count and therefore change the result.
409 return $ yieldsDone || queueDone
411 {-# INLINE isWorkDoneAhead #-}
412 isWorkDoneAhead sv q ref = do
414 (hp, _) <- readIORef ref
415 return (H.size hp <= 0)
416 queueDone <- isQueueDoneAhead sv q
417 return $ heapDone && queueDone
420 (xs, _) <- readIORef q
423{-# INLINABLE newAheadVar #-}
424newAheadVar :: MonadAsync m
427 -> ( IORef ([t m a], Int)
428 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
434newAheadVar st m wloop = do
436 sv <- liftIO $ getAheadSVar st wloop mrun
439-------------------------------------------------------------------------------
441-------------------------------------------------------------------------------
443-- XXX To be implemented. Use a linked queue like WAsync and put back the
444-- remaining computation at the back of the queue instead of the heap, and
445-- increment the sequence number.
447-------------------------------------------------------------------------------
449-------------------------------------------------------------------------------
451getParallelSVar :: MonadIO m
452 => SVarStopStyle -> State t m a -> RunInIO m -> IO (SVar t m a)
453getParallelSVar ss st mrun = do
454 outQ <- newIORef ([], 0)
455 outQRev <- newIORef ([], 0)
456 outQMv <- newEmptyMVar
457 outQMvRev <- newEmptyMVar
459 running <- newIORef S.empty
460 yl <- case getYieldLimit st of
461 Nothing -> return Nothing
462 Just x -> Just <$> newIORef x
463 rateInfo <- getYieldRateInfo st
465 case getMaxBuffer st of
466 Unlimited -> undefined
467 Limited x -> fromIntegral x
468 remBuf <- newIORef bufLim
471 stats <- newSVarStats
476 StopBy -> liftIO $ newIORef undefined
477 _ -> return undefined
480 SVar { outputQueue = outQ
481 , outputQueueFromConsumer = outQRev
483 , maxBufferLimit = getMaxBuffer st
484 , pushBufferSpace = remBuf
485 , pushBufferPolicy = PushBufferBlock
486 , pushBufferMVar = pbMVar
487 , maxWorkerLimit = Unlimited
488 -- Used only for diagnostics
489 , yieldRateInfo = rateInfo
490 , outputDoorBell = outQMv
491 , outputDoorBellFromConsumer = outQMvRev
492 , readOutputQ = readOutputQPar sv
493 , postProcess = allThreadsDone sv
494 , workerThreads = running
495 , workLoop = undefined
496 , enqueue = undefined
497 , isWorkDone = undefined
498 , isQueueDone = undefined
499 , needDoorBell = undefined
500 , svarStyle = ParallelVar
502 , svarStopBy = stopBy
504 , workerCount = active
505 , accountThread = modifyThread sv
506 , workerStopMVar = undefined
508 , svarInspectMode = getInspectMode st
510 , aheadWorkQueue = undefined
511 , outputHeap = undefined
518 readOutputQPar sv = liftIO $ do
519 withDiagMVar sv "readOutputQPar: doorbell"
520 $ takeMVar (outputDoorBell sv)
521 case yieldRateInfo sv of
523 Just yinfo -> void $ collectLatency sv yinfo False
524 r <- fst `fmap` readOutputQRaw sv
526 void $ tryTakeMVar (pushBufferMVar sv)
529 void $ tryPutMVar (pushBufferMVar sv) ()
532{-# INLINABLE newParallelVar #-}
533newParallelVar :: MonadAsync m
534 => SVarStopStyle -> State t m a -> m (SVar t m a)
535newParallelVar ss st = do
537 liftIO $ getParallelSVar ss st mrun
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list