@@ -5,41 +5,49 @@
55-- Maintainer : streamly@composewell.com
66-- Stability : experimental
77-- Portability : GHC
8−--
98
109module Streamly.Internal.Data.Stream.Time
1110 (
12− -- Primitives
13− interjectSuffix
14− , ticks
11+ -- * Imports for Examples
12+ -- $setup
13+
14+ -- * Timers
15+ periodic -- XXX Should go to Streamly.Data.Stream
16+ , ticks -- XXX Should go to Streamly.Data.Stream
17+ , ticksRate
18+ , interject
19+
20+ -- * Trimming
1521 , takeInterval
1622 , takeLastInterval
1723 , dropInterval
1824 , dropLastInterval
25+
26+ -- * Chunking
1927 , intervalsOf
2028 , chunksOfTimeout
2129
22− -- Sessions
30+ -- * Sampling
31+ , sampleIntervalEnd
32+ , sampleIntervalStart
33+ , sampleBurst
34+ , sampleBurstEnd
35+ , sampleBurstStart
36+
37+ -- * Windowed Sessions
2338 , classifySessionsByGeneric
2439 , classifySessionsBy
2540 , classifySessionsOf
2641 , classifyKeepAliveSessions
2742
28− -- ** Buffering and Sampling
43+ -- XXX This should go in the concurrent module
44+ -- * Buffering
2945 -- | Evaluate strictly using a buffer of results. When the buffer becomes
3046 -- 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
32− -- the rate of the consumer.
33− , sampleOld
34− , sampleNew
35− , sampleRate
36−
37− -- Sampling
38− , sampleIntervalEnd
39− , sampleIntervalStart
40− , sampleBurst
41− , sampleBurstEnd
42− , sampleBurstStart
47+ -- insert the new at the end.
48+ , bufferLatest
49+ , bufferLatestN
50+ , bufferOldestN
4351 )
4452where
4553
@@ -71,8 +79,7 @@ import qualified Streamly.Data.Unfold as Unfold
7179import qualified Streamly.Internal.Data.Fold as Fold (Step(..))
7280import qualified Streamly.Internal.Data.IsMap as IsMap
7381import qualified Streamly.Internal.Data.Stream as Stream
74− ( catMaybes
75− , scanlMAfter'
82+ ( scanlMAfter'
7683 , timeIndexed
7784 , timestamped
7885 )
@@ -89,9 +96,9 @@ import Streamly.Internal.Data.Stream.Concurrent
8996-- >>> import qualified Streamly.Data.Fold as Fold
9097-- >>> import qualified Streamly.Data.Parser as Parser
9198-- >>> 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)
93100-- >>> 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
95102-- >>> import Prelude hiding (concatMap, concat)
96103-- >>> :{
97104-- delay n = do
@@ -104,71 +111,74 @@ import Streamly.Internal.Data.Stream.Concurrent
104111-- Combinators
105112--------------------------------------------------------------------------------
106113
107−-- | Intersperse a monadic action into the input stream after every @n@
108−-- seconds.
114+-- | Generate a stream by running an action periodically at the specified time
115+-- interval.
109116--
110−-- >>> Stream.fold Fold.drain $ Concur.interjectSuffix 1.05 (putChar ',') $ Stream.mapM (\x -> threadDelay 1000000 >> putChar x) $ Stream.fromList "hello"
111−-- h,e,l,l,o
112−--
113−-- /Pre-release/
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]
117+{-# INLINE periodic #-}
118+periodic :: MonadIO m => m a -> Double -> Stream m a
119+periodic action n = Stream.repeatM timed
117120
118121 where
119122
120− timed = liftIO (threadDelay (round $ n * 1000000)) >> f
123+ timed = liftIO (threadDelay (round $ n * 1000000)) >> action
121124
122−-- | Generate ticks at the specified rate. The rate is adaptive, the tick
123−-- generation speed can be increased or decreased at different times to achieve
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.
125+-- | Generate a tick stream consisting of '()' elements, each tick is generated
126+-- after the specified time delay given in seconds.
127127--
128−-- /Unimplemented/
128+-- >>> ticks = Stream.periodic (return ())
129129--
130130{-# INLINE ticks #-}
131−ticks :: -- (MonadAsync m) =>
132− Rate -> Stream m ()
133−ticks = undefined
131+ticks :: MonadIO m => Double -> Stream m ()
132+ticks = periodic (return ())
134133
135−-- XXX Notes from D.takeByTime (which was removed)
136−-- XXX using getTime in the loop can be pretty expensive especially for
137−-- computations where iterations are lightweight. We have the following
138−-- options:
134+-- | Generate a tick stream, ticks are generated at the specified 'Rate'. The
135+-- rate is adaptive, the tick generation speed can be increased or decreased at
136+-- different times to achieve the specified rate. The specific behavior for
137+-- different styles of 'Rate' specifications is documented under 'Rate'. The
138+-- effective maximum rate achieved by a stream is governed by the processor
139+-- speed.
139140--
140−-- 1) Run a timeout thread updating a flag asynchronously and check that
141−-- flag here, that way we can have a cheap termination check.
141+-- >>> tickStream = Stream.repeatM (return ())
142+-- >>> ticksRate r = Concur.evalWith (Concur.rate (Just r)) tickStream
142143--
143−-- 2) Use COARSE clock to get time with lower resolution but more efficiently.
144−--
145−-- 3) Use rdtscp/rdtsc to get time directly from the processor, compute the
146−-- termination value of rdtsc in the beginning and then in each iteration just
147−-- get rdtsc and check if we should terminate.
144+{-# INLINE ticksRate #-}
145+ticksRate :: MonadAsync m => Rate -> Stream m ()
146+ticksRate r = evalWith (rate (Just r)) $ Stream.repeatM (return ())
148147
148+-- XXX The case when the interval is 0, we should run only the stream being
149+-- interjected.
149150
150−-- | @takeInterval duration@ yields stream elements upto specified time
151−-- @duration@ in seconds. The duration starts when the stream is evaluated for
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
154−-- stream stops.
151+-- | Intersperse a monadic action into the input stream after every @n@
152+-- seconds.
155153--
156−-- The total time taken in executing the stream is guaranteed to be /at least/
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.
154+-- Definition:
160155--
161−-- No element is yielded if the duration is zero. At least one element is
162−-- yielded if the duration is non-zero.
156+-- >>> interject n f xs = Concur.parallelFst [xs, Stream.periodic f n]
163157--
164−-- /Pre-release/
158+-- Example:
159+--
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
163+-- h,e,l,l,o
164+--
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]
168+
169+-- XXX No element should be yielded if the duration is zero.
170+
171+-- | @takeInterval interval@ runs the stream only upto the specified time
172+-- @interval@ in seconds.
173+--
174+-- The interval starts when the stream is evaluated for the first time.
165175--
166176{-# INLINE takeInterval #-}
167177takeInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
168178takeInterval d =
169179 Stream.catMaybes
170180 . Stream.takeWhile isNothing
171− . interjectSuffix d (return Nothing) . fmap Just
181+ . interject (return Nothing) d . fmap Just
172182
173183-- | Take time interval @i@ seconds at the end of the stream.
174184--
@@ -180,26 +190,19 @@ takeLastInterval :: -- MonadAsync m =>
180190 Double -> Stream m a -> Stream m a
181191takeLastInterval = undefined
182192
183−-- | @dropInterval duration@ drops stream elements until specified @duration@ in
184−-- seconds has passed. The duration begins when the stream is evaluated for the
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.
187−--
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.
192−--
193−-- All elements are yielded if the duration is zero.
193+-- XXX All elements should be yielded if the duration is zero.
194+
195+-- | @dropInterval interval@ drops all the stream elements that are generated
196+-- before the specified @interval@ in seconds has passed.
194197--
195−-- /Pre-release/
198+-- The interval begins when the stream is evaluated for the first time.
196199--
197200{-# INLINE dropInterval #-}
198201dropInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
199202dropInterval d =
200203 Stream.catMaybes
201204 . Stream.dropWhile isNothing
202− . interjectSuffix d (return Nothing) . fmap Just
205+ . interject (return Nothing) d . fmap Just
203206
204207-- | Drop time interval @i@ seconds at the end of the stream.
205208--
@@ -211,32 +214,31 @@ dropLastInterval :: -- MonadAsync m =>
211214 Int -> Stream m a -> Stream m a
212215dropLastInterval = undefined
213216
214−-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
215−-- XXX add this example after fixing the serial stream rate control
216−--
217217-- | Group the input stream into windows of @n@ second each and then fold each
218218-- group using the provided fold function.
219219--
220−-- >>> Stream.toList $ Stream.take 5 $ Stream.intervalsOf 1 Fold.sum $ Stream.constRate 2 $ Stream.enumerateFrom 1
221−-- [...,...,...,...,...]
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
223+-- [...,...]
222224--
223225{-# INLINE intervalsOf #-}
224226intervalsOf :: MonadAsync m => Double -> Fold m a b -> Stream m a -> Stream m b
225227intervalsOf n f xs =
226228 Stream.foldMany
227229 (Fold.takeEndBy isNothing (Fold.catMaybes f))
228− (interjectSuffix n (return Nothing) (fmap Just xs))
230+ (interject (return Nothing) n (fmap Just xs))
229231
230232-- XXX This can be implemented more efficiently by sharing a Clock.Timer across
231233-- parallel threads and resetting it whenever a span is emitted.
232−--
234+
233235-- | Like 'chunksOf' but if the chunk is not completed within the specified
234236-- time interval then emit whatever we have collected till now. The chunk
235237-- timeout is reset whenever a chunk is emitted. The granularity of the clock
236238-- is 100 ms.
237239--
238240-- >>> 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
240242--
241243-- /Pre-release/
242244{-# INLINE chunksOfTimeout #-}
@@ -249,7 +251,6 @@ chunksOfTimeout n timeout f =
249251 . Stream.timestamped
250252 . fmap ((),)
251253
252−
253254------------------------------------------------------------------------------
254255-- Windowed classification
255256------------------------------------------------------------------------------
@@ -537,7 +538,7 @@ classifySessionsByGeneric _ tick reset ejectPred tmout
537538 (Fold step initial extract) input =
538539 Stream.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)
539540 $ Stream.scanlMAfter' sstep (return szero) (flush extract)
540− $ interjectSuffix tick (return Nothing)
541+ $ interject (return Nothing) tick
541542 $ fmap Just input
542543
543544 where
@@ -709,7 +710,7 @@ classifySessionsByGeneric _ tick reset ejectPred tmout
709710--
710711-- >>> :{
711712-- 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)
713714-- $ Stream.timestamped
714715-- $ Stream.delay 0.1
715716-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
@@ -781,9 +782,7 @@ classifyChunksOf wsize = classifyChunksBy wsize False
781782-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
782783-- option set to 'False'.
783784--
784−-- @
785−-- classifySessionsOf = classifySessionsBy 1 False
786−-- @
785+-- >>> classifySessionsOf = Stream.classifySessionsBy 1 False
787786--
788787-- /Pre-release/
789788--
@@ -801,16 +800,12 @@ classifySessionsOf = classifySessionsBy 1 False
801800-- Sampling
802801------------------------------------------------------------------------------
803802
804−-- | Continuously evaluate the input stream and sample the last event in time
805−-- window of @n@ seconds.
803+-- | Continuously evaluate the input stream and sample the last event in each
804+-- time window of @n@ seconds.
806805--
807806-- This is also known as @throttle@ in some libraries.
808807--
809−-- @
810−-- sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
811−-- @
812−--
813−-- /Pre-release/
808+-- >>> sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.latest
814809--
815810{-# INLINE sampleIntervalEnd #-}
816811sampleIntervalEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
@@ -818,11 +813,7 @@ sampleIntervalEnd n = Stream.catMaybes . intervalsOf n Fold.latest
818813
819814-- | Like 'sampleInterval' but samples at the beginning of the time window.
820815--
821−-- @
822−-- sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
823−-- @
824−--
825−-- /Pre-release/
816+-- >>> sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
826817--
827818{-# INLINE sampleIntervalStart #-}
828819sampleIntervalStart :: MonadAsync m => Double -> Stream m a -> Stream m a
@@ -843,7 +834,7 @@ sampleBurst sampleAtEnd gap xs =
843834 Stream.mapMaybe extract
844835 $ Stream.scan (Fold.foldl' step BurstNone)
845836 $ Stream.timeIndexed
846− $ interjectSuffix 0.01 (return Nothing) (fmap Just xs)
837+ $ interject (return Nothing) 0.01 (fmap Just xs)
847838
848839 where
849840
@@ -889,8 +880,6 @@ sampleBurst sampleAtEnd gap xs =
889880--
890881-- The clock granularity is 10 ms.
891882--
892−-- /Pre-release/
893−--
894883{-# INLINE sampleBurstEnd #-}
895884sampleBurstEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
896885sampleBurstEnd = sampleBurst True
@@ -898,8 +887,6 @@ sampleBurstEnd = sampleBurst True
898887-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
899888-- instead of at the end of it.
900889--
901−-- /Pre-release/
902−--
903890{-# INLINE sampleBurstStart #-}
904891sampleBurstStart :: MonadAsync m => Double -> Stream m a -> Stream m a
905892sampleBurstStart = sampleBurst False
@@ -913,38 +900,50 @@ sampleBurstStart = sampleBurst False
913900-- support that we can decouple evaluation and sampling in independent stages.
914901-- The sampling stage would strictly evaluate and sample, the evaluation stage
915902-- would control the evaluation rate.
903+--
904+-- bufferLatest - evaluate retaining only the latest element. Fork a thread
905+-- that keeps writing the stream to an IORef, another parallel thread reads
906+-- from the IORef.
907+--
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.
911+--
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.
916915
917916-- | Evaluate the input stream continuously and keep only the oldest @n@
918917-- 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
920−-- FIFO manner.
918+-- the output stream is evaluated the collected buffer is streamed and the
919+-- buffer starts filling again.
921920--
922921-- /Unimplemented/
923922--
924−{-# INLINE sampleOld #-}
925−sampleOld :: -- MonadAsync m =>
923+{-# INLINE bufferOldestN #-}
924+bufferOldestN :: -- MonadAsync m =>
926925 Int -> Stream m a -> Stream m a
927−sampleOld = undefined
926+bufferOldestN = undefined
928927
929928-- | Evaluate the input stream continuously and keep only the latest @n@
930929-- 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
932−-- from the buffer in a FIFO manner.
930+-- the new ones. When the output stream is evaluated the buffer collected till
931+-- now is streamed and it starts filling again.
933932--
934933-- /Unimplemented/
935934--
936−{-# INLINE sampleNew #-}
937−sampleNew :: -- MonadAsync m =>
935+{-# INLINE bufferLatestN #-}
936+bufferLatestN :: -- MonadAsync m =>
938937 Int -> Stream m a -> Stream m a
939−sampleNew = undefined
938+bufferLatestN = undefined
940939
941−-- | Like 'sampleNew' but samples at uniform intervals to match the consumer
942−-- rate. Note that 'sampleNew' leads to non-uniform sampling depending on the
943−-- consumer pattern.
940+-- | Always produce the latest available element from the stream without any
941+-- delay. The stream is continuously evaluated at the highest possible rate and
942+-- only the latest element is retained for sampling.
944943--
945944-- /Unimplemented/
946945--
947−{-# INLINE sampleRate #-}
948−sampleRate :: -- MonadAsync m =>
949− Double -> Stream m a -> Stream m a
950−sampleRate = undefined
946+{-# INLINE bufferLatest #-}
947+bufferLatest :: -- MonadAsync m =>
948+ Stream m a -> Stream m (Maybe a)
949+bufferLatest = undefined