2-- Module : Streamly.Internal.Data.Stream.Time
3-- Copyright : (c) 2020 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
10module Streamly.Internal.Data.Stream.Time
23 , classifySessionsByGeneric
26 , classifyKeepAliveSessions
28 -- ** Buffering and Sampling
29 -- | 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
31 -- insert the new at the end or keep dropping elements uniformly to match
32 -- the rate of the consumer.
46import Control.Concurrent (threadDelay)
47import Control.Exception (assert)
48import Control.Monad.IO.Class (MonadIO(..))
49import Data.Heap (Entry(..))
50import Data.Kind (Type)
52import Data.Maybe (isNothing)
53import Data.Proxy (Proxy(..))
54import Streamly.Data.Fold (Fold)
55import Streamly.Internal.Data.Fold.Type (Fold (..))
56import Streamly.Internal.Data.IsMap (IsMap(..))
57import Streamly.Internal.Data.Stream.Type (Stream)
58import Streamly.Internal.Data.Time.Units
65import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)
67import qualified Data.Heap as H
68import qualified Streamly.Data.Fold as Fold
69import qualified Streamly.Data.Stream as Stream
70import qualified Streamly.Data.Unfold as Unfold
71import qualified Streamly.Internal.Data.Fold as Fold (Step(..))
72import qualified Streamly.Internal.Data.IsMap as IsMap
73import qualified Streamly.Internal.Data.Stream as Stream
80import Streamly.Internal.Data.Stream.Concurrent
84-- Imports for example snippets in this module.
87-- >>> import Control.Concurrent (threadDelay)
88-- >>> import qualified Streamly.Data.Array as Array
89-- >>> import qualified Streamly.Data.Fold as Fold
90-- >>> import qualified Streamly.Data.Parser as Parser
91-- >>> import qualified Streamly.Data.Stream as Stream
92-- >>> import qualified Streamly.Internal.Data.Stream as Stream (timestamped)
93-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Concur
94-- >>> import qualified Streamly.Internal.Data.Stream.Time as Concur
95-- >>> import Prelude hiding (concatMap, concat)
98-- threadDelay (n * 1000000) -- sleep for n seconds
99-- putStrLn (show n ++ " sec") -- print "n sec"
103--------------------------------------------------------------------------------
105--------------------------------------------------------------------------------
107-- | Intersperse a monadic action into the input stream after every @n@
110-- >>> Stream.fold Fold.drain $ Concur.interjectSuffix 1.05 (putChar ',') $ Stream.mapM (\x -> threadDelay 1000000 >> putChar x) $ Stream.fromList "hello"
114{-# INLINE interjectSuffix #-}
115interjectSuffix :: MonadAsync m => Double -> m a -> Stream m a -> Stream m a
116interjectSuffix n f xs = parallelFst [xs, repeatM timed]
120 timed = liftIO (threadDelay (round $ n * 1000000)) >> f
121 repeatM = Stream.sequence . Stream.repeat
123-- | Generate ticks at the specified rate. The rate is adaptive, the tick
124-- generation speed can be increased or decreased at different times to achieve
125-- the specified rate. The specific behavior for different styles of 'Rate'
126-- specifications is documented under 'Rate'. The effective maximum rate
127-- achieved by a stream is governed by the processor speed.
132ticks :: -- (MonadAsync m) =>
136-- XXX Notes from D.takeByTime (which was removed)
137-- XXX using getTime in the loop can be pretty expensive especially for
138-- computations where iterations are lightweight. We have the following
141-- 1) Run a timeout thread updating a flag asynchronously and check that
142-- flag here, that way we can have a cheap termination check.
144-- 2) Use COARSE clock to get time with lower resolution but more efficiently.
146-- 3) Use rdtscp/rdtsc to get time directly from the processor, compute the
147-- termination value of rdtsc in the beginning and then in each iteration just
148-- get rdtsc and check if we should terminate.
151-- | @takeInterval duration@ yields stream elements upto specified time
152-- @duration@ in seconds. The duration starts when the stream is evaluated for
153-- the first time, before the first element is yielded. The time duration is
154-- checked before generating each element, if the duration has expired the
157-- The total time taken in executing the stream is guaranteed to be /at least/
158-- @duration@, however, because the duration is checked before generating an
159-- element, the upper bound is indeterminate and depends on the time taken in
160-- generating and processing the last element.
162-- No element is yielded if the duration is zero. At least one element is
163-- yielded if the duration is non-zero.
167{-# INLINE takeInterval #-}
168takeInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
171 . Stream.takeWhile isNothing
172 . interjectSuffix d (return Nothing) . fmap Just
174-- | Take time interval @i@ seconds at the end of the stream.
176-- O(n) space, where n is the number elements taken.
179{-# INLINE takeLastInterval #-}
180takeLastInterval :: -- MonadAsync m =>
181 Double -> Stream m a -> Stream m a
182takeLastInterval = undefined
184-- | @dropInterval duration@ drops stream elements until specified @duration@ in
185-- seconds has passed. The duration begins when the stream is evaluated for the
186-- first time. The time duration is checked /after/ generating a stream element,
187-- the element is yielded if the duration has expired otherwise it is dropped.
189-- The time elapsed before starting to generate the first element is /at most/
190-- @duration@, however, because the duration expiry is checked after the
191-- element is generated, the lower bound is indeterminate and depends on the
192-- time taken in generating an element.
194-- All elements are yielded if the duration is zero.
198{-# INLINE dropInterval #-}
199dropInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
202 . Stream.dropWhile isNothing
203 . interjectSuffix d (return Nothing) . fmap Just
205-- | Drop time interval @i@ seconds at the end of the stream.
207-- O(n) space, where n is the number elements dropped.
210{-# INLINE dropLastInterval #-}
211dropLastInterval :: -- MonadAsync m =>
212 Int -> Stream m a -> Stream m a
213dropLastInterval = undefined
215-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
216-- XXX add this example after fixing the serial stream rate control
218-- | Group the input stream into windows of @n@ second each and then fold each
219-- group using the provided fold function.
221-- >>> Stream.toList $ Stream.take 5 $ Stream.intervalsOf 1 Fold.sum $ Stream.constRate 2 $ Stream.enumerateFrom 1
222-- [...,...,...,...,...]
224{-# INLINE intervalsOf #-}
225intervalsOf :: MonadAsync m => Double -> Fold m a b -> Stream m a -> Stream m b
228 (Fold.takeEndBy isNothing (Fold.catMaybes f))
229 (interjectSuffix n (return Nothing) (fmap Just xs))
231-- XXX This can be implemented more efficiently by sharing a Clock.Timer across
232-- parallel threads and resetting it whenever a span is emitted.
234-- | Like 'chunksOf' but if the chunk is not completed within the specified
235-- time interval then emit whatever we have collected till now. The chunk
236-- timeout is reset whenever a chunk is emitted. The granularity of the clock
239-- >>> s = Stream.delayPost 0.3 $ Stream.fromList [1..1000]
240-- >>> f = Stream.fold (Fold.drainBy print) $ Stream.chunksOfTimeout 5 1 Fold.toList s
243{-# INLINE chunksOfTimeout #-}
244chunksOfTimeout :: MonadAsync m
245 => Int -> Double -> Fold m a b -> Stream m a -> Stream m b
246chunksOfTimeout n timeout f =
249 0.1 False (const (return False)) timeout (Fold.take n f)
254------------------------------------------------------------------------------
255-- Windowed classification
256------------------------------------------------------------------------------
258-- TODO: To mark the position in space or time we can have Indexed or
259-- TimeStamped types. That can make it easy to deal with the position indices
262------------------------------------------------------------------------------
263-- Keyed Sliding Windows
264------------------------------------------------------------------------------
267{-# INLINABLE classifySlidingChunks #-}
269 :: (IsStream t, MonadAsync m, Ord k)
270 => Int -- ^ window size
271 -> Int -- ^ window slide
272 -> Fold m a b -- ^ Fold to be applied to window events
273 -> t m (k, a, Bool) -- ^ window key, data, close event
275classifySlidingChunks wsize wslide (Fold step initial extract) str
278-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
279-- slide the send window when an ack is received and we slide the receive
280-- window when a sequence is complete. Sliding is stateful in case of TCP,
281-- sliding releases the send buffer or makes data available to the user from
282-- the receive buffer.
283{-# INLINABLE classifySlidingSessions #-}
284classifySlidingSessions
285 :: (IsStream t, MonadAsync m, Ord k)
286 => Double -- ^ timer tick in seconds
287 -> Double -- ^ time window size
288 -> Double -- ^ window slide
289 -> Fold m a b -- ^ Fold to be applied to window events
290 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
292classifySlidingSessions tick interval slide (Fold step initial extract) str
296------------------------------------------------------------------------------
297-- Sliding Window Buffers
298------------------------------------------------------------------------------
300-- These buffered versions could be faster than concurrent incremental folds of
301-- all overlapping windows as in many cases we may not need all the values to
302-- compute the fold, we can just compute the result using the old value and new
303-- value. However, we may need the buffer once in a while, for example for
304-- string search we usually compute the hash incrementally but when the hash
305-- matches the hash of the pattern we need to compare the whole string.
307-- XXX we should be able to implement sequence based splitting combinators
308-- using this combinator.
311-- | Buffer n elements of the input in a ring buffer. When t new elements are
312-- collected, slide the window to remove the same number of oldest elements,
313-- insert the new elements, and apply an incremental fold on the sliding
314-- window, supplying the outgoing elements, the new ring buffer as arguments.
316 :: (IsStream t, Monad m, Ord a, Unboxed a)
317 => Int -- window size
318 -> Int -- window slide
319 -> Fold m (Ring a, Array a) b
322slidingChunkBuffer = undefined
324-- Buffer n seconds worth of stream elements of the input in a radix tree.
325-- Every t seconds, remove the items that are older than n seconds, and apply
326-- an incremental fold on the sliding window, supplying the outgoing elements,
327-- and the new radix tree buffer as arguments.
329 :: (IsStream t, Monad m, Ord a, Unboxed a)
330 => Int -- window size
332 -> Fold m (RTree a, Array a) b
335slidingSessionBuffer = undefined
338------------------------------------------------------------------------------
339-- Keyed Session Windows
340------------------------------------------------------------------------------
343-- | Keyed variable size space windows. Close the window if we do not receive a
344-- window event in the next "spaceout" elements.
345{-# INLINABLE classifyChunksBy #-}
347 :: (IsStream t, MonadAsync m, Ord k)
348 => Int -- ^ window spaceout (spread)
349 -> Bool -- ^ reset the spaceout when a chunk window element is received
350 -> Fold m a b -- ^ Fold to be applied to chunk window elements
351 -> t m (k, a, Bool) -- ^ chunk key, data, last element
353classifyChunksBy spanout reset (Fold step initial extract) str = undefined
355-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
356-- received within the chunk size window. The chunk gets closed only if no
357-- element is received within the chunk window.
359{-# INLINABLE classifyKeepAliveChunks #-}
360classifyKeepAliveChunks
361 :: (IsStream t, MonadAsync m, Ord k)
362 => Int -- ^ window spaceout (spread)
363 -> Fold m a b -- ^ Fold to be applied to chunk window elements
364 -> t m (k, a, Bool) -- ^ chunk key, data, last element
366classifyKeepAliveChunks spanout = classifyChunksBy spanout True
369data SessionState t m f s b = SessionState
370 { sessionCurTime :: !AbsTime -- ^ time since last event
371 , sessionEventTime :: !AbsTime -- ^ time as per last event
372 -- We can use the Map size instead of maintaining a count, but if we have
373 -- to switch to HashMap then it can be useful.
374 , sessionCount :: !Int -- ^ total number of sessions in progress
375 , sessionTimerHeap :: H.Heap (H.Entry AbsTime (Key f)) -- ^ heap for timeouts
376 , sessionKeyValueMap :: f s -- ^ Stored sessions for keys
377 , sessionOutputStream :: t (m :: Type -> Type) (Key f, b) -- ^ Completed sessions
380data SessionEntry s = LiveSession !AbsTime !s | ZombieSession
382-- delete from map and output the fold accumulator
383ejectEntry :: (Monad m, IsMap f) =>
387 -> Stream m (Key f, b)
391 -> m (heap, f entry, Stream m (Key f, b), Int)
392ejectEntry extract hp mp out cnt acc key = do
394 let out1 = (key, sess) `Stream.cons` out
395 let mp1 = IsMap.mapDelete key mp
396 return (hp, mp1, out1, cnt - 1)
398{-# NOINLINE flush #-}
399flush :: (Monad m, IsMap f) =>
401 -> SessionState Stream m f (SessionEntry s) b
402 -> m (SessionState Stream m f (SessionEntry s) b)
403flush extract session@SessionState{..} = do
404 (hp', mp', out, count) <-
412 { sessionCount = count
413 , sessionTimerHeap = hp'
414 , sessionKeyValueMap = mp'
415 , sessionOutputStream = out
420 ejectAll (hp, mp, out, !cnt) = do
421 let hres = H.uncons hp
423 Just (Entry _ key, hp1) -> do
424 r <- case IsMap.mapLookup key mp of
425 Nothing -> return (hp1, mp, out, cnt)
426 Just ZombieSession ->
427 return (hp1, IsMap.mapDelete key mp, out, cnt)
428 Just (LiveSession _ acc) ->
429 ejectEntry extract hp1 mp out cnt acc key
432 assert (IsMap.mapNull mp) (return ())
433 return (hp, mp, out, cnt)
435{-# NOINLINE ejectOne #-}
436ejectOne :: (IsMap f, Monad m) =>
439 -> ( H.Heap (Entry AbsTime (Key f))
440 , f (SessionEntry acc)
441 , Stream m (Key f, b)
444 -> m ( H.Heap (Entry AbsTime (Key f))
445 , f (SessionEntry acc)
446 , Stream m (Key f, b), Int
448ejectOne reset extract = go
452 go (hp, mp, out, !cnt) = do
453 let hres = H.uncons hp
455 Just (Entry expiry key, hp1) ->
456 case IsMap.mapLookup key mp of
457 Nothing -> go (hp1, mp, out, cnt)
458 Just ZombieSession ->
459 go (hp1, IsMap.mapDelete key mp, out, cnt)
460 Just (LiveSession expiry1 acc) -> do
461 if not reset || expiry1 <= expiry
462 then ejectEntry extract hp1 mp out cnt acc key
464 -- reset the session timeout and continue
465 let hp2 = H.insert (Entry expiry1 key) hp1
466 in go (hp2, mp, out, cnt)
468 assert (IsMap.mapNull mp) (return ())
469 return (hp, mp, out, cnt)
471{-# NOINLINE ejectExpired #-}
472ejectExpired :: (IsMap f, Monad m) =>
476 -> SessionState Stream m f (SessionEntry acc) b
478 -> m (SessionState Stream m f (SessionEntry acc) b)
479ejectExpired reset ejectPred extract session@SessionState{..} curTime = do
480 (hp', mp', out, count) <-
482 sessionTimerHeap sessionKeyValueMap Stream.nil sessionCount
484 { sessionCurTime = curTime
485 , sessionCount = count
486 , sessionTimerHeap = hp'
487 , sessionKeyValueMap = mp'
488 , sessionOutputStream = out
493 ejectLoop hp mp out !cnt = do
494 let hres = H.uncons hp
496 Just (Entry expiry key, hp1) -> do
499 then return (True, False)
505 case IsMap.mapLookup key mp of
506 Nothing -> ejectLoop hp1 mp out cnt
507 Just ZombieSession ->
508 ejectLoop hp1 (IsMap.mapDelete key mp) out cnt
509 Just (LiveSession expiry1 acc) -> do
510 if expiry1 <= curTime || not reset || force
512 (hp2,mp1,out1,cnt1) <-
513 ejectEntry extract hp1 mp out cnt acc key
514 ejectLoop hp2 mp1 out1 cnt1
516 -- reset the session timeout and continue
517 let hp2 = H.insert (Entry expiry1 key) hp1
518 in ejectLoop hp2 mp out cnt
519 else return (hp, mp, out, cnt)
521 assert (IsMap.mapNull mp) (return ())
522 return (hp, mp, out, cnt)
524-- XXX Use mutable IORef in accumulator
525{-# INLINE classifySessionsByGeneric #-}
526classifySessionsByGeneric
527 :: forall m f a b. (MonadAsync m, IsMap f)
528 => Proxy (f :: (Type -> Type))
529 -> Double -- ^ timer tick in seconds
530 -> Bool -- ^ reset the timer when an event is received
531 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
532 -> Double -- ^ session timeout in seconds
533 -> Fold m a b -- ^ Fold to be applied to session data
534 -> Stream m (AbsTime, (Key f, a)) -- ^ timestamp, (session key, session
536 -> Stream m (Key f, b) -- ^ session key, fold result
537classifySessionsByGeneric _ tick reset ejectPred tmout
538 (Fold step initial extract) input =
539 Stream.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)
540 $ Stream.scanlMAfter' sstep (return szero) (flush extract)
541 $ interjectSuffix tick (return Nothing)
546 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
547 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
549 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
550 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
552 , sessionTimerHeap = H.empty
553 , sessionKeyValueMap = IsMap.mapEmpty :: f s
554 , sessionOutputStream = Stream.nil
557 -- We can eject sessions based on the current session count to limit
558 -- memory consumption. There are two possible strategies:
560 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
561 -- threshold even before timeout, effectively reduce the timeout.
562 -- 2) Drop creation of new sessions but keep accepting new events for the
565 -- We use the first strategy as of now.
567 -- Got a new stream input element
568 sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
569 -- XXX instead of a heap we could use a timer wheel.
571 -- XXX if the key is an Int, we can also use an IntMap for slightly
572 -- better performance.
576 -- Values for each key are collected in a map using the supplied fold.
577 -- When we insert a key in the Map we insert an entry into the heap as
578 -- well with the session expiry as the sort key. The Map entry
579 -- consists of the fold result, and the expiry time of the session. If
580 -- "reset" is True the expiry time is readjusted whenever a new event
581 -- is processed. If the fold terminates and a new session is started
582 -- for the same key the expiry time is set to the first timestamp of
585 -- The heap must have at most one entry for any given key. The heap is
586 -- processed periodically to remove the expired entries. We pick up an
587 -- expired entry from the top of the heap and if the session has
588 -- expired based on the expiry time in the Map entry then we remove the
589 -- session from the Map and yield its fold output. Otherwise, we
590 -- reinsert the entry into the heap based on the current expiry in the
593 -- If an entry is removed from the Map and not removed from the heap
594 -- and in the meantime it is inserted again in the Map (using the same
595 -- key) then how do we avoid inserting multiple entries in the heap?
596 -- For this reason we maintain the invariant that the Map entry is
597 -- removed only when the heap entry is removed. Even if the fold has
598 -- finished we still keep a dummy Map entry (ZombieSession) until the
599 -- heap entry is removed. That way if we have a Map entry we do not
600 -- insert a heap entry because we know it is already there.
601 -- XXX The ZombieSession mechanism does not work as expected as we
602 -- ignore ZombieSession when inserting a new entry. Anyway, we can
603 -- remove this mechanism as at most only two heap entries may be
604 -- created and they will be ultimately cleaned up.
606 -- Heap processing needs the map and map processing needs the heap,
607 -- therefore we cannot separate the two for modularity unless we have a
608 -- way to achieve mutual recursion.
610 let curTime = max sessionEventTime timestamp
611 mOld = IsMap.mapLookup key sessionKeyValueMap
613 -- deleting a key from the heap is expensive, so we never
614 -- delete a key from heap, we just purge it from the Map and it
615 -- gets purged from the heap on timeout. We just need an extra
616 -- lookup in the Map when the key is purged from the heap, that
617 -- should not be expensive.
619 let (mp, cnt) = case mOld of
620 Just (LiveSession _ _) ->
622 key ZombieSession sessionKeyValueMap
625 _ -> (sessionKeyValueMap, sessionCount)
627 { sessionCurTime = curTime
628 , sessionEventTime = curTime
630 , sessionKeyValueMap = mp
631 , sessionOutputStream = Stream.fromPure (key, fb)
634 let expiry = addToAbsTime timestamp timeoutMs
635 (hp1, mp1, out1, cnt1) <- do
636 let vars = (sessionTimerHeap, sessionKeyValueMap,
637 Stream.nil, sessionCount)
639 -- inserting new entry
641 -- Eject a session from heap and map if needed
642 eject <- ejectPred sessionCount
643 (hp, mp, out, cnt) <-
645 then ejectOne reset extract vars
648 -- Insert the new session in heap
649 let hp' = H.insert (Entry expiry key) hp
650 in return (hp', mp, out, cnt + 1)
651 -- updating old entry
652 Just _ -> return vars
654 let acc = LiveSession expiry fs1
655 mp2 = IsMap.mapInsert key acc mp1
656 return $ SessionState
657 { sessionCurTime = curTime
658 , sessionEventTime = curTime
659 , sessionCount = cnt1
660 , sessionTimerHeap = hp1
661 , sessionKeyValueMap = mp2
662 , sessionOutputStream = out1
666 Just (LiveSession _ acc) -> return $ Fold.Partial acc
670 error $ "classifySessionsBy: "
671 ++ "The supplied fold must consume at least one input"
672 Fold.Partial fs -> do
675 Fold.Done fb -> done fb
676 Fold.Partial fs1 -> partial fs1
678 -- Got a timer tick event
679 sstep sessionState@SessionState{..} Nothing =
680 let curTime = addToAbsTime sessionCurTime tickMs
681 in ejectExpired reset ejectPred extract sessionState curTime
683-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
684-- classifies an input event @stream@ consisting of @(timestamp, (key,
685-- value))@ into sessions based on the @key@, folding all the values
686-- corresponding to the same key into a session using the supplied @fold@.
688-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
689-- session key and the folded value is emitted in the output stream. The
690-- timeout is measured from the first event in the session. If the @keepalive@
691-- option is set to 'True' the timeout is reset to 0 whenever an event is
694-- The @timestamp@ in the input stream is an absolute time from some epoch,
695-- characterizing the time when the input event was generated. The notion of
696-- current time is maintained by a monotonic event time clock using the
697-- timestamps seen in the input stream. The latest timestamp seen till now is
698-- used as the base for the current time. When no new events are seen, a timer
699-- is started with a clock resolution of @tick@ seconds. This timer is used to
700-- detect session timeouts in the absence of new events.
702-- To ensure an upper bound on the memory used the number of sessions can be
703-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
704-- oldest session is ejected before inserting a new session.
706-- When the stream ends any buffered sessions are ejected immediately.
708-- If a session key is received even after a session has finished, another
709-- session is created for that key.
712-- Stream.fold (Fold.drainMapM print)
713-- $ Concur.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
714-- $ Stream.timestamped
716-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
723{-# INLINE classifySessionsBy #-}
725 :: (MonadAsync m, Ord k)
726 => Double -- ^ timer tick in seconds
727 -> Bool -- ^ reset the timer when an event is received
728 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
729 -> Double -- ^ session timeout in seconds
730 -> Fold m a b -- ^ Fold to be applied to session data
731 -> Stream m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
732 -> Stream m (k, b) -- ^ session key, fold result
733classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k))
735-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
736-- option set to 'True'.
739-- classifyKeepAliveSessions = classifySessionsBy 1 True
744{-# INLINE classifyKeepAliveSessions #-}
745classifyKeepAliveSessions ::
746 (MonadAsync m, Ord k)
747 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
748 -> Double -- ^ session inactive timeout
749 -> Fold m a b -- ^ Fold to be applied to session payload data
750 -> Stream m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
752classifyKeepAliveSessions = classifySessionsBy 1 True
754------------------------------------------------------------------------------
755-- Keyed tumbling windows
756------------------------------------------------------------------------------
758-- Tumbling windows is a special case of sliding windows where the window slide
759-- is the same as the window size. Or it can be a special case of session
760-- windows where the reset flag is set to False.
762-- XXX instead of using the early termination flag in the stream, we can use an
763-- early terminating fold instead.
766-- | Split the stream into fixed size chunks of specified size. Within each
767-- such chunk fold the elements in buckets identified by the keys. A particular
768-- bucket fold can be terminated early if a closing flag is encountered in an
769-- element for that key.
772{-# INLINABLE classifyChunksOf #-}
774 :: (IsStream t, MonadAsync m, Ord k)
775 => Int -- ^ window size
776 -> Fold m a b -- ^ Fold to be applied to window events
777 -> t m (k, a, Bool) -- ^ window key, data, close event
779classifyChunksOf wsize = classifyChunksBy wsize False
782-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
783-- option set to 'False'.
786-- classifySessionsOf = classifySessionsBy 1 False
791{-# INLINE classifySessionsOf #-}
793 (MonadAsync m, Ord k)
794 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
795 -> Double -- ^ time window size
796 -> Fold m a b -- ^ Fold to be applied to session data
797 -> Stream m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
799classifySessionsOf = classifySessionsBy 1 False
801------------------------------------------------------------------------------
803------------------------------------------------------------------------------
805-- | Continuously evaluate the input stream and sample the last event in time
806-- window of @n@ seconds.
808-- This is also known as @throttle@ in some libraries.
811-- sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
816{-# INLINE sampleIntervalEnd #-}
817sampleIntervalEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
818sampleIntervalEnd n = Stream.catMaybes . intervalsOf n Fold.latest
820-- | Like 'sampleInterval' but samples at the beginning of the time window.
823-- sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
828{-# INLINE sampleIntervalStart #-}
829sampleIntervalStart :: MonadAsync m => Double -> Stream m a -> Stream m a
830sampleIntervalStart n = Stream.catMaybes . intervalsOf n Fold.one
836 | BurstDoneNext !x !t !x
838{-# INLINE sampleBurst #-}
839sampleBurst :: MonadAsync m => Bool -> Double -> Stream m a -> Stream m a
840sampleBurst sampleAtEnd gap xs =
841 -- XXX Ideally we should schedule a timer event exactly after gap time,
842 -- but the tick stream should work well as long as the timer
843 -- granularity is small enough compared to the gap.
844 Stream.mapMaybe extract
845 $ Stream.scan (Fold.foldl' step BurstNone)
847 $ interjectSuffix 0.01 (return Nothing) (fmap Just xs)
851 gap1 = toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
854 step BurstNone (t1, Just x1) = BurstWait t1 x1
855 step BurstNone _ = BurstNone
857 step (BurstDone _) (t1, Just x1) = BurstWait t1 x1
858 step (BurstDone _) _ = BurstNone
860 step old@(BurstWait t0 x0) (t1, Nothing)
861 | t1 - t0 >= gap1 = BurstDone x0
863 -- This can happen due to scheduling delays, if we received back to
864 -- back events spaced by more than the timeout without an
865 -- intervening timeout event then we emit the old event instead of
866 -- replacing it by the new.
867 step (BurstWait t0 x0) (t1, Just x1)
868 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
869 | sampleAtEnd = BurstWait t1 x1
870 | otherwise = BurstWait t1 x0
872 step (BurstDoneNext _ t0 x0) (t1, Nothing)
873 | t1 - t0 >= gap1 = BurstDone x0
874 | otherwise = BurstWait t0 x0
875 step (BurstDoneNext _ t0 x0) (t1, Just x1)
876 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
877 | sampleAtEnd = BurstWait t1 x1
878 | otherwise = BurstWait t1 x0
880 {-# INLINE extract #-}
881 extract (BurstDoneNext x _ _) = Just x
882 extract (BurstDone x) = Just x
885-- | Sample one event at the end of each burst of events. A burst is a group
886-- of events close together in time, it ends when an event is spaced by more
887-- than the specified time interval (in seconds) from the previous event.
889-- This is known as @debounce@ in some libraries.
891-- The clock granularity is 10 ms.
895{-# INLINE sampleBurstEnd #-}
896sampleBurstEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
897sampleBurstEnd = sampleBurst True
899-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
900-- instead of at the end of it.
904{-# INLINE sampleBurstStart #-}
905sampleBurstStart :: MonadAsync m => Double -> Stream m a -> Stream m a
906sampleBurstStart = sampleBurst False
908------------------------------------------------------------------------------
910------------------------------------------------------------------------------
912-- XXX We could use 'maxBuffer Block/Drop/Rotate/Sample' instead. However we
913-- may want to have the evaluation rate independent of the sampling rate. To
914-- support that we can decouple evaluation and sampling in independent stages.
915-- The sampling stage would strictly evaluate and sample, the evaluation stage
916-- would control the evaluation rate.
918-- | Evaluate the input stream continuously and keep only the oldest @n@
919-- elements in the buffer, discard the new ones when the buffer is full. When
920-- the output stream is evaluated it consumes the values from the buffer in a
925{-# INLINE sampleOld #-}
926sampleOld :: -- MonadAsync m =>
927 Int -> Stream m a -> Stream m a
930-- | Evaluate the input stream continuously and keep only the latest @n@
931-- elements in a ring buffer, keep discarding the older ones to make space for
932-- the new ones. When the output stream is evaluated it consumes the values
933-- from the buffer in a FIFO manner.
937{-# INLINE sampleNew #-}
938sampleNew :: -- MonadAsync m =>
939 Int -> Stream m a -> Stream m a
942-- | Like 'sampleNew' but samples at uniform intervals to match the consumer
943-- rate. Note that 'sampleNew' leads to non-uniform sampling depending on the
948{-# INLINE sampleRate #-}
949sampleRate :: -- MonadAsync m =>
950 Double -> Stream m a -> Stream m a
951sampleRate = undefined
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