1{-# LANGUAGE UndecidableInstances #-}
4-- Module : Streamly.Internal.Data.Stream.Ahead
5-- Copyright : (c) 2017 Composewell Technologies
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
12-- To run examples in this module:
14-- >>> import qualified Streamly.Prelude as Stream
15-- >>> import Control.Concurrent (threadDelay)
18-- threadDelay (n * 1000000) -- sleep for n seconds
19-- putStrLn (show n ++ " sec") -- print "n sec"
23module Streamly.Internal.Data.Stream.Ahead
32import Control.Concurrent.MVar (putMVar, takeMVar)
33import Control.Exception (assert)
34import Control.Monad (void, when)
35import Control.Monad.Base (MonadBase(..), liftBaseDefault)
36import Control.Monad.Catch (MonadThrow, throwM)
37-- import Control.Monad.Error.Class (MonadError(..))
38import Control.Monad.IO.Class (MonadIO(..))
39import Control.Monad.Reader.Class (MonadReader(..))
40import Control.Monad.State.Class (MonadState(..))
41import Control.Monad.Trans.Class (MonadTrans(lift))
42import Data.Heap (Heap, Entry(..))
43import Data.IORef (IORef, readIORef, atomicModifyIORef, writeIORef)
44import Data.Maybe (fromJust)
45#if __GLASGOW_HASKELL__ < 808
46import Data.Semigroup (Semigroup(..))
48import GHC.Exts (inline)
50import qualified Data.Heap as H
52import Streamly.Internal.Control.Concurrent
53 (MonadRunInIO, MonadAsync, RunInIO(..), askRunInIO, restoreM)
54import Streamly.Internal.Data.Stream.Serial (SerialT(..))
55import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
57import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
58import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
60import Streamly.Internal.Data.Stream.SVar.Generate
61import Streamly.Internal.Data.SVar
62import Prelude hiding (map)
64#include "Instances.hs"
67-- >>> import qualified Streamly.Prelude as Stream
68-- >>> import Control.Concurrent (threadDelay)
71-- threadDelay (n * 1000000) -- sleep for n seconds
72-- putStrLn (show n ++ " sec") -- print "n sec"
76-------------------------------------------------------------------------------
78-------------------------------------------------------------------------------
80-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
81-- but always serve them in the same order as they appear in the stream. To
82-- implement lookahead streams efficiently we assign a sequence number to each
83-- task when the task is picked up for execution. When the task finishes, the
84-- output is tagged with the same sequence number and we rearrange the outputs
85-- in sequence based on that number.
87-- To explain the mechanism imagine that the current task at the head of the
88-- stream has a "token" to yield to the outputQueue. The ownership of the token
89-- is determined by the current sequence number is maintained in outputHeap.
90-- Sequence number is assigned when a task is queued. When a thread dequeues a
91-- task it picks up the sequence number as well and when the output is ready it
92-- uses the sequence number to queue the output to the outputQueue.
94-- The thread with current sequence number sends the output directly to the
95-- outputQueue. Other threads push the output to the outputHeap. When the task
96-- being queued on the heap is a stream of many elements we evaluate only the
97-- first element and keep the rest of the unevaluated computation in the heap.
98-- When such a task gets the "token" for outputQueue it evaluates and directly
99-- yields all the elements to the outputQueue without checking for the
102-- Note that no two outputs in the heap can have the same sequence numbers and
103-- therefore we do not need a stable heap. We have also separated the buffer
104-- for the current task (outputQueue) and the pending tasks (outputHeap) so
105-- that the pending tasks cannot interfere with the current task. Note that for
106-- a single task just the outputQueue is enough and for the case of many
107-- threads just a heap is good enough. However we balance between these two
108-- cases, so that both are efficient.
110-- For bigger streams it may make sense to have separate buffers for each
111-- stream. However, for singleton streams this may become inefficient. However,
112-- if we do not have separate buffers, then the streams that come later in
113-- sequence may hog the buffer, hindering the streams that are ahead. For this
114-- reason we have a single element buffer limitation for the streams being
115-- executed in advance.
117-- This scheme works pretty efficiently with less than 40% extra overhead
118-- compared to the Async streams where we do not have any kind of sequencing of
119-- the outputs. It is especially devised so that we are most efficient when we
120-- have short tasks and need just a single thread. Also when a thread yields
121-- many items it can hold lockfree access to the outputQueue and do it
124-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
125-- that they do not hog the resources and hinder the progress of the threads in
128-- Left associated ahead expressions are expensive. We start a new SVar for
129-- each left associative expression. The queue is used only for right
130-- associated expression, we queue the right expression and execute the left.
131-- Thererefore the queue never has more than on item in it.
133-- XXX Also note that limiting concurrency for cases like "take 10" would not
134-- work well with left associative expressions, because we have no visibility
135-- about how much the left side of the expression would yield.
137-- XXX It may be a good idea to increment sequence numbers for each yield,
138-- currently a stream on the left side of the expression may yield many
139-- elements with the same sequene number. We can then use the seq number to
140-- enforce yieldMax and yieldLImit as well.
144-- * A worker should always ensure that it pushes all the consecutive items in
145-- the heap to the outputQueue especially the items on behalf of the workers
146-- that have already left when we were holding the token. This avoids deadlock
147-- conditions when the later workers completion depends on the consumption of
148-- earlier results. For more details see comments in the consumer pull side
151{-# INLINE underMaxHeap #-}
154 -> Heap (Entry Int (AheadHeapEntry Stream m a))
156underMaxHeap sv hp = do
157 (_, len) <- readIORef (outputQueue sv)
160 let maxHeap = case maxBufferLimit sv of
161 Limited lim -> Limited $
162 max 0 (lim - fromIntegral len)
163 Unlimited -> Unlimited
167 active <- readIORef (workerCount sv)
168 return $ H.size hp + active <= fromIntegral lim
169 Unlimited -> return True
176 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Maybe Int)
178preStopCheck sv heap =
179 -- check the stop condition under a lock before actually
180 -- stopping so that the whole herd does not stop at once.
181 withIORef heap $ \(hp, _) -> do
182 heapOk <- underMaxHeap sv hp
183 takeMVar (workerStopMVar sv)
185 putMVar (workerStopMVar sv) ()
188 putMVar (workerStopMVar sv) ()
192 case yieldRateInfo sv of
195 rateOk <- isBeyondMaxRate sv yinfo
196 if rateOk then continue else stop
200 IORef ([Stream m a], Int)
205abortExecution q sv winfo m = do
206 reEnqueueAhead sv q m
207 incrementYieldLimit sv
210-- XXX In absence of a "noyield" primitive (i.e. do not pre-empt inside a
211-- critical section) from GHC RTS, we have a difficult problem. Assume we have
212-- a 100,000 threads producing output and queuing it to the heap for
213-- sequencing. The heap can be drained only by one thread at a time, any thread
214-- that finds that heap can be drained now, takes a lock and starts draining
215-- it, however the thread may get prempted in the middle of it holding the
216-- lock. Since that thread is holding the lock, the other threads cannot pick
217-- up the draining task, therefore they proceed to picking up the next task to
218-- execute. If the draining thread could yield voluntarily at a point where it
219-- has released the lock, then the next threads could pick up the draining
220-- instead of executing more tasks. When there are 100,000 threads the drainer
221-- gets a cpu share to run only 1:100000 of the time. This makes the heap
222-- accumulate a lot of output when we the buffer size is large.
224-- The solutions to this problem are:
225-- 1) make the other threads wait in a queue until the draining finishes
226-- 2) make the other threads queue and go away if draining is in progress
228-- In both cases we give the drainer a chance to run more often.
232 => IORef ([Stream m a], Int)
233 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)
237 -> AheadHeapEntry Stream m a
239 -> Bool -- we are draining the heap before we stop
241processHeap q heap st sv winfo entry sno stopping = loopHeap sno entry
245 stopIfNeeded ent seqNo r = do
246 stopIt <- liftIO $ preStopCheck sv heap
249 -- put the entry back in the heap and stop
250 requeueOnHeapTop heap (Entry seqNo ent) seqNo
252 else runStreamWithYieldLimit True seqNo r
256 AheadEntryNull -> nextHeap seqNo
257 AheadEntryPure a -> do
258 -- Use 'send' directly so that we do not account this in worker
259 -- latency as this will not be the real latency.
260 -- Don't stop the worker in this case as we are just
261 -- transferring available results from heap to outputQueue.
262 void $ liftIO $ send sv (ChildYield a)
264 AheadEntryStream (RunInIO runin, r) ->
266 then stopIfNeeded ent seqNo r
268 res <- liftIO $ runin (runStreamWithYieldLimit True seqNo r)
272 nextHeap prevSeqNo = do
273 res <- liftIO $ dequeueFromHeapSeq heap (prevSeqNo + 1)
275 Ready (Entry seqNo hent) -> loopHeap seqNo hent
276 Clearing -> liftIO $ sendStop sv winfo
280 r <- liftIO $ preStopCheck sv heap
282 then liftIO $ sendStop sv winfo
283 else processWorkQueue prevSeqNo
284 else inline processWorkQueue prevSeqNo
286 processWorkQueue prevSeqNo = do
287 work <- dequeueAhead q
289 Nothing -> liftIO $ sendStop sv winfo
290 Just (m, seqNo) -> do
291 yieldLimitOk <- liftIO $ decrementYieldLimit sv
294 if seqNo == prevSeqNo + 1
295 then processWithToken q heap st sv winfo m seqNo
296 else processWithoutToken q heap st sv winfo m seqNo
297 else liftIO $ abortExecution q sv winfo m
299 -- We do not stop the worker on buffer full here as we want to proceed to
300 -- nextHeap anyway so that we can clear any subsequent entries. We stop
301 -- only in yield continuation where we may have a remaining stream to be
302 -- pushed on the heap.
303 singleStreamFromHeap seqNo a = do
304 void $ liftIO $ sendYield sv winfo (ChildYield a)
307 -- XXX when we have an unfinished stream on the heap we cannot account all
308 -- the yields of that stream until it finishes, so if we have picked up
309 -- and executed more actions beyond that in the parent stream and put them
310 -- on the heap then they would eat up some yield limit which is not
311 -- correct, we will think that our yield limit is over even though we have
312 -- to yield items from unfinished stream before them. For this reason, if
313 -- there are pending items in the heap we drain them unconditionally
314 -- without considering the yield limit.
315 runStreamWithYieldLimit continue seqNo r = do
316 _ <- liftIO $ decrementYieldLimit sv
317 if continue -- see comment above -- && yieldLimitOk
320 liftIO (incrementYieldLimit sv)
322 K.foldStreamShared st
323 (yieldStreamFromHeap seqNo)
324 (singleStreamFromHeap seqNo)
329 let ent = Entry seqNo (AheadEntryStream (runIn, r))
331 requeueOnHeapTop heap ent seqNo
332 incrementYieldLimit sv
335 yieldStreamFromHeap seqNo a r = do
336 continue <- liftIO $ sendYield sv winfo (ChildYield a)
337 runStreamWithYieldLimit continue seqNo r
339{-# NOINLINE drainHeap #-}
342 => IORef ([Stream m a], Int)
343 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)
348drainHeap q heap st sv winfo = do
349 r <- liftIO $ dequeueFromHeap heap
351 Ready (Entry seqNo hent) ->
352 processHeap q heap st sv winfo hent seqNo True
353 _ -> liftIO $ sendStop sv winfo
355data HeapStatus = HContinue | HStop
356data WorkerStatus = Continue | Suspend
360 => IORef ([Stream m a], Int)
361 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)
368processWithoutToken q heap st sv winfo m seqNo = do
369 -- we have already decremented the yield limit for m
371 liftIO (incrementYieldLimit sv)
372 -- If the stream stops without yielding anything, and we do not put
373 -- anything on heap, but if heap was waiting for this seq number
374 -- then it will keep waiting forever, because we are never going to
375 -- put it on heap. So we have to put a null entry on heap even when
377 toHeap AheadEntryNull
378 mrun = runInIO $ svarMrun sv
381 K.foldStreamShared st
384 toHeap $ AheadEntryStream (runIn, K.cons a r))
385 (toHeap . AheadEntryPure)
390 Continue -> workLoopAhead q heap st sv winfo
391 Suspend -> drainHeap q heap st sv winfo
395 -- XXX to reduce contention each CPU can have its own heap
397 -- Heap insertion is an expensive affair so we use a non CAS based
398 -- modification, otherwise contention and retries can make a thread
399 -- context switch and throw it behind other threads which come later in
401 newHp <- liftIO $ atomicModifyIORef heap $ \(hp, snum) ->
402 let hp' = H.insert (Entry seqNo ent) hp
403 in assert (heapIsSane snum seqNo) ((hp', snum), hp')
405 when (svarInspectMode sv) $
407 maxHp <- readIORef (maxHeapSize $ svarStats sv)
408 when (H.size newHp > maxHp) $
409 writeIORef (maxHeapSize $ svarStats sv) (H.size newHp)
411 heapOk <- liftIO $ underMaxHeap sv newHp
413 case yieldRateInfo sv of
414 Nothing -> return HContinue
418 rateOk <- liftIO $ workerRateControl sv yinfo info
420 then return HContinue
422 Nothing -> return HContinue
427 HContinue -> return Continue
428 HStop -> return Suspend
431data TokenWorkerStatus = TokenContinue Int | TokenSuspend
435 => IORef ([Stream m a], Int)
436 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)
443processWithToken q heap st sv winfo action sno = do
444 -- Note, we enter this function with yield limit already decremented
445 -- XXX deduplicate stop in all invocations
447 liftIO (incrementYieldLimit sv)
448 return $ TokenContinue (sno + 1)
449 mrun = runInIO $ svarMrun sv
452 K.foldStreamShared st (yieldOutput sno) (singleOutput sno) stop action
456 TokenContinue seqNo -> loopWithToken seqNo
457 TokenSuspend -> drainHeap q heap st sv winfo
461 singleOutput seqNo a = do
462 continue <- liftIO $ sendYield sv winfo (ChildYield a)
464 then return $ TokenContinue (seqNo + 1)
466 liftIO $ updateHeapSeq heap (seqNo + 1)
469 -- XXX use a wrapper function around stop so that we never miss
470 -- incrementing the yield in a stop continuation. Essentiatlly all
471 -- "unstream" calls in this function must increment yield limit on stop.
472 yieldOutput seqNo a r = do
473 continue <- liftIO $ sendYield sv winfo (ChildYield a)
474 yieldLimitOk <- liftIO $ decrementYieldLimit sv
475 if continue && yieldLimitOk
478 liftIO (incrementYieldLimit sv)
479 return $ TokenContinue (seqNo + 1)
480 K.foldStreamShared st
487 let ent = Entry seqNo (AheadEntryStream (runIn, r))
488 liftIO $ requeueOnHeapTop heap ent seqNo
489 liftIO $ incrementYieldLimit sv
492 loopWithToken nextSeqNo = do
493 work <- dequeueAhead q
496 liftIO $ updateHeapSeq heap nextSeqNo
497 workLoopAhead q heap st sv winfo
499 Just (m, seqNo) -> do
500 yieldLimitOk <- liftIO $ decrementYieldLimit sv
501 let undo = liftIO $ do
502 updateHeapSeq heap nextSeqNo
503 reEnqueueAhead sv q m
504 incrementYieldLimit sv
507 if seqNo == nextSeqNo
510 liftIO (incrementYieldLimit sv)
511 return $ TokenContinue (seqNo + 1)
512 mrun = runInIO $ svarMrun sv
514 K.foldStreamShared st
521 TokenContinue seqNo1 -> loopWithToken seqNo1
522 TokenSuspend -> drainHeap q heap st sv winfo
525 -- To avoid a race when another thread puts something
526 -- on the heap and goes away, the consumer will not get
527 -- a doorBell and we will not clear the heap before
528 -- executing the next action. If the consumer depends
529 -- on the output that is stuck in the heap then this
530 -- will result in a deadlock. So we always clear the
531 -- heap before executing the next action.
532 undo >> workLoopAhead q heap st sv winfo
533 else undo >> drainHeap q heap st sv winfo
535-- XXX the yield limit changes increased the performance overhead by 30-40%.
536-- Just like AsyncT we can use an implementation without yeidlimit and even
537-- without pacing code to keep the performance higher in the unlimited and
540-- XXX The yieldLimit stuff is pretty invasive. We can instead do it by using
541-- three hooks, a pre-execute hook, a yield hook and a stop hook. In fact these
542-- hooks can be used for a more general implementation to even check predicates
543-- and not just yield limit.
545-- XXX we can remove the sv parameter as it can be derived from st
549 => IORef ([Stream m a], Int)
550 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)
555workLoopAhead q heap st sv winfo = do
556 r <- liftIO $ dequeueFromHeap heap
558 Ready (Entry seqNo hent) ->
559 processHeap q heap st sv winfo hent seqNo False
560 Clearing -> liftIO $ sendStop sv winfo
562 -- Before we execute the next item from the work queue we check
563 -- if we are beyond the yield limit. It is better to check the
564 -- yield limit before we pick up the next item. Otherwise we
565 -- may have already started more tasks even though we may have
566 -- reached the yield limit. We can avoid this by taking active
567 -- workers into account, but that is not as reliable, because
568 -- workers may go away without picking up work and yielding a
571 -- Rate control can be done either based on actual yields in
572 -- the output queue or based on any yield either to the heap or
573 -- to the output queue. In both cases we may have one issue or
574 -- the other. We chose to do this based on actual yields to the
575 -- output queue because it makes the code common to both async
576 -- and ahead streams.
578 work <- dequeueAhead q
580 Nothing -> liftIO $ sendStop sv winfo
581 Just (m, seqNo) -> do
582 yieldLimitOk <- liftIO $ decrementYieldLimit sv
586 then processWithToken q heap st sv winfo m seqNo
587 else processWithoutToken q heap st sv winfo m seqNo
588 -- If some worker decremented the yield limit but then
589 -- did not yield anything and therefore incremented it
590 -- later, then if we did not requeue m here we may find
591 -- the work queue empty and therefore miss executing
592 -- the remaining action.
593 else liftIO $ abortExecution q sv winfo m
595-------------------------------------------------------------------------------
597-------------------------------------------------------------------------------
599-- XXX To be implemented. Use a linked queue like WAsync and put back the
600-- remaining computation at the back of the queue instead of the heap, and
601-- increment the sequence number.
603-- The only difference between forkSVarAsync and this is that we run the left
604-- computation without a shared SVar.
605forkSVarAhead :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
606forkSVarAhead m1 m2 = K.mkStream $ \st yld sng stp -> do
607 sv <- newAheadVar st (concurrently m1 m2)
609 K.foldStream st yld sng stp $ getSerialT (fromSVar sv)
611 concurrently ma mb = K.mkStream $ \st yld sng stp -> do
612 runInIO <- askRunInIO
613 liftIO $ enqueue (fromJust $ streamVar st) (runInIO, mb)
614 K.foldStream st yld sng stp ma
617aheadK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
618aheadK m1 m2 = K.mkStream $ \st yld sng stp ->
620 Just sv | svarStyle sv == AheadVar -> do
621 runInIO <- askRunInIO
622 liftIO $ enqueue sv (runInIO, m2)
623 -- Always run the left side on a new SVar to avoid complexity in
624 -- sequencing results. This means the left side cannot further
625 -- split into more ahead computations on the same SVar.
626 K.foldStream st yld sng stp m1
627 _ -> K.foldStreamShared st yld sng stp (forkSVarAhead m1 m2)
629-- | XXX we can implement it more efficienty by directly implementing instead
630-- of combining streams using ahead.
632{-# SPECIALIZE consM :: IO a -> AheadT IO a -> AheadT IO a #-}
633consM :: MonadAsync m => m a -> AheadT m a -> AheadT m a
634consM m (AheadT r) = AheadT $ aheadK (K.fromEffect m) r
636------------------------------------------------------------------------------
638------------------------------------------------------------------------------
640-- | For 'AheadT' streams:
643-- (<>) = 'Streamly.Prelude.ahead'
644-- (>>=) = flip . 'Streamly.Prelude.concatMapWith' 'Streamly.Prelude.ahead'
647-- A single 'Monad' bind behaves like a @for@ loop with iterations executed
648-- concurrently, ahead of time, producing side effects of iterations out of
649-- order, but results in order:
652-- Stream.toList $ Stream.fromAhead $ do
653-- x <- Stream.fromList [2,1] -- foreach x in stream
654-- Stream.fromEffect $ delay x
660-- Nested monad binds behave like nested @for@ loops with nested iterations
661-- executed concurrently, ahead of time:
664-- Stream.toList $ Stream.fromAhead $ do
665-- x <- Stream.fromList [1,2] -- foreach x in stream
666-- y <- Stream.fromList [2,4] -- foreach y in stream
667-- Stream.fromEffect $ delay (x + y)
675-- The behavior can be explained as follows. All the iterations corresponding
676-- to the element @1@ in the first stream constitute one output stream and all
677-- the iterations corresponding to @2@ constitute another output stream and
678-- these two output streams are merged using 'ahead'.
680-- /Since: 0.3.0 ("Streamly")/
683newtype AheadT m a = AheadT {getAheadT :: Stream m a}
684 deriving (MonadTrans)
686-- | A serial IO stream of elements of type @a@ with concurrent lookahead. See
687-- 'AheadT' documentation for more details.
689-- /Since: 0.3.0 ("Streamly")/
692type Ahead = AheadT IO
694------------------------------------------------------------------------------
696------------------------------------------------------------------------------
699{-# SPECIALIZE append :: AheadT IO a -> AheadT IO a -> AheadT IO a #-}
700append :: MonadAsync m => AheadT m a -> AheadT m a -> AheadT m a
701append (AheadT m1) (AheadT m2) = AheadT $ aheadK m1 m2
703instance MonadAsync m => Semigroup (AheadT m a) where
706------------------------------------------------------------------------------
708------------------------------------------------------------------------------
710instance MonadAsync m => Monoid (AheadT m a) where
711 mempty = AheadT K.nil
714------------------------------------------------------------------------------
716------------------------------------------------------------------------------
718{-# INLINE apAhead #-}
719apAhead :: MonadAsync m => AheadT m (a -> b) -> AheadT m a -> AheadT m b
720apAhead (AheadT m1) (AheadT m2) =
721 let f x1 = K.concatMapWith aheadK (pure . x1) m2
722 in AheadT $ K.concatMapWith aheadK f m1
724instance (Monad m, MonadAsync m) => Applicative (AheadT m) where
726 pure = AheadT . K.fromPure
731------------------------------------------------------------------------------
733------------------------------------------------------------------------------
735{-# INLINE bindAhead #-}
736{-# SPECIALIZE bindAhead ::
737 AheadT IO a -> (a -> AheadT IO b) -> AheadT IO b #-}
738bindAhead :: MonadAsync m => AheadT m a -> (a -> AheadT m b) -> AheadT m b
739bindAhead (AheadT m) f = AheadT $ K.bindWith aheadK m (getAheadT . f)
741instance MonadAsync m => Monad (AheadT m) where
747------------------------------------------------------------------------------
749------------------------------------------------------------------------------
751MONAD_COMMON_INSTANCES(AheadT, MONADPARALLEL)
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