@@ -5,41 +5,49 @@
5-- Maintainer : streamly@composewell.com
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
6-- Stability : experimental
7-- Portability : GHC
7-- Portability : GHC
10module Streamly.Internal.Data.Stream.Time
9module Streamly.Internal.Data.Stream.Time
12− -- Primitives
11+ -- * Imports for Examples
13− interjectSuffix
12+ -- $setup
15+ periodic -- XXX Should go to Streamly.Data.Stream
16+ , ticks -- XXX Should go to Streamly.Data.Stream
15 , takeInterval
21 , takeInterval
16 , takeLastInterval
22 , takeLastInterval
17 , dropInterval
23 , dropInterval
18 , dropLastInterval
24 , dropLastInterval
19 , intervalsOf
27 , intervalsOf
20 , chunksOfTimeout
28 , chunksOfTimeout
32+ , sampleIntervalStart
22− -- Sessions
37+ -- * Windowed Sessions
23 , classifySessionsByGeneric
38 , classifySessionsByGeneric
24 , classifySessionsBy
39 , classifySessionsBy
25 , classifySessionsOf
40 , classifySessionsOf
26 , classifyKeepAliveSessions
41 , classifyKeepAliveSessions
43+ -- XXX This should go in the concurrent module
28− -- ** Buffering and Sampling
44+ -- * Buffering
29 -- | Evaluate strictly using a buffer of results. When the buffer becomes
45 -- | Evaluate strictly using a buffer of results. When the buffer becomes
30 -- full we can block, drop the new elements, drop the oldest element and
46 -- full we can block, drop the new elements, drop the oldest element and
31− -- insert the new at the end or keep dropping elements uniformly to match
47+ -- insert the new at the end.
32− -- the rate of the consumer.
48+ , bufferLatest
33− , sampleOld
49+ , bufferLatestN
34− , sampleNew
50+ , bufferOldestN
39− , sampleIntervalStart
@@ -71,8 +79,7 @@ import qualified Streamly.Data.Unfold as Unfold
71import qualified Streamly.Internal.Data.Fold as Fold (Step(..))
79import qualified Streamly.Internal.Data.Fold as Fold (Step(..))
72import qualified Streamly.Internal.Data.IsMap as IsMap
80import qualified Streamly.Internal.Data.IsMap as IsMap
73import qualified Streamly.Internal.Data.Stream as Stream
81import qualified Streamly.Internal.Data.Stream as Stream
75− , scanlMAfter'
82+ ( scanlMAfter'
76 , timeIndexed
83 , timeIndexed
77 , timestamped
84 , timestamped
@@ -89,9 +96,9 @@ import Streamly.Internal.Data.Stream.Concurrent
89-- >>> import qualified Streamly.Data.Fold as Fold
96-- >>> import qualified Streamly.Data.Fold as Fold
90-- >>> import qualified Streamly.Data.Parser as Parser
97-- >>> import qualified Streamly.Data.Parser as Parser
91-- >>> import qualified Streamly.Data.Stream as Stream
98-- >>> import qualified Streamly.Data.Stream as Stream
92−-- >>> import qualified Streamly.Internal.Data.Stream as Stream (timestamped)
99+-- >>> import qualified Streamly.Internal.Data.Stream as Stream (delayPost, timestamped)
93-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Concur
100-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Concur
94−-- >>> import qualified Streamly.Internal.Data.Stream.Time as Concur
101+-- >>> import qualified Streamly.Internal.Data.Stream.Time as Stream
95-- >>> import Prelude hiding (concatMap, concat)
102-- >>> import Prelude hiding (concatMap, concat)
97-- delay n = do
104-- delay n = do
@@ -104,71 +111,74 @@ import Streamly.Internal.Data.Stream.Concurrent
104-- Combinators
111-- Combinators
105--------------------------------------------------------------------------------
112--------------------------------------------------------------------------------
107−-- | Intersperse a monadic action into the input stream after every @n@
114+-- | Generate a stream by running an action periodically at the specified time
108−-- seconds.
115+-- interval.
110−-- >>> Stream.fold Fold.drain $ Concur.interjectSuffix 1.05 (putChar ',') $ Stream.mapM (\x -> threadDelay 1000000 >> putChar x) $ Stream.fromList "hello"
117+{-# INLINE periodic #-}
111−-- h,e,l,l,o
118+periodic :: MonadIO m => m a -> Double -> Stream m a
114−{-# INLINE interjectSuffix #-}
115−interjectSuffix :: MonadAsync m => Double -> m a -> Stream m a -> Stream m a
116−interjectSuffix n f xs = parallelFst [xs, Stream.repeatM timed]
119+periodic action n = Stream.repeatM timed
120− timed = liftIO (threadDelay (round $ n * 1000000)) >> f
123+ timed = liftIO (threadDelay (round $ n * 1000000)) >> action
122−-- | Generate ticks at the specified rate. The rate is adaptive, the tick
125+-- | Generate a tick stream consisting of '()' elements, each tick is generated
123−-- generation speed can be increased or decreased at different times to achieve
126+-- after the specified time delay given in seconds.
124−-- the specified rate. The specific behavior for different styles of 'Rate'
125−-- specifications is documented under 'Rate'. The effective maximum rate
126−-- achieved by a stream is governed by the processor speed.
128−-- /Unimplemented/
128+-- >>> ticks = Stream.periodic (return ())
130{-# INLINE ticks #-}
130{-# INLINE ticks #-}
131−ticks :: -- (MonadAsync m) =>
131+ticks :: MonadIO m => Double -> Stream m ()
132− Rate -> Stream m ()
132+ticks = periodic (return ())
135−-- XXX Notes from D.takeByTime (which was removed)
134+-- | Generate a tick stream, ticks are generated at the specified 'Rate'. The
136−-- XXX using getTime in the loop can be pretty expensive especially for
135+-- rate is adaptive, the tick generation speed can be increased or decreased at
137−-- computations where iterations are lightweight. We have the following
136+-- different times to achieve the specified rate. The specific behavior for
138−-- options:
137+-- different styles of 'Rate' specifications is documented under 'Rate'. The
138+-- effective maximum rate achieved by a stream is governed by the processor
140−-- 1) Run a timeout thread updating a flag asynchronously and check that
141+-- >>> tickStream = Stream.repeatM (return ())
141−-- flag here, that way we can have a cheap termination check.
142+-- >>> ticksRate r = Concur.evalWith (Concur.rate (Just r)) tickStream
143−-- 2) Use COARSE clock to get time with lower resolution but more efficiently.
144+{-# INLINE ticksRate #-}
144−--
145+ticksRate :: MonadAsync m => Rate -> Stream m ()
145−-- 3) Use rdtscp/rdtsc to get time directly from the processor, compute the
146+ticksRate r = evalWith (rate (Just r)) $ Stream.repeatM (return ())
146−-- termination value of rdtsc in the beginning and then in each iteration just
147−-- get rdtsc and check if we should terminate.
148+-- XXX The case when the interval is 0, we should run only the stream being
150−-- | @takeInterval duration@ yields stream elements upto specified time
151+-- | Intersperse a monadic action into the input stream after every @n@
151−-- @duration@ in seconds. The duration starts when the stream is evaluated for
152+-- seconds.
152−-- the first time, before the first element is yielded. The time duration is
153−-- checked before generating each element, if the duration has expired the
156−-- The total time taken in executing the stream is guaranteed to be /at least/
154+-- Definition:
157−-- @duration@, however, because the duration is checked before generating an
158−-- element, the upper bound is indeterminate and depends on the time taken in
159−-- generating and processing the last element.
161−-- No element is yielded if the duration is zero. At least one element is
156+-- >>> interject n f xs = Concur.parallelFst [xs, Stream.periodic f n]
162−-- yielded if the duration is non-zero.
164−-- /Pre-release/
158+-- Example:
160+-- >>> s = Stream.fromList "hello"
161+-- >>> input = Stream.mapM (\x -> threadDelay 1000000 >> putChar x) s
162+-- >>> Stream.fold Fold.drain $ Stream.interject (putChar ',') 1.05 input
165+{-# INLINE interject #-}
166+interject :: MonadAsync m => m a -> Double -> Stream m a -> Stream m a
167+interject f n xs = parallelFst [xs, periodic f n]
169+-- XXX No element should be yielded if the duration is zero.
171+-- | @takeInterval interval@ runs the stream only upto the specified time
172+-- @interval@ in seconds.
174+-- The interval starts when the stream is evaluated for the first time.
166{-# INLINE takeInterval #-}
176{-# INLINE takeInterval #-}
167takeInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
177takeInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
168takeInterval d =
178takeInterval d =
169 Stream.catMaybes
179 Stream.catMaybes
170 . Stream.takeWhile isNothing
180 . Stream.takeWhile isNothing
171− . interjectSuffix d (return Nothing) . fmap Just
181+ . interject (return Nothing) d . fmap Just
173-- | Take time interval @i@ seconds at the end of the stream.
183-- | Take time interval @i@ seconds at the end of the stream.
@@ -180,26 +190,19 @@ takeLastInterval :: -- MonadAsync m =>
180 Double -> Stream m a -> Stream m a
190 Double -> Stream m a -> Stream m a
181takeLastInterval = undefined
191takeLastInterval = undefined
193+-- XXX All elements should be yielded if the duration is zero.
183−-- | @dropInterval duration@ drops stream elements until specified @duration@ in
195+-- | @dropInterval interval@ drops all the stream elements that are generated
184−-- seconds has passed. The duration begins when the stream is evaluated for the
196+-- before the specified @interval@ in seconds has passed.
185−-- first time. The time duration is checked /after/ generating a stream element,
186−-- the element is yielded if the duration has expired otherwise it is dropped.
188−-- The time elapsed before starting to generate the first element is /at most/
189−-- @duration@, however, because the duration expiry is checked after the
190−-- element is generated, the lower bound is indeterminate and depends on the
191−-- time taken in generating an element.
193−-- All elements are yielded if the duration is zero.
195−-- /Pre-release/
198+-- The interval begins when the stream is evaluated for the first time.
197{-# INLINE dropInterval #-}
200{-# INLINE dropInterval #-}
198dropInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
201dropInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
199dropInterval d =
202dropInterval d =
200 Stream.catMaybes
203 Stream.catMaybes
201 . Stream.dropWhile isNothing
204 . Stream.dropWhile isNothing
202− . interjectSuffix d (return Nothing) . fmap Just
205+ . interject (return Nothing) d . fmap Just
204-- | Drop time interval @i@ seconds at the end of the stream.
207-- | Drop time interval @i@ seconds at the end of the stream.
@@ -211,32 +214,31 @@ dropLastInterval :: -- MonadAsync m =>
211 Int -> Stream m a -> Stream m a
214 Int -> Stream m a -> Stream m a
212dropLastInterval = undefined
215dropLastInterval = undefined
214−-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
215−-- XXX add this example after fixing the serial stream rate control
217-- | Group the input stream into windows of @n@ second each and then fold each
217-- | Group the input stream into windows of @n@ second each and then fold each
218-- group using the provided fold function.
218-- group using the provided fold function.
220−-- >>> Stream.toList $ Stream.take 5 $ Stream.intervalsOf 1 Fold.sum $ Stream.constRate 2 $ Stream.enumerateFrom 1
220+-- >>> twoPerSec = Concur.evalWith (Concur.constRate 2) $ Stream.enumerateFrom 1
221+-- >>> intervals = Stream.intervalsOf 1 Fold.toList twoPerSec
222+-- >>> Stream.fold Fold.toList $ Stream.take 2 intervals
221−-- [...,...,...,...,...]
223+-- [...,...]
223{-# INLINE intervalsOf #-}
225{-# INLINE intervalsOf #-}
224intervalsOf :: MonadAsync m => Double -> Fold m a b -> Stream m a -> Stream m b
226intervalsOf :: MonadAsync m => Double -> Fold m a b -> Stream m a -> Stream m b
225intervalsOf n f xs =
227intervalsOf n f xs =
226 Stream.foldMany
228 Stream.foldMany
227 (Fold.takeEndBy isNothing (Fold.catMaybes f))
229 (Fold.takeEndBy isNothing (Fold.catMaybes f))
228− (interjectSuffix n (return Nothing) (fmap Just xs))
230+ (interject (return Nothing) n (fmap Just xs))
230-- XXX This can be implemented more efficiently by sharing a Clock.Timer across
232-- XXX This can be implemented more efficiently by sharing a Clock.Timer across
231-- parallel threads and resetting it whenever a span is emitted.
233-- parallel threads and resetting it whenever a span is emitted.
233-- | Like 'chunksOf' but if the chunk is not completed within the specified
235-- | Like 'chunksOf' but if the chunk is not completed within the specified
234-- time interval then emit whatever we have collected till now. The chunk
236-- time interval then emit whatever we have collected till now. The chunk
235-- timeout is reset whenever a chunk is emitted. The granularity of the clock
237-- timeout is reset whenever a chunk is emitted. The granularity of the clock
236-- is 100 ms.
238-- is 100 ms.
238-- >>> s = Stream.delayPost 0.3 $ Stream.fromList [1..1000]
240-- >>> s = Stream.delayPost 0.3 $ Stream.fromList [1..1000]
239−-- >>> f = Stream.fold (Fold.drainBy print) $ Stream.chunksOfTimeout 5 1 Fold.toList s
241+-- >>> f = Stream.fold (Fold.drainMapM print) $ Stream.chunksOfTimeout 5 1 Fold.toList s
241-- /Pre-release/
243-- /Pre-release/
242{-# INLINE chunksOfTimeout #-}
244{-# INLINE chunksOfTimeout #-}
@@ -249,7 +251,6 @@ chunksOfTimeout n timeout f =
249 . Stream.timestamped
251 . Stream.timestamped
250 . fmap ((),)
252 . fmap ((),)
253------------------------------------------------------------------------------
254------------------------------------------------------------------------------
254-- Windowed classification
255-- Windowed classification
255------------------------------------------------------------------------------
256------------------------------------------------------------------------------
@@ -537,7 +538,7 @@ classifySessionsByGeneric _ tick reset ejectPred tmout
537 (Fold step initial extract) input =
538 (Fold step initial extract) input =
538 Stream.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)
539 Stream.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)
539 $ Stream.scanlMAfter' sstep (return szero) (flush extract)
540 $ Stream.scanlMAfter' sstep (return szero) (flush extract)
540− $ interjectSuffix tick (return Nothing)
541+ $ interject (return Nothing) tick
541 $ fmap Just input
542 $ fmap Just input
@@ -709,7 +710,7 @@ classifySessionsByGeneric _ tick reset ejectPred tmout
711-- Stream.fold (Fold.drainMapM print)
712-- Stream.fold (Fold.drainMapM print)
712−-- $ Concur.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
713+-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
713-- $ Stream.timestamped
714-- $ Stream.timestamped
714-- $ Stream.delay 0.1
715-- $ Stream.delay 0.1
715-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
716-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
@@ -781,9 +782,7 @@ classifyChunksOf wsize = classifyChunksBy wsize False
781-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
782-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
782-- option set to 'False'.
783-- option set to 'False'.
785−-- classifySessionsOf = classifySessionsBy 1 False
785+-- >>> classifySessionsOf = Stream.classifySessionsBy 1 False
788-- /Pre-release/
787-- /Pre-release/
@@ -801,16 +800,12 @@ classifySessionsOf = classifySessionsBy 1 False
801-- Sampling
800-- Sampling
802------------------------------------------------------------------------------
801------------------------------------------------------------------------------
804−-- | Continuously evaluate the input stream and sample the last event in time
803+-- | Continuously evaluate the input stream and sample the last event in each
805−-- window of @n@ seconds.
804+-- time window of @n@ seconds.
807-- This is also known as @throttle@ in some libraries.
806-- This is also known as @throttle@ in some libraries.
810−-- sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
808+-- >>> sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.latest
815{-# INLINE sampleIntervalEnd #-}
810{-# INLINE sampleIntervalEnd #-}
816sampleIntervalEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
811sampleIntervalEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
@@ -818,11 +813,7 @@ sampleIntervalEnd n = Stream.catMaybes . intervalsOf n Fold.latest
819-- | Like 'sampleInterval' but samples at the beginning of the time window.
814-- | Like 'sampleInterval' but samples at the beginning of the time window.
822−-- sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
816+-- >>> sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
827{-# INLINE sampleIntervalStart #-}
818{-# INLINE sampleIntervalStart #-}
828sampleIntervalStart :: MonadAsync m => Double -> Stream m a -> Stream m a
819sampleIntervalStart :: MonadAsync m => Double -> Stream m a -> Stream m a
@@ -843,7 +834,7 @@ sampleBurst sampleAtEnd gap xs =
843 Stream.mapMaybe extract
834 Stream.mapMaybe extract
844 $ Stream.scan (Fold.foldl' step BurstNone)
835 $ Stream.scan (Fold.foldl' step BurstNone)
845 $ Stream.timeIndexed
836 $ Stream.timeIndexed
846− $ interjectSuffix 0.01 (return Nothing) (fmap Just xs)
837+ $ interject (return Nothing) 0.01 (fmap Just xs)
@@ -889,8 +880,6 @@ sampleBurst sampleAtEnd gap xs =
890-- The clock granularity is 10 ms.
881-- The clock granularity is 10 ms.
894{-# INLINE sampleBurstEnd #-}
883{-# INLINE sampleBurstEnd #-}
895sampleBurstEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
884sampleBurstEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
896sampleBurstEnd = sampleBurst True
885sampleBurstEnd = sampleBurst True
@@ -898,8 +887,6 @@ sampleBurstEnd = sampleBurst True
898-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
887-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
899-- instead of at the end of it.
888-- instead of at the end of it.
903{-# INLINE sampleBurstStart #-}
890{-# INLINE sampleBurstStart #-}
904sampleBurstStart :: MonadAsync m => Double -> Stream m a -> Stream m a
891sampleBurstStart :: MonadAsync m => Double -> Stream m a -> Stream m a
905sampleBurstStart = sampleBurst False
892sampleBurstStart = sampleBurst False
@@ -913,38 +900,50 @@ sampleBurstStart = sampleBurst False
913-- support that we can decouple evaluation and sampling in independent stages.
900-- support that we can decouple evaluation and sampling in independent stages.
914-- The sampling stage would strictly evaluate and sample, the evaluation stage
901-- The sampling stage would strictly evaluate and sample, the evaluation stage
915-- would control the evaluation rate.
902-- would control the evaluation rate.
904+-- bufferLatest - evaluate retaining only the latest element. Fork a thread
905+-- that keeps writing the stream to an IORef, another parallel thread reads
908+-- bufferLatestN - Fork a thread that keeps writing to a sliding ring buffer,
909+-- and the stream unfolds the entire buffer when it is read, and the empty
910+-- buffer starts filling again in the same way.
912+-- bufferOldestN - Fork a thread that keeps writing to a buffer until it is
913+-- filled and keeps dropping once it is filled. When the stream is read the
914+-- buffer is unfolded to a stream and the empty buffer starts filling again.
917-- | Evaluate the input stream continuously and keep only the oldest @n@
916-- | Evaluate the input stream continuously and keep only the oldest @n@
918-- elements in the buffer, discard the new ones when the buffer is full. When
917-- elements in the buffer, discard the new ones when the buffer is full. When
919−-- the output stream is evaluated it consumes the values from the buffer in a
918+-- the output stream is evaluated the collected buffer is streamed and the
920−-- FIFO manner.
919+-- buffer starts filling again.
922-- /Unimplemented/
921-- /Unimplemented/
924−{-# INLINE sampleOld #-}
923+{-# INLINE bufferOldestN #-}
925−sampleOld :: -- MonadAsync m =>
924+bufferOldestN :: -- MonadAsync m =>
926 Int -> Stream m a -> Stream m a
925 Int -> Stream m a -> Stream m a
927−sampleOld = undefined
926+bufferOldestN = undefined
929-- | Evaluate the input stream continuously and keep only the latest @n@
928-- | Evaluate the input stream continuously and keep only the latest @n@
930-- elements in a ring buffer, keep discarding the older ones to make space for
929-- elements in a ring buffer, keep discarding the older ones to make space for
931−-- the new ones. When the output stream is evaluated it consumes the values
930+-- the new ones. When the output stream is evaluated the buffer collected till
932−-- from the buffer in a FIFO manner.
931+-- now is streamed and it starts filling again.
934-- /Unimplemented/
933-- /Unimplemented/
936−{-# INLINE sampleNew #-}
935+{-# INLINE bufferLatestN #-}
937−sampleNew :: -- MonadAsync m =>
936+bufferLatestN :: -- MonadAsync m =>
938 Int -> Stream m a -> Stream m a
937 Int -> Stream m a -> Stream m a
939−sampleNew = undefined
938+bufferLatestN = undefined
941−-- | Like 'sampleNew' but samples at uniform intervals to match the consumer
940+-- | Always produce the latest available element from the stream without any
942−-- rate. Note that 'sampleNew' leads to non-uniform sampling depending on the
941+-- delay. The stream is continuously evaluated at the highest possible rate and
943−-- consumer pattern.
942+-- only the latest element is retained for sampling.
945-- /Unimplemented/
944-- /Unimplemented/
947−{-# INLINE sampleRate #-}
946+{-# INLINE bufferLatest #-}
948−sampleRate :: -- MonadAsync m =>
947+bufferLatest :: -- MonadAsync m =>
949− Double -> Stream m a -> Stream m a
948+ Stream m a -> Stream m (Maybe a)
950−sampleRate = undefined
949+bufferLatest = undefined