2{-# LANGUAGE ConstraintKinds #-}
3{-# LANGUAGE FlexibleContexts #-}
4{-# LANGUAGE FlexibleInstances #-}
5{-# LANGUAGE GeneralizedNewtypeDeriving#-}
6{-# LANGUAGE InstanceSigs #-}
7{-# LANGUAGE MultiParamTypeClasses #-}
8{-# LANGUAGE UndecidableInstances #-} -- XXX
11-- Module : Streamly.Streams.Ahead
12-- Copyright : (c) 2017 Harendra Kumar
15-- Maintainer : harendra.kumar@gmail.com
16-- Stability : experimental
20module Streamly.Streams.Ahead
29import Control.Concurrent.MVar (putMVar, takeMVar)
30import Control.Exception (assert)
31import Control.Monad (ap, void, when)
32import Control.Monad.Base (MonadBase(..), liftBaseDefault)
33import Control.Monad.Catch (MonadThrow, throwM)
34-- import Control.Monad.Error.Class (MonadError(..))
35import Control.Monad.IO.Class (MonadIO(..))
36import Control.Monad.Reader.Class (MonadReader(..))
37import Control.Monad.State.Class (MonadState(..))
38import Control.Monad.Trans.Class (MonadTrans(lift))
39import Data.Heap (Heap, Entry(..))
40import Data.IORef (IORef, readIORef, atomicModifyIORef, writeIORef)
41import Data.Maybe (fromJust)
42import Data.Semigroup (Semigroup(..))
43import GHC.Exts (inline)
45import qualified Data.Heap as H
47import Streamly.Streams.SVar (fromSVar)
48import Streamly.Streams.Serial (map)
50import Streamly.Streams.StreamK
51 (IsStream(..), Stream, mkStream, foldStream, foldStreamShared,
53import qualified Streamly.Streams.StreamK as K
55import Prelude hiding (map)
57#include "Instances.hs"
59-------------------------------------------------------------------------------
61-------------------------------------------------------------------------------
63-- Lookahead streams can execute multiple tasks concurrently, ahead of time,
64-- but always serve them in the same order as they appear in the stream. To
65-- implement lookahead streams efficiently we assign a sequence number to each
66-- task when the task is picked up for execution. When the task finishes, the
67-- output is tagged with the same sequence number and we rearrange the outputs
68-- in sequence based on that number.
70-- To explain the mechanism imagine that the current task at the head of the
71-- stream has a "token" to yield to the outputQueue. The ownership of the token
72-- is determined by the current sequence number is maintained in outputHeap.
73-- Sequence number is assigned when a task is queued. When a thread dequeues a
74-- task it picks up the sequence number as well and when the output is ready it
75-- uses the sequence number to queue the output to the outputQueue.
77-- The thread with current sequence number sends the output directly to the
78-- outputQueue. Other threads push the output to the outputHeap. When the task
79-- being queued on the heap is a stream of many elements we evaluate only the
80-- first element and keep the rest of the unevaluated computation in the heap.
81-- When such a task gets the "token" for outputQueue it evaluates and directly
82-- yields all the elements to the outputQueue without checking for the
85-- Note that no two outputs in the heap can have the same sequence numbers and
86-- therefore we do not need a stable heap. We have also separated the buffer
87-- for the current task (outputQueue) and the pending tasks (outputHeap) so
88-- that the pending tasks cannot interfere with the current task. Note that for
89-- a single task just the outputQueue is enough and for the case of many
90-- threads just a heap is good enough. However we balance between these two
91-- cases, so that both are efficient.
93-- For bigger streams it may make sense to have separate buffers for each
94-- stream. However, for singleton streams this may become inefficient. However,
95-- if we do not have separate buffers, then the streams that come later in
96-- sequence may hog the buffer, hindering the streams that are ahead. For this
97-- reason we have a single element buffer limitation for the streams being
98-- executed in advance.
100-- This scheme works pretty efficiently with less than 40% extra overhead
101-- compared to the Async streams where we do not have any kind of sequencing of
102-- the outputs. It is especially devised so that we are most efficient when we
103-- have short tasks and need just a single thread. Also when a thread yields
104-- many items it can hold lockfree access to the outputQueue and do it
107-- XXX Maybe we can start the ahead threads at a lower cpu and IO priority so
108-- that they do not hog the resources and hinder the progress of the threads in
111-- Left associated ahead expressions are expensive. We start a new SVar for
112-- each left associative expression. The queue is used only for right
113-- associated expression, we queue the right expression and execute the left.
114-- Thererefore the queue never has more than on item in it.
116-- XXX Also note that limiting concurrency for cases like "take 10" would not
117-- work well with left associative expressions, because we have no visibility
118-- about how much the left side of the expression would yield.
120-- XXX It may be a good idea to increment sequence numbers for each yield,
121-- currently a stream on the left side of the expression may yield many
122-- elements with the same sequene number. We can then use the seq number to
123-- enforce yieldMax and yieldLImit as well.
127-- * A worker should always ensure that it pushes all the consecutive items in
128-- the heap to the outputQueue especially the items on behalf of the workers
129-- that have already left when we were holding the token. This avoids deadlock
130-- conditions when the later workers completion depends on the consumption of
131-- earlier results. For more details see comments in the consumer pull side
134{-# INLINE underMaxHeap #-}
137 -> Heap (Entry Int (AheadHeapEntry Stream m a))
139underMaxHeap sv hp = do
140 (_, len) <- readIORef (outputQueue sv)
143 let maxHeap = case maxBufferLimit sv of
144 Limited lim -> Limited $
145 max 0 (lim - fromIntegral len)
146 Unlimited -> Unlimited
150 active <- readIORef (workerCount sv)
151 return $ H.size hp + active <= fromIntegral lim
152 Unlimited -> return True
159 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Maybe Int)
161preStopCheck sv heap =
162 -- check the stop condition under a lock before actually
163 -- stopping so that the whole herd does not stop at once.
164 withIORef heap $ \(hp, _) -> do
165 heapOk <- underMaxHeap sv hp
166 takeMVar (workerStopMVar sv)
168 putMVar (workerStopMVar sv) ()
171 putMVar (workerStopMVar sv) ()
175 case yieldRateInfo sv of
178 rateOk <- isBeyondMaxRate sv yinfo
179 if rateOk then continue else stop
183 IORef ([Stream m a], Int)
188abortExecution q sv winfo m = do
189 reEnqueueAhead sv q m
190 incrementYieldLimit sv
193-- XXX In absence of a "noyield" primitive (i.e. do not pre-empt inside a
194-- critical section) from GHC RTS, we have a difficult problem. Assume we have
195-- a 100,000 threads producing output and queuing it to the heap for
196-- sequencing. The heap can be drained only by one thread at a time, any thread
197-- that finds that heap can be drained now, takes a lock and starts draining
198-- it, however the thread may get prempted in the middle of it holding the
199-- lock. Since that thread is holding the lock, the other threads cannot pick
200-- up the draining task, therefore they proceed to picking up the next task to
201-- execute. If the draining thread could yield voluntarily at a point where it
202-- has released the lock, then the next threads could pick up the draining
203-- instead of executing more tasks. When there are 100,000 threads the drainer
204-- gets a cpu share to run only 1:100000 of the time. This makes the heap
205-- accumulate a lot of output when we the buffer size is large.
207-- The solutions to this problem are:
208-- 1) make the other threads wait in a queue until the draining finishes
209-- 2) make the other threads queue and go away if draining is in progress
211-- In both cases we give the drainer a chance to run more often.
213processHeap :: MonadIO m
214 => IORef ([Stream m a], Int)
215 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)
219 -> AheadHeapEntry Stream m a
221 -> Bool -- we are draining the heap before we stop
223processHeap q heap st sv winfo entry sno stopping = loopHeap sno entry
227 stopIfNeeded ent seqNo r = do
228 stopIt <- liftIO $ preStopCheck sv heap
231 -- put the entry back in the heap and stop
232 requeueOnHeapTop heap (Entry seqNo ent) seqNo
234 else runStreamWithYieldLimit True seqNo r
238 AheadEntryNull -> nextHeap seqNo
239 AheadEntryPure a -> do
240 -- Use 'send' directly so that we do not account this in worker
241 -- latency as this will not be the real latency.
242 -- Don't stop the worker in this case as we are just
243 -- transferring available results from heap to outputQueue.
244 void $ liftIO $ send sv (ChildYield a)
246 AheadEntryStream r ->
248 then stopIfNeeded ent seqNo r
249 else runStreamWithYieldLimit True seqNo r
251 nextHeap prevSeqNo = do
252 res <- liftIO $ dequeueFromHeapSeq heap (prevSeqNo + 1)
254 Ready (Entry seqNo hent) -> loopHeap seqNo hent
255 Clearing -> liftIO $ sendStop sv winfo
259 r <- liftIO $ preStopCheck sv heap
261 then liftIO $ sendStop sv winfo
262 else processWorkQueue prevSeqNo
263 else inline processWorkQueue prevSeqNo
265 processWorkQueue prevSeqNo = do
266 work <- dequeueAhead q
268 Nothing -> liftIO $ sendStop sv winfo
269 Just (m, seqNo) -> do
270 yieldLimitOk <- liftIO $ decrementYieldLimit sv
273 if seqNo == prevSeqNo + 1
274 then processWithToken q heap st sv winfo m seqNo
275 else processWithoutToken q heap st sv winfo m seqNo
276 else liftIO $ abortExecution q sv winfo m
278 -- We do not stop the worker on buffer full here as we want to proceed to
279 -- nextHeap anyway so that we can clear any subsequent entries. We stop
280 -- only in yield continuation where we may have a remaining stream to be
281 -- pushed on the heap.
282 singleStreamFromHeap seqNo a = do
283 void $ liftIO $ sendYield sv winfo (ChildYield a)
286 -- XXX when we have an unfinished stream on the heap we cannot account all
287 -- the yields of that stream until it finishes, so if we have picked up
288 -- and executed more actions beyond that in the parent stream and put them
289 -- on the heap then they would eat up some yield limit which is not
290 -- correct, we will think that our yield limit is over even though we have
291 -- to yield items from unfinished stream before them. For this reason, if
292 -- there are pending items in the heap we drain them unconditionally
293 -- without considering the yield limit.
294 runStreamWithYieldLimit continue seqNo r = do
295 _ <- liftIO $ decrementYieldLimit sv
296 if continue -- see comment above -- && yieldLimitOk
299 liftIO (incrementYieldLimit sv)
302 (yieldStreamFromHeap seqNo)
303 (singleStreamFromHeap seqNo)
307 let ent = Entry seqNo (AheadEntryStream r)
308 liftIO $ requeueOnHeapTop heap ent seqNo
309 incrementYieldLimit sv
312 yieldStreamFromHeap seqNo a r = do
313 continue <- liftIO $ sendYield sv winfo (ChildYield a)
314 runStreamWithYieldLimit continue seqNo r
316{-# NOINLINE drainHeap #-}
317drainHeap :: MonadIO m
318 => IORef ([Stream m a], Int)
319 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)
324drainHeap q heap st sv winfo = do
325 r <- liftIO $ dequeueFromHeap heap
327 Ready (Entry seqNo hent) ->
328 processHeap q heap st sv winfo hent seqNo True
329 _ -> liftIO $ sendStop sv winfo
331data HeapStatus = HContinue | HStop
333processWithoutToken :: MonadIO m
334 => IORef ([Stream m a], Int)
335 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)
342processWithoutToken q heap st sv winfo m seqNo = do
343 -- we have already decremented the yield limit for m
345 liftIO (incrementYieldLimit sv)
346 -- If the stream stops without yielding anything, and we do not put
347 -- anything on heap, but if heap was waiting for this seq number
348 -- then it will keep waiting forever, because we are never going to
349 -- put it on heap. So we have to put a null entry on heap even when
351 toHeap AheadEntryNull
354 (\a r -> toHeap $ AheadEntryStream $ K.cons a r)
355 (toHeap . AheadEntryPure)
361 -- XXX to reduce contention each CPU can have its own heap
363 -- Heap insertion is an expensive affair so we use a non CAS based
364 -- modification, otherwise contention and retries can make a thread
365 -- context switch and throw it behind other threads which come later in
367 newHp <- liftIO $ atomicModifyIORef heap $ \(hp, snum) ->
368 let hp' = H.insert (Entry seqNo ent) hp
369 in assert (heapIsSane snum seqNo) ((hp', snum), hp')
371 when (svarInspectMode sv) $
373 maxHp <- readIORef (maxHeapSize $ svarStats sv)
374 when (H.size newHp > maxHp) $
375 writeIORef (maxHeapSize $ svarStats sv) (H.size newHp)
377 heapOk <- liftIO $ underMaxHeap sv newHp
378 let drainAndStop = drainHeap q heap st sv winfo
379 mainLoop = workLoopAhead q heap st sv winfo
381 case yieldRateInfo sv of
382 Nothing -> return HContinue
386 rateOk <- liftIO $ workerRateControl sv yinfo info
388 then return HContinue
390 Nothing -> return HContinue
395 HContinue -> mainLoop
396 HStop -> drainAndStop
399processWithToken :: MonadIO m
400 => IORef ([Stream m a], Int)
401 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)
408processWithToken q heap st sv winfo action sno = do
409 -- Note, we enter this function with yield limit already decremented
410 -- XXX deduplicate stop in all invocations
412 liftIO (incrementYieldLimit sv)
413 loopWithToken (sno + 1)
415 foldStreamSVar sv st (yieldOutput sno) (singleOutput sno) stop action
419 singleOutput seqNo a = do
420 continue <- liftIO $ sendYield sv winfo (ChildYield a)
422 then loopWithToken (seqNo + 1)
424 liftIO $ updateHeapSeq heap (seqNo + 1)
425 drainHeap q heap st sv winfo
427 -- XXX use a wrapper function around stop so that we never miss
428 -- incrementing the yield in a stop continuation. Essentiatlly all
429 -- "unstream" calls in this function must increment yield limit on stop.
430 yieldOutput seqNo a r = do
431 continue <- liftIO $ sendYield sv winfo (ChildYield a)
432 yieldLimitOk <- liftIO $ decrementYieldLimit sv
433 if continue && yieldLimitOk
436 liftIO (incrementYieldLimit sv)
437 loopWithToken (seqNo + 1)
444 let ent = Entry seqNo (AheadEntryStream r)
445 liftIO $ requeueOnHeapTop heap ent seqNo
446 liftIO $ incrementYieldLimit sv
447 drainHeap q heap st sv winfo
449 loopWithToken nextSeqNo = do
450 work <- dequeueAhead q
453 liftIO $ updateHeapSeq heap nextSeqNo
454 workLoopAhead q heap st sv winfo
456 Just (m, seqNo) -> do
457 yieldLimitOk <- liftIO $ decrementYieldLimit sv
458 let undo = liftIO $ do
459 updateHeapSeq heap nextSeqNo
460 reEnqueueAhead sv q m
461 incrementYieldLimit sv
464 if seqNo == nextSeqNo
467 liftIO (incrementYieldLimit sv)
468 loopWithToken (seqNo + 1)
475 -- To avoid a race when another thread puts something
476 -- on the heap and goes away, the consumer will not get
477 -- a doorBell and we will not clear the heap before
478 -- executing the next action. If the consumer depends
479 -- on the output that is stuck in the heap then this
480 -- will result in a deadlock. So we always clear the
481 -- heap before executing the next action.
482 undo >> workLoopAhead q heap st sv winfo
483 else undo >> drainHeap q heap st sv winfo
485-- XXX the yield limit changes increased the performance overhead by 30-40%.
486-- Just like AsyncT we can use an implementation without yeidlimit and even
487-- without pacing code to keep the performance higher in the unlimited and
490-- XXX The yieldLimit stuff is pretty invasive. We can instead do it by using
491-- three hooks, a pre-execute hook, a yield hook and a stop hook. In fact these
492-- hooks can be used for a more general implementation to even check predicates
493-- and not just yield limit.
495-- XXX we can remove the sv parameter as it can be derived from st
497workLoopAhead :: MonadIO m
498 => IORef ([Stream m a], Int)
499 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)
504workLoopAhead q heap st sv winfo = do
505 r <- liftIO $ dequeueFromHeap heap
507 Ready (Entry seqNo hent) ->
508 processHeap q heap st sv winfo hent seqNo False
509 Clearing -> liftIO $ sendStop sv winfo
511 -- Before we execute the next item from the work queue we check
512 -- if we are beyond the yield limit. It is better to check the
513 -- yield limit before we pick up the next item. Otherwise we
514 -- may have already started more tasks even though we may have
515 -- reached the yield limit. We can avoid this by taking active
516 -- workers into account, but that is not as reliable, because
517 -- workers may go away without picking up work and yielding a
520 -- Rate control can be done either based on actual yields in
521 -- the output queue or based on any yield either to the heap or
522 -- to the output queue. In both cases we may have one issue or
523 -- the other. We chose to do this based on actual yields to the
524 -- output queue because it makes the code common to both async
525 -- and ahead streams.
527 work <- dequeueAhead q
529 Nothing -> liftIO $ sendStop sv winfo
530 Just (m, seqNo) -> do
531 yieldLimitOk <- liftIO $ decrementYieldLimit sv
535 then processWithToken q heap st sv winfo m seqNo
536 else processWithoutToken q heap st sv winfo m seqNo
537 -- If some worker decremented the yield limit but then
538 -- did not yield anything and therefore incremented it
539 -- later, then if we did not requeue m here we may find
540 -- the work queue empty and therefore miss executing
541 -- the remaining action.
542 else liftIO $ abortExecution q sv winfo m
544-------------------------------------------------------------------------------
546-------------------------------------------------------------------------------
548-- XXX To be implemented. Use a linked queue like WAsync and put back the
549-- remaining computation at the back of the queue instead of the heap, and
550-- increment the sequence number.
552-- The only difference between forkSVarAsync and this is that we run the left
553-- computation without a shared SVar.
554forkSVarAhead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
555forkSVarAhead m1 m2 = mkStream $ \st stp sng yld -> do
556 sv <- newAheadVar st (concurrently (toStream m1) (toStream m2))
558 foldStream st stp sng yld (fromSVar sv)
560 concurrently ma mb = mkStream $ \st stp sng yld -> do
561 liftIO $ enqueue (fromJust $ streamVar st) mb
562 foldStream st stp sng yld ma
564-- | Polymorphic version of the 'Semigroup' operation '<>' of 'AheadT'.
565-- Merges two streams sequentially but with concurrent lookahead.
569ahead :: (IsStream t, MonadAsync m) => t m a -> t m a -> t m a
570ahead m1 m2 = mkStream $ \st stp sng yld ->
572 Just sv | svarStyle sv == AheadVar -> do
573 liftIO $ enqueue sv (toStream m2)
574 -- Always run the left side on a new SVar to avoid complexity in
575 -- sequencing results. This means the left side cannot further
576 -- split into more ahead computations on the same SVar.
577 foldStream st stp sng yld m1
578 _ -> foldStreamShared st stp sng yld (forkSVarAhead m1 m2)
580-- | XXX we can implement it more efficienty by directly implementing instead
581-- of combining streams using ahead.
582{-# INLINE consMAhead #-}
583{-# SPECIALIZE consMAhead :: IO a -> AheadT IO a -> AheadT IO a #-}
584consMAhead :: MonadAsync m => m a -> AheadT m a -> AheadT m a
585consMAhead m r = fromStream $ K.yieldM m `ahead` (toStream r)
587------------------------------------------------------------------------------
589------------------------------------------------------------------------------
591-- | Deep ahead composition or ahead composition with depth first traversal.
592-- The semigroup composition of 'AheadT' appends streams in a depth first
593-- manner just like 'SerialT' except that it can produce elements concurrently
594-- ahead of time. It is like 'AsyncT' except that 'AsyncT' produces the output
595-- as it arrives whereas 'AheadT' orders the output in the traversal order.
598-- main = ('toList' . 'aheadly' $ (fromFoldable [1,2]) \<> (fromFoldable [3,4])) >>= print
604-- Any exceptions generated by a constituent stream are propagated to the
607-- Similarly, the monad instance of 'AheadT' may run each iteration
608-- concurrently ahead of time but presents the results in the same order as
613-- import qualified "Streamly.Prelude" as S
614-- import Control.Concurrent
616-- main = 'drain' . 'aheadly' $ do
617-- n <- return 3 \<\> return 2 \<\> return 1
619-- threadDelay (n * 1000000)
620-- myThreadId >>= \\tid -> putStrLn (show tid ++ ": Delay " ++ show n)
623-- ThreadId 40: Delay 1
624-- ThreadId 39: Delay 2
625-- ThreadId 38: Delay 3
628-- All iterations may run in the same thread if they do not block.
630-- Note that ahead composition with depth first traversal can be used to
631-- combine infinite number of streams as it explores only a bounded number of
635newtype AheadT m a = AheadT {getAheadT :: Stream m a}
636 deriving (MonadTrans)
638-- | A serial IO stream of elements of type @a@ with concurrent lookahead. See
639-- 'AheadT' documentation for more details.
642type Ahead = AheadT IO
644-- | Fix the type of a polymorphic stream as 'AheadT'.
647aheadly :: IsStream t => AheadT m a -> t m a
650instance IsStream AheadT where
656------------------------------------------------------------------------------
658------------------------------------------------------------------------------
660{-# INLINE mappendAhead #-}
661{-# SPECIALIZE mappendAhead :: AheadT IO a -> AheadT IO a -> AheadT IO a #-}
662mappendAhead :: MonadAsync m => AheadT m a -> AheadT m a -> AheadT m a
663mappendAhead m1 m2 = fromStream $ ahead (toStream m1) (toStream m2)
665instance MonadAsync m => Semigroup (AheadT m a) where
668------------------------------------------------------------------------------
670------------------------------------------------------------------------------
672instance MonadAsync m => Monoid (AheadT m a) where
676------------------------------------------------------------------------------
678------------------------------------------------------------------------------
680{-# INLINE concatMapAhead #-}
681{-# SPECIALIZE concatMapAhead :: (a -> AheadT IO b) -> AheadT IO a -> AheadT IO b #-}
682concatMapAhead :: MonadAsync m => (a -> AheadT m b) -> AheadT m a -> AheadT m b
683concatMapAhead f m = fromStream $
684 K.concatMapBy ahead (\a -> K.adapt $ f a) (K.adapt m)
686instance MonadAsync m => Monad (AheadT m) where
689 (>>=) = flip concatMapAhead
691instance (Monad m, MonadAsync m) => Applicative (AheadT m) where
692 pure = AheadT . K.yield
696------------------------------------------------------------------------------
698------------------------------------------------------------------------------
700MONAD_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