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 (MonadAsync, askRunInIO)
51import Streamly.Internal.Data.Atomics
52 (atomicModifyIORefCAS, atomicModifyIORefCAS_, writeBarrier)
53import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
54import Streamly.Internal.Data.Time.Units (NanoSecond64(..))
56import qualified Data.Heap as H
57import qualified Data.Set as S
59import Streamly.Internal.Data.SVar.Dispatch
60import Streamly.Internal.Data.SVar.Pull
61import Streamly.Internal.Data.SVar.Type
62import Streamly.Internal.Data.SVar.Worker
64-------------------------------------------------------------------------------
66-------------------------------------------------------------------------------
68getYieldRateInfo :: State t m a -> IO (Maybe YieldRateInfo)
69getYieldRateInfo st = do
70 -- convert rate in Hertz to latency in Nanoseconds
71 let rateToLatency r = if r <= 0 then maxBound else round $ 1.0e9 / r
72 case getStreamRate st of
73 Just (Rate low goal high buf) ->
74 let l = rateToLatency goal
75 minl = rateToLatency high
76 maxl = rateToLatency low
77 in mkYieldRateInfo l (LatencyRange minl maxl) buf
78 Nothing -> return Nothing
82 mkYieldRateInfo latency latRange buf = do
83 measured <- newIORef 0
84 wcur <- newIORef (0,0,0)
85 wcol <- newIORef (0,0,0)
86 now <- getTime Monotonic
87 wlong <- newIORef (0,now)
89 gainLoss <- newIORef (Count 0)
91 return $ Just YieldRateInfo
92 { svarLatencyTarget = latency
93 , svarLatencyRange = latRange
94 , svarRateBuffer = buf
95 , svarGainedLostYields = gainLoss
96 , workerBootstrapLatency = getStreamLatency st
97 , workerPollingInterval = period
98 , workerMeasuredLatency = measured
99 , workerPendingLatency = wcur
100 , workerCollectedLatency = wcol
101 , svarAllTimeLatency = wlong
104newSVarStats :: IO SVarStats
111 avgLat <- newIORef (0, NanoSecond64 0)
112 maxLat <- newIORef (NanoSecond64 0)
113 minLat <- newIORef (NanoSecond64 0)
114 stpTime <- newIORef Nothing
117 { totalDispatches = disp
118 , maxWorkers = maxWrk
119 , maxOutQSize = maxOq
120 , maxHeapSize = maxHs
121 , maxWorkQSize = maxWq
122 , avgWorkerLatency = avgLat
123 , minWorkerLatency = minLat
124 , maxWorkerLatency = maxLat
125 , svarStopTime = stpTime
128-------------------------------------------------------------------------------
130-------------------------------------------------------------------------------
132-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
133-- but always serve them in the same order as they appear in the stream. To
134-- implement lookahead streams efficiently we assign a sequence number to each
135-- task when the task is picked up for execution. When the task finishes, the
136-- output is tagged with the same sequence number and we rearrange the outputs
137-- in sequence based on that number.
139-- To explain the mechanism imagine that the current task at the head of the
140-- stream has a "token" to yield to the outputQueue. The ownership of the token
141-- is determined by the current sequence number is maintained in outputHeap.
142-- Sequence number is assigned when a task is queued. When a thread dequeues a
143-- task it picks up the sequence number as well and when the output is ready it
144-- uses the sequence number to queue the output to the outputQueue.
146-- The thread with current sequence number sends the output directly to the
147-- outputQueue. Other threads push the output to the outputHeap. When the task
148-- being queued on the heap is a stream of many elements we evaluate only the
149-- first element and keep the rest of the unevaluated computation in the heap.
150-- When such a task gets the "token" for outputQueue it evaluates and directly
151-- yields all the elements to the outputQueue without checking for the
154-- Note that no two outputs in the heap can have the same sequence numbers and
155-- therefore we do not need a stable heap. We have also separated the buffer
156-- for the current task (outputQueue) and the pending tasks (outputHeap) so
157-- that the pending tasks cannot interfere with the current task. Note that for
158-- a single task just the outputQueue is enough and for the case of many
159-- threads just a heap is good enough. However we balance between these two
160-- cases, so that both are efficient.
162-- For bigger streams it may make sense to have separate buffers for each
163-- stream. However, for singleton streams this may become inefficient. However,
164-- if we do not have separate buffers, then the streams that come later in
165-- sequence may hog the buffer, hindering the streams that are ahead. For this
166-- reason we have a single element buffer limitation for the streams being
167-- executed in advance.
169-- This scheme works pretty efficiently with less than 40% extra overhead
170-- compared to the Async streams where we do not have any kind of sequencing of
171-- the outputs. It is especially devised so that we are most efficient when we
172-- have short tasks and need just a single thread. Also when a thread yields
173-- many items it can hold lockfree access to the outputQueue and do it
176-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
177-- that they do not hog the resources and hinder the progress of the threads in
180-- XXX Left associated ahead expressions are expensive. We start a new SVar for
181-- each left associative expression. The queue is used only for right
182-- associated expression, we queue the right expression and execute the left.
183-- Thererefore the queue never has more than one item in it.
185-- XXX we can fix this. When we queue more than one item on the queue we can
186-- mark the previously queued item as not-runnable. The not-runnable item is
187-- not dequeued until the already running one has finished and at that time we
188-- would also know the exact sequence number of the already queued item.
190-- we can even run the already queued items but they will have to be sorted in
191-- layers in the heap. We can use a list of heaps for that.
192{-# INLINE enqueueAhead #-}
193enqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> (RunInIO m, t m a) -> IO ()
194enqueueAhead sv q m = do
195 atomicModifyIORefCAS_ q $ \ case
196 ([], n) -> ([snd m], n + 1) -- increment sequence
197 _ -> error "enqueueAhead: queue is not empty"
200-- enqueue without incrementing the sequence number
201{-# INLINE reEnqueueAhead #-}
202reEnqueueAhead :: SVar t m a -> IORef ([t m a], Int) -> t m a -> IO ()
203reEnqueueAhead sv q m = do
204 atomicModifyIORefCAS_ q $ \ case
205 ([], n) -> ([m], n) -- DO NOT increment sequence
206 _ -> error "reEnqueueAhead: queue is not empty"
209-- Normally the thread that has the token should never go away. The token gets
210-- handed over to another thread, but someone or the other has the token at any
211-- point of time. But if the task that has the token finds that the outputQueue
212-- is full, in that case it can go away without even handing over the token to
213-- another thread. In that case it sets the nextSequence number in the heap its
214-- own sequence number before going away. To handle this case, any task that
215-- does not have the token tries to dequeue from the heap first before
216-- dequeuing from the work queue. If it finds that the task at the top of the
217-- heap is the one that owns the current sequence number then it grabs the
218-- token and starts with that.
220-- XXX instead of queueing just the head element and the remaining computation
221-- on the heap, evaluate as many as we can and place them on the heap. But we
222-- need to give higher priority to the lower sequence numbers so that lower
223-- priority tasks do not fill up the heap making higher priority tasks block
224-- due to full heap. Maybe we can have a weighted space for them in the heap.
225-- The weight is inversely proportional to the sequence number.
227-- XXX review for livelock
229{-# INLINE queueEmptyAhead #-}
230queueEmptyAhead :: MonadIO m => IORef ([t m a], Int) -> m Bool
231queueEmptyAhead q = liftIO $ do
232 (xs, _) <- readIORef q
235{-# INLINE dequeueAhead #-}
236dequeueAhead :: MonadIO m
237 => IORef ([t m a], Int) -> m (Maybe (t m a, Int))
238dequeueAhead q = liftIO $
239 atomicModifyIORefCAS q $ \case
240 ([], n) -> (([], n), Nothing)
241 (x : [], n) -> (([], n), Just (x, n))
242 _ -> error "more than one item on queue"
244-------------------------------------------------------------------------------
246-------------------------------------------------------------------------------
248withIORef :: IORef a -> (a -> IO b) -> IO b
249withIORef ref f = readIORef ref >>= f
251atomicModifyIORef_ :: IORef a -> (a -> a) -> IO ()
252atomicModifyIORef_ ref f =
253 atomicModifyIORef ref $ \x -> (f x, ())
255data HeapDequeueResult t m a =
258 | Ready (Entry Int (AheadHeapEntry t m a))
260{-# INLINE dequeueFromHeap #-}
262 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
263 -> IO (HeapDequeueResult t m a)
264dequeueFromHeap hpVar =
265 atomicModifyIORef hpVar $ \pair@(hp, snum) ->
267 Nothing -> (pair, Clearing)
271 Just (ent@(Entry seqNo _ev), hp') ->
273 then ((hp', Nothing), Ready ent)
274 else assert (seqNo >= n) (pair, Waiting n)
275 Nothing -> (pair, Waiting n)
277{-# INLINE dequeueFromHeapSeq #-}
279 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
281 -> IO (HeapDequeueResult t m a)
282dequeueFromHeapSeq hpVar i =
283 atomicModifyIORef hpVar $ \(hp, snum) ->
288 Just (ent@(Entry seqNo _ev), hp') ->
290 then ((hp', Nothing), Ready ent)
291 else assert (seqNo >= i) ((hp, Just i), Waiting i)
292 Nothing -> ((hp, Just i), Waiting i)
293 Just _ -> error "dequeueFromHeapSeq: unreachable"
295heapIsSane :: Maybe Int -> Int -> Bool
296heapIsSane snum seqNo =
301{-# INLINE requeueOnHeapTop #-}
303 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
304 -> Entry Int (AheadHeapEntry t m a)
307requeueOnHeapTop hpVar ent seqNo =
308 atomicModifyIORef_ hpVar $ \(hp, snum) ->
309 assert (heapIsSane snum seqNo) (H.insert ent hp, Just seqNo)
311{-# INLINE updateHeapSeq #-}
313 :: IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
316updateHeapSeq hpVar seqNo =
317 atomicModifyIORef_ hpVar $ \(hp, snum) ->
318 assert (heapIsSane snum seqNo) (hp, Just seqNo)
320-- XXX remove polymorphism in t, inline f
321getAheadSVar :: MonadAsync m
323 -> ( IORef ([t m a], Int)
324 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
331getAheadSVar st f mrun = do
332 outQ <- newIORef ([], 0)
333 -- the second component of the tuple is "Nothing" when heap is being
334 -- cleared, "Just n" when we are expecting sequence number n to arrive
335 -- before we can start clearing the heap.
336 outH <- newIORef (H.empty, Just 0)
337 outQMv <- newEmptyMVar
339 wfw <- newIORef False
340 running <- newIORef S.empty
341 -- Sequence number is incremented whenever something is queued, therefore,
342 -- first sequence number would be 0
343 q <- newIORef ([], -1)
344 stopMVar <- newMVar ()
345 yl <- case getYieldLimit st of
346 Nothing -> return Nothing
347 Just x -> Just <$> newIORef x
348 rateInfo <- getYieldRateInfo st
350 stats <- newSVarStats
353 let getSVar sv readOutput postProc = SVar
355 , outputQueueFromConsumer = undefined
357 , maxBufferLimit = getMaxBuffer st
358 , pushBufferSpace = undefined
359 , pushBufferPolicy = undefined
360 , pushBufferMVar = undefined
361 , maxWorkerLimit = min (getMaxThreads st) (getMaxBuffer st)
362 , yieldRateInfo = rateInfo
363 , outputDoorBell = outQMv
364 , outputDoorBellFromConsumer = undefined
365 , readOutputQ = readOutput sv
366 , postProcess = postProc sv
367 , workerThreads = running
368 , workLoop = f q outH st{streamVar = Just sv} sv
369 , enqueue = enqueueAhead sv q
370 , isWorkDone = isWorkDoneAhead sv q outH
371 , isQueueDone = isQueueDoneAhead sv q
373 , svarStyle = AheadVar
374 , svarStopStyle = StopNone
375 , svarStopBy = undefined
377 , workerCount = active
378 , accountThread = delThread sv
379 , workerStopMVar = stopMVar
381 , svarInspectMode = getInspectMode st
389 case getStreamRate st of
390 Nothing -> getSVar sv readOutputQBounded postProcessBounded
391 Just _ -> getSVar sv readOutputQPaced postProcessPaced
396 {-# INLINE isQueueDoneAhead #-}
397 isQueueDoneAhead sv q = do
398 queueDone <- checkEmpty q
400 case remainingWork sv of
404 Nothing -> return False
405 -- XXX note that yieldsDone can only be authoritative only when there
406 -- are no workers running. If there are active workers they can
407 -- later increment the yield count and therefore change the result.
408 return $ yieldsDone || queueDone
410 {-# INLINE isWorkDoneAhead #-}
411 isWorkDoneAhead sv q ref = do
413 (hp, _) <- readIORef ref
414 return (H.size hp <= 0)
415 queueDone <- isQueueDoneAhead sv q
416 return $ heapDone && queueDone
419 (xs, _) <- readIORef q
422{-# INLINABLE newAheadVar #-}
423newAheadVar :: MonadAsync m
426 -> ( IORef ([t m a], Int)
427 -> IORef (Heap (Entry Int (AheadHeapEntry t m a)), Maybe Int)
433newAheadVar st m wloop = do
435 sv <- liftIO $ getAheadSVar st wloop mrun
438-------------------------------------------------------------------------------
440-------------------------------------------------------------------------------
442-- XXX To be implemented. Use a linked queue like WAsync and put back the
443-- remaining computation at the back of the queue instead of the heap, and
444-- increment the sequence number.
446-------------------------------------------------------------------------------
448-------------------------------------------------------------------------------
450getParallelSVar :: MonadIO m
451 => SVarStopStyle -> State t m a -> RunInIO m -> IO (SVar t m a)
452getParallelSVar ss st mrun = do
453 outQ <- newIORef ([], 0)
454 outQRev <- newIORef ([], 0)
455 outQMv <- newEmptyMVar
456 outQMvRev <- newEmptyMVar
458 running <- newIORef S.empty
459 yl <- case getYieldLimit st of
460 Nothing -> return Nothing
461 Just x -> Just <$> newIORef x
462 rateInfo <- getYieldRateInfo st
464 case getMaxBuffer st of
465 Unlimited -> undefined
466 Limited x -> fromIntegral x
467 remBuf <- newIORef bufLim
470 stats <- newSVarStats
475 StopBy -> liftIO $ newIORef undefined
476 _ -> return undefined
479 SVar { outputQueue = outQ
480 , outputQueueFromConsumer = outQRev
482 , maxBufferLimit = getMaxBuffer st
483 , pushBufferSpace = remBuf
484 , pushBufferPolicy = PushBufferBlock
485 , pushBufferMVar = pbMVar
486 , maxWorkerLimit = Unlimited
487 -- Used only for diagnostics
488 , yieldRateInfo = rateInfo
489 , outputDoorBell = outQMv
490 , outputDoorBellFromConsumer = outQMvRev
491 , readOutputQ = readOutputQPar sv
492 , postProcess = allThreadsDone sv
493 , workerThreads = running
494 , workLoop = undefined
495 , enqueue = undefined
496 , isWorkDone = undefined
497 , isQueueDone = undefined
498 , needDoorBell = undefined
499 , svarStyle = ParallelVar
501 , svarStopBy = stopBy
503 , workerCount = active
504 , accountThread = modifyThread sv
505 , workerStopMVar = undefined
507 , svarInspectMode = getInspectMode st
509 , aheadWorkQueue = undefined
510 , outputHeap = undefined
517 readOutputQPar sv = liftIO $ do
518 withDiagMVar sv "readOutputQPar: doorbell"
519 $ takeMVar (outputDoorBell sv)
520 case yieldRateInfo sv of
522 Just yinfo -> void $ collectLatency sv yinfo False
523 r <- fst `fmap` readOutputQRaw sv
525 void $ tryTakeMVar (pushBufferMVar sv)
528 void $ tryPutMVar (pushBufferMVar sv) ()
531{-# INLINABLE newParallelVar #-}
532newParallelVar :: MonadAsync m
533 => SVarStopStyle -> State t m a -> m (SVar t m a)
534newParallelVar ss st = do
536 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