git.iko.soy/streamly
on master help sign in
Add correct impl of takeInterval & dropInterval to Stream.Parallel

- Remove the incorrect implementation of from IsStream - Remove takeByTime and dropByTime from StreamD

commitba4d23b489eccbebfbd2f1facb9cbebcd1dff689 authorAdithya Kumar <adithya@composewell.com> dateFri 16 Sep 2022, 17:04 (4 years ago) parent99846a8
4 files changed+131−165
compare with master
unifiedsplit
benchmark/Streamly/Benchmark/Data/Stream/Transform.hs+12−17 core/src/Streamly/Internal/Data/Stream/StreamD/Transform.hs+1−84 src/Streamly/Internal/Data/Stream/IsStream/Transform.hs+2−42 src/Streamly/Internal/Data/Stream/Parallel.hs+116−22
benchmark/Streamly/Benchmark/Data/Stream/Transform.hs+12−17
@@ -35,6 +35,7 @@ import qualified Streamly.Internal.Data.Unfold as Unfold
3535import qualified Streamly.Internal.Data.Stream.IsStream as Stream
3636import Streamly.Internal.Data.Time.Units
3737#else
38+import qualified Streamly.Internal.Data.Stream.Parallel as Stream
3839import qualified Streamly.Internal.Data.Stream as Stream
3940#endif
4041
@@ -301,17 +302,17 @@ takeWhileTrue value n = composeN n $ Stream.takeWhile (<= (value + 1))
301302takeWhileMTrue :: MonadIO m => Int -> Int -> Stream m Int -> m ()
302303takeWhileMTrue value n = composeN n $ Stream.takeWhileM (return . (<= (value + 1)))
303304
304−#ifdef USE_PRELUDE
305305{-# INLINE takeInterval #-}
306−takeInterval :: NanoSecond64 -> Int -> Stream IO Int -> IO ()
306+takeInterval :: Double -> Int -> Stream IO Int -> IO ()
307307takeInterval i n = composeN n (Stream.takeInterval i)
308308
309+-- Inspection testing is disabled for takeInterval
310+-- Enable it when looking at it throughly
309311#ifdef INSPECTION
310312-- inspect $ hasNoType 'takeInterval ''SPEC
311−inspect $ hasNoTypeClasses 'takeInterval
313+-- inspect $ hasNoTypeClasses 'takeInterval
312314-- inspect $ 'takeInterval `hasNoType` ''D.Step
313315#endif
314−#endif
315316
316317{-# INLINE dropOne #-}
317318dropOne :: MonadIO m => Int -> Stream m Int -> m ()
@@ -338,16 +339,18 @@ dropWhileFalse value n = composeN n $ Stream.dropWhile (> (value + 1))
338339{-# INLINE _intervalsOfSum #-}
339340_intervalsOfSum :: MonadAsync m => Double -> Int -> Stream m Int -> m ()
340341_intervalsOfSum i n = composeN n (Stream.intervalsOf i FL.sum)
342+#endif
341343
342344{-# INLINE dropInterval #-}
343−dropInterval :: NanoSecond64 -> Int -> Stream IO Int -> IO ()
345+dropInterval :: Double -> Int -> Stream IO Int -> IO ()
344346dropInterval i n = composeN n (Stream.dropInterval i)
345347
348+-- Inspection testing is disabled for dropInterval
349+-- Enable it when looking at it throughly
346350#ifdef INSPECTION
347−inspect $ hasNoTypeClasses 'dropInterval
351+-- inspect $ hasNoTypeClasses 'dropInterval
348352-- inspect $ 'dropInterval `hasNoType` ''D.Step
349353#endif
350−#endif
351354
352355{-# INLINE findIndices #-}
353356findIndices :: MonadIO m => Int -> Int -> Stream m Int -> m ()
@@ -403,16 +406,8 @@ o_1_space_filtering value =
403406 -- , benchIOSink value "takeWhileM-true" (_takeWhileMTrue value 1)
404407 , benchIOSink value "drop-one" (dropOne 1)
405408 , benchIOSink value "drop-all" (dropAll value 1)
406−#ifdef USE_PRELUDE
407− , benchIOSink
408− value
409− "takeInterval-all"
410− (takeInterval (NanoSecond64 maxBound) 1)
411− , benchIOSink
412− value
413− "dropInterval-all"
414− (dropInterval (NanoSecond64 maxBound) 1)
415−#endif
409+ , benchIOSink value "takeInterval-all" (takeInterval 10000 1)
410+ , benchIOSink value "dropInterval-all" (dropInterval 10000 1)
416411 , benchIOSink value "dropWhile-true" (dropWhileTrue value 1)
417412 -- , benchIOSink value "dropWhileM-true" (_dropWhileMTrue value 1)
418413 , benchIOSink
core/src/Streamly/Internal/Data/Stream/StreamD/Transform.hs+1−84
@@ -78,11 +78,9 @@ module Streamly.Internal.Data.Stream.StreamD.Transform
7878 -- * Trimming
7979 -- | Produce a subset of the stream trimmed at ends.
8080 , take
81− , takeByTime
8281 , takeWhile
8382 , takeWhileM
8483 , drop
85− , dropByTime
8684 , dropWhile
8785 , dropWhileM
8886
@@ -133,17 +131,14 @@ import Control.Monad.Trans.Class (MonadTrans(lift))
133131import Data.Maybe (fromJust, isJust)
134132import Fusion.Plugin.Types (Fuse(..))
135133import GHC.Types (SPEC(..))
136−import qualified Data.Set as Set
137134
138135import Streamly.Internal.Control.Concurrent (MonadAsync)
139136import Streamly.Internal.Control.ForkLifted (forkManaged)
140137import Streamly.Internal.Data.Fold.Type (Fold(..))
141138import Streamly.Internal.Data.Pipe.Type (Pipe(..), PipeState(..))
142139import Streamly.Internal.Data.SVar.Type (defState, adaptState)
143−import Streamly.Internal.Data.Time.Clock (Clock(Monotonic), getTime)
144−import Streamly.Internal.Data.Time.Units
145− (TimeUnit64, toRelTime64, diffAbsTime64)
146140
141+import qualified Data.Set as Set
147142import qualified Streamly.Internal.Data.Fold.Type as FL
148143import qualified Streamly.Internal.Data.IORef.Unboxed as Unboxed
149144import qualified Streamly.Internal.Data.Pipe.Type as Pipe
@@ -791,84 +786,6 @@ deleteBy eq x (Stream step state) = Stream step' (state, False)
791786-- Trimming
792787------------------------------------------------------------------------------
793788
794−-- XXX using getTime in the loop can be pretty expensive especially for
795−-- computations where iterations are lightweight. We have the following
796−-- options:
797−--
798−-- 1) Run a timeout thread updating a flag asynchronously and check that
799−-- flag here, that way we can have a cheap termination check.
800−--
801−-- 2) Use COARSE clock to get time with lower resolution but more efficiently.
802−--
803−-- 3) Use rdtscp/rdtsc to get time directly from the processor, compute the
804−-- termination value of rdtsc in the beginning and then in each iteration just
805−-- get rdtsc and check if we should terminate.
806−--
807−data TakeByTime st s
808− = TakeByTimeInit st
809− | TakeByTimeCheck st s
810− | TakeByTimeYield st s
811−
812−{-# INLINE_NORMAL takeByTime #-}
813−takeByTime :: (MonadIO m, TimeUnit64 t) => t -> Stream m a -> Stream m a
814−takeByTime duration (Stream step1 state1) = Stream step (TakeByTimeInit state1)
815− where
816−
817− lim = toRelTime64 duration
818−
819− {-# INLINE_LATE step #-}
820− step _ (TakeByTimeInit _) | lim == 0 = return Stop
821− step _ (TakeByTimeInit st) = do
822− t0 <- liftIO $ getTime Monotonic
823− return $ Skip (TakeByTimeYield st t0)
824− step _ (TakeByTimeCheck st t0) = do
825− t <- liftIO $ getTime Monotonic
826− return $
827− if diffAbsTime64 t t0 > lim
828− then Stop
829− else Skip (TakeByTimeYield st t0)
830− step gst (TakeByTimeYield st t0) = do
831− r <- step1 gst st
832− return $ case r of
833− Yield x s -> Yield x (TakeByTimeCheck s t0)
834− Skip s -> Skip (TakeByTimeCheck s t0)
835− Stop -> Stop
836−
837−data DropByTime st s x
838− = DropByTimeInit st
839− | DropByTimeGen st s
840− | DropByTimeCheck st s x
841− | DropByTimeYield st
842−
843−{-# INLINE_NORMAL dropByTime #-}
844−dropByTime :: (MonadIO m, TimeUnit64 t) => t -> Stream m a -> Stream m a
845−dropByTime duration (Stream step1 state1) = Stream step (DropByTimeInit state1)
846− where
847−
848− lim = toRelTime64 duration
849−
850− {-# INLINE_LATE step #-}
851− step _ (DropByTimeInit st) = do
852− t0 <- liftIO $ getTime Monotonic
853− return $ Skip (DropByTimeGen st t0)
854− step gst (DropByTimeGen st t0) = do
855− r <- step1 gst st
856− return $ case r of
857− Yield x s -> Skip (DropByTimeCheck s t0 x)
858− Skip s -> Skip (DropByTimeGen s t0)
859− Stop -> Stop
860− step _ (DropByTimeCheck st t0 x) = do
861− t <- liftIO $ getTime Monotonic
862− if diffAbsTime64 t t0 <= lim
863− then return $ Skip $ DropByTimeGen st t0
864− else return $ Yield x $ DropByTimeYield st
865− step gst (DropByTimeYield st) = do
866− r <- step1 gst st
867− return $ case r of
868− Yield x s -> Yield x (DropByTimeYield s)
869− Skip s -> Skip (DropByTimeYield s)
870− Stop -> Stop
871−
872789-- Adapted from the vector package
873790{-# INLINE_NORMAL drop #-}
874791drop :: Monad m => Int -> Stream m a -> Stream m a
src/Streamly/Internal/Data/Stream/IsStream/Transform.hs+2−42
@@ -85,7 +85,6 @@ module Streamly.Internal.Data.Stream.IsStream.Transform
8585 -- | Produce a subset of the stream trimmed at ends.
8686
8787 , take
88− , takeInterval
8988 , takeLast
9089 , takeLastInterval
9190 , takeWhile
@@ -93,7 +92,6 @@ module Streamly.Internal.Data.Stream.IsStream.Transform
9392 , takeWhileLast
9493 , takeWhileAround
9594 , drop
96− , dropInterval
9795 , dropLast
9896 , dropLastInterval
9997 , dropWhile
@@ -250,7 +248,7 @@ import Streamly.Internal.Data.Stream.IsStream.Type
250248 (IsStream(..), fromStreamD, toStreamD, toConsK)
251249import Streamly.Internal.Data.Stream.Serial (SerialT)
252250import Streamly.Internal.Data.SVar (Rate(..))
253−import Streamly.Internal.Data.Time.Units (TimeUnit64, AbsTime, RelTime64)
251+import Streamly.Internal.Data.Time.Units (AbsTime, RelTime64)
254252
255253import qualified Streamly.Data.Fold as FL
256254import qualified Streamly.Internal.Data.Stream.Parallel as Par
@@ -259,7 +257,7 @@ import qualified Streamly.Internal.Data.Stream.StreamD as D
259257 (transform, foldrT, tap, tapOffsetEvery, pollCounts, mapM, scanOnce
260258 , scanMany, postscanOnce, scanlx', scanlM', scanl', postscanl', prescanl'
261259 , prescanlM', scanl1M', scanl1', filter, filterM, uniq, deleteBy, takeWhileM
262− , takeByTime, dropWhile, dropWhileM, dropByTime, insertBy, intersperse
260+ , dropWhile, dropWhileM, insertBy, intersperse
263261 , intersperseM_, intersperseSuffix, intersperseSuffix_
264262 , intersperseSuffixBySpan, indexed, indexedR, rollingMap, rollingMapM
265263 , rollingMap2, mapMaybe, mapMaybeM)
@@ -1059,26 +1057,6 @@ takeWhileAround :: -- (IsStream t, Monad m) =>
10591057 (a -> Bool) -> t m a -> t m a
10601058takeWhileAround = undefined -- fromStreamD $ D.takeWhileAround n $ toStreamD m
10611059
1062−-- | @takeInterval duration@ yields stream elements upto specified time
1063−-- @duration@. The duration starts when the stream is evaluated for the first
1064−-- time, before the first element is yielded. The time duration is checked
1065−-- before generating each element, if the duration has expired the stream
1066−-- stops.
1067−--
1068−-- The total time taken in executing the stream is guaranteed to be /at least/
1069−-- @duration@, however, because the duration is checked before generating an
1070−-- element, the upper bound is indeterminate and depends on the time taken in
1071−-- generating and processing the last element.
1072−--
1073−-- No element is yielded if the duration is zero. At least one element is
1074−-- yielded if the duration is non-zero.
1075−--
1076−-- /Pre-release/
1077−--
1078−{-# INLINE takeInterval #-}
1079−takeInterval ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
1080−takeInterval d = fromStreamD . D.takeByTime d . toStreamD
1081−
10821060-- | Drop elements in the stream as long as the predicate succeeds and then
10831061-- take the rest of the stream.
10841062--
@@ -1094,24 +1072,6 @@ dropWhile p m = fromStreamD $ D.dropWhile p $ toStreamD m
10941072dropWhileM :: (IsStream t, Monad m) => (a -> m Bool) -> t m a -> t m a
10951073dropWhileM p m = fromStreamD $ D.dropWhileM p $ toStreamD m
10961074
1097−-- | @dropInterval duration@ drops stream elements until specified @duration@ has
1098−-- passed. The duration begins when the stream is evaluated for the first
1099−-- time. The time duration is checked /after/ generating a stream element, the
1100−-- element is yielded if the duration has expired otherwise it is dropped.
1101−--
1102−-- The time elapsed before starting to generate the first element is /at most/
1103−-- @duration@, however, because the duration expiry is checked after the
1104−-- element is generated, the lower bound is indeterminate and depends on the
1105−-- time taken in generating an element.
1106−--
1107−-- All elements are yielded if the duration is zero.
1108−--
1109−-- /Pre-release/
1110−--
1111−{-# INLINE dropInterval #-}
1112−dropInterval ::(MonadIO m, IsStream t, TimeUnit64 d) => d -> t m a -> t m a
1113−dropInterval d = fromStreamD . D.dropByTime d . toStreamD
1114−
11151075-- | Drop @n@ elements at the end of the stream.
11161076--
11171077-- O(n) space, where n is the number elements dropped.
src/Streamly/Internal/Data/Stream/Parallel.hs+116−22
@@ -11,7 +11,7 @@
1111--
1212-- To run examples in this module:
1313--
14−-- >>> import qualified Streamly.Prelude as Stream
14+-- >>> import qualified Streamly.Data.Stream as Stream
1515-- >>> import Control.Concurrent (threadDelay)
1616-- >>> :{
1717-- delay n = do
@@ -42,10 +42,15 @@ module Streamly.Internal.Data.Stream.Parallel
4242
4343 -- * Callbacks
4444 , newCallbackStream
45+
46+ -- * Combinators
47+ , interjectSuffix
48+ , takeInterval
49+ , dropInterval
4550 )
4651where
4752
48−import Control.Concurrent (myThreadId, takeMVar)
53+import Control.Concurrent (myThreadId, takeMVar, threadDelay)
4954import Control.Monad (when)
5055import Control.Monad.Base (MonadBase(..), liftBaseDefault)
5156import Control.Monad.Catch (MonadThrow, throwM)
@@ -56,37 +61,36 @@ import Control.Monad.State.Class (MonadState(..))
5661import Control.Monad.Trans.Class (MonadTrans(lift))
5762import Data.Functor (void)
5863import Data.IORef (readIORef, writeIORef)
59−import Data.Maybe (fromJust)
64+import Data.Maybe (fromJust, isNothing)
6065#if __GLASGOW_HASKELL__ < 808
6166import Data.Semigroup (Semigroup(..))
6267#endif
63−import Prelude hiding (map)
64−
65−import qualified Data.Set as Set
6668
6769import Streamly.Data.Fold (Fold)
6870import Streamly.Internal.Control.Concurrent (MonadAsync)
6971import Streamly.Internal.Data.Stream.StreamD.Type (Step(..))
70−import Streamly.Internal.Data.Stream.StreamK.Type (Stream)
72+import Streamly.Internal.Data.Stream.Type (Stream)
7173
74+import qualified Data.Set as Set
7275import qualified Streamly.Internal.Data.Stream.StreamK.Type as K
73− (foldStreamShared, mkStream, foldStream, fromEffect
76+ (Stream, foldStreamShared, mkStream, foldStream, fromEffect
7477 , nil, concatMapWith, fromPure, bindWith, withLocal)
7578import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
7679 (Stream(..), mapM, toStreamK, fromStreamK)
7780import qualified Streamly.Internal.Data.Stream.SVar.Generate as SVar
7881import qualified Streamly.Internal.Data.Stream.SVar.Eliminate as SVar
7982import qualified Streamly.Internal.Data.Stream as Stream
80− (toStreamK, fromStreamK)
83+ (catMaybes, dropWhile, fromStreamK, repeat, sequence, takeWhile, toStreamK)
8184
8285import Streamly.Internal.Data.SVar
86+import Prelude hiding (map)
8387
8488#include "inline.hs"
8589#include "Instances.hs"
8690
8791--
8892-- $setup
89−-- >>> import qualified Streamly.Prelude as Stream
93+-- >>> import qualified Streamly.Data.Stream as Stream
9094-- >>> import Control.Concurrent (threadDelay)
9195-- >>> :{
9296-- delay n = do
@@ -106,7 +110,7 @@ import Streamly.Internal.Data.SVar
106110{-# NOINLINE runOne #-}
107111runOne
108112 :: MonadIO m
109− => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
113+ => State K.Stream m a -> K.Stream m a -> Maybe WorkerInfo -> m ()
110114runOne st m0 winfo =
111115 case getYieldLimit st of
112116 Nothing -> go m0
@@ -129,7 +133,7 @@ runOne st m0 winfo =
129133
130134runOneLimited
131135 :: MonadIO m
132− => State Stream m a -> Stream m a -> Maybe WorkerInfo -> m ()
136+ => State K.Stream m a -> K.Stream m a -> Maybe WorkerInfo -> m ()
133137runOneLimited st m0 winfo = go m0
134138
135139 where
@@ -166,7 +170,7 @@ runOneLimited st m0 winfo = go m0
166170
167171{-# NOINLINE forkSVarPar #-}
168172forkSVarPar :: MonadAsync m
169− => SVarStopStyle -> Stream m a -> Stream m a -> Stream m a
173+ => SVarStopStyle -> K.Stream m a -> K.Stream m a -> K.Stream m a
170174forkSVarPar ss m r = K.mkStream $ \st yld sng stp -> do
171175 sv <- newParallelVar ss st
172176 pushWorkerPar sv (runOne st{streamVar = Just sv} m)
@@ -179,8 +183,13 @@ forkSVarPar ss m r = K.mkStream $ \st yld sng stp -> do
179183 K.foldStream st yld sng stp $ Stream.toStreamK (SVar.fromSVar sv)
180184
181185{-# INLINE joinStreamVarPar #-}
182−joinStreamVarPar :: MonadAsync m
183− => SVarStyle -> SVarStopStyle -> Stream m a -> Stream m a -> Stream m a
186+joinStreamVarPar ::
187+ MonadAsync m
188+ => SVarStyle
189+ -> SVarStopStyle
190+ -> K.Stream m a
191+ -> K.Stream m a
192+ -> K.Stream m a
184193joinStreamVarPar style ss m1 m2 = K.mkStream $ \st yld sng stp ->
185194 case streamVar st of
186195 Just sv | svarStyle sv == style && svarStopStyle sv == ss -> do
@@ -230,7 +239,7 @@ joinStreamVarPar style ss m1 m2 = K.mkStream $ \st yld sng stp ->
230239-------------------------------------------------------------------------------
231240
232241{-# INLINE parallelK #-}
233−parallelK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
242+parallelK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a
234243parallelK = joinStreamVarPar ParallelVar StopNone
235244
236245-- | XXX we can implement it more efficienty by directly implementing instead
@@ -248,7 +257,7 @@ consM m (ParallelT r) = ParallelT $ parallelK (K.fromEffect m) r
248257--
249258-- /Pre-release/
250259{-# INLINE parallelFstK #-}
251−parallelFstK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
260+parallelFstK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a
252261parallelFstK = joinStreamVarPar ParallelVar StopBy
253262
254263-- This is a race like combinator for streams.
@@ -258,7 +267,7 @@ parallelFstK = joinStreamVarPar ParallelVar StopBy
258267--
259268-- /Pre-release/
260269{-# INLINE parallelMinK #-}
261−parallelMinK :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
270+parallelMinK :: MonadAsync m => K.Stream m a -> K.Stream m a -> K.Stream m a
262271parallelMinK = joinStreamVarPar ParallelVar StopAny
263272
264273------------------------------------------------------------------------------
@@ -269,7 +278,7 @@ parallelMinK = joinStreamVarPar ParallelVar StopAny
269278--
270279-- /Pre-release/
271280--
272−mkParallelK :: MonadAsync m => Stream m a -> Stream m a
281+mkParallelK :: MonadAsync m => K.Stream m a -> K.Stream m a
273282mkParallelK m = K.mkStream $ \st yld sng stp -> do
274283 sv <- newParallelVar StopNone (adaptState st)
275284 -- pushWorkerPar sv (runOne st{streamVar = Just sv} $ toStream m)
@@ -342,7 +351,8 @@ mkParallelD m = D.Stream step Nothing
342351--
343352-- /Pre-release/
344353{-# INLINE tapAsyncK #-}
345−tapAsyncK :: MonadAsync m => (Stream m a -> m b) -> Stream m a -> Stream m a
354+tapAsyncK ::
355+ MonadAsync m => (K.Stream m a -> m b) -> K.Stream m a -> K.Stream m a
346356tapAsyncK f m = K.mkStream $ \st yld sng stp -> do
347357 sv <- SVar.newFoldSVar st (f . Stream.toStreamK)
348358 K.foldStreamShared st yld sng stp
@@ -420,7 +430,7 @@ tapAsyncF f (D.Stream step1 state1) = D.Stream step TapInit
420430-- /Since: 0.7.0 (maxBuffer applies to ParallelT streams)/
421431--
422432-- @since 0.8.0
423−newtype ParallelT m a = ParallelT {getParallelT :: Stream m a}
433+newtype ParallelT m a = ParallelT {getParallelT :: K.Stream m a}
424434 deriving (MonadTrans)
425435
426436-- | A parallely composing IO stream of elements of type @a@.
@@ -507,7 +517,7 @@ MONAD_COMMON_INSTANCES(ParallelT, MONADPARALLEL)
507517-- /Pre-release/
508518--
509519{-# INLINE_NORMAL newCallbackStream #-}
510−newCallbackStream :: MonadAsync m => m (a -> m (), Stream m a)
520+newCallbackStream :: MonadAsync m => m (a -> m (), K.Stream m a)
511521newCallbackStream = do
512522 sv <- newParallelVar StopNone defState
513523
@@ -520,3 +530,87 @@ newCallbackStream = do
520530 -- XXX we can return an SVar and then the consumer can unfold from the
521531 -- SVar?
522532 return (callback, D.toStreamK (SVar.fromSVarD sv))
533+
534+------------------------------------------------------------------------------
535+-- Combinators
536+------------------------------------------------------------------------------
537+
538+{-# INLINE parallelFst #-}
539+parallelFst :: MonadAsync m => Stream m a -> Stream m a -> Stream m a
540+parallelFst m1 m2 =
541+ Stream.fromStreamK
542+ $ parallelFstK (Stream.toStreamK m1) (Stream.toStreamK m2)
543+
544+-- | Intersperse a monadic action into the input stream after every @n@
545+-- seconds.
546+--
547+-- >>> import qualified Streamly.Data.Fold as Fold
548+-- >>> import qualified Streamly.Internal.Data.Stream.Parallel as Parallel
549+-- >>> Stream.fold Fold.drain $ Parallel.interjectSuffix 1.05 (putChar ',') $ Stream.mapM (\x -> threadDelay 1000000 >> putChar x) $ Stream.fromList "hello"
550+-- h,e,l,l,o
551+--
552+-- /Pre-release/
553+{-# INLINE interjectSuffix #-}
554+interjectSuffix :: MonadAsync m => Double -> m a -> Stream m a -> Stream m a
555+interjectSuffix n f xs = xs `parallelFst` repeatM timed
556+ where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
557+ repeatM = Stream.sequence . Stream.repeat
558+
559+-- XXX Notes from D.takeByTime (which was removed)
560+-- XXX using getTime in the loop can be pretty expensive especially for
561+-- computations where iterations are lightweight. We have the following
562+-- options:
563+--
564+-- 1) Run a timeout thread updating a flag asynchronously and check that
565+-- flag here, that way we can have a cheap termination check.
566+--
567+-- 2) Use COARSE clock to get time with lower resolution but more efficiently.
568+--
569+-- 3) Use rdtscp/rdtsc to get time directly from the processor, compute the
570+-- termination value of rdtsc in the beginning and then in each iteration just
571+-- get rdtsc and check if we should terminate.
572+
573+
574+-- | @takeInterval duration@ yields stream elements upto specified time
575+-- @duration@ in seconds. The duration starts when the stream is evaluated for
576+-- the first time, before the first element is yielded. The time duration is
577+-- checked before generating each element, if the duration has expired the
578+-- stream stops.
579+--
580+-- The total time taken in executing the stream is guaranteed to be /at least/
581+-- @duration@, however, because the duration is checked before generating an
582+-- element, the upper bound is indeterminate and depends on the time taken in
583+-- generating and processing the last element.
584+--
585+-- No element is yielded if the duration is zero. At least one element is
586+-- yielded if the duration is non-zero.
587+--
588+-- /Pre-release/
589+--
590+{-# INLINE takeInterval #-}
591+takeInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
592+takeInterval d =
593+ Stream.catMaybes
594+ . Stream.takeWhile isNothing
595+ . interjectSuffix d (return Nothing) . fmap Just
596+
597+-- | @dropInterval duration@ drops stream elements until specified @duration@ in
598+-- seconds has passed. The duration begins when the stream is evaluated for the
599+-- first time. The time duration is checked /after/ generating a stream element,
600+-- the element is yielded if the duration has expired otherwise it is dropped.
601+--
602+-- The time elapsed before starting to generate the first element is /at most/
603+-- @duration@, however, because the duration expiry is checked after the
604+-- element is generated, the lower bound is indeterminate and depends on the
605+-- time taken in generating an element.
606+--
607+-- All elements are yielded if the duration is zero.
608+--
609+-- /Pre-release/
610+--
611+{-# INLINE dropInterval #-}
612+dropInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
613+dropInterval d =
614+ Stream.catMaybes
615+ . Stream.dropWhile isNothing
616+ . interjectSuffix d (return Nothing) . fmap Just