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.Unboxed 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-- [...,...,...,...,...]
225{-# INLINE intervalsOf #-}
226intervalsOf :: MonadAsync m => Double -> Fold m a b -> Stream m a -> Stream m b
229 (Fold.takeEndBy isNothing (Fold.catMaybes f))
230 (interjectSuffix n (return Nothing) (fmap Just xs))
232-- XXX This can be implemented more efficiently by sharing a Clock.Timer across
233-- parallel threads and resetting it whenever a span is emitted.
235-- | Like 'chunksOf' but if the chunk is not completed within the specified
236-- time interval then emit whatever we have collected till now. The chunk
237-- timeout is reset whenever a chunk is emitted. The granularity of the clock
240-- >>> s = Stream.delayPost 0.3 $ Stream.fromList [1..1000]
241-- >>> f = Stream.fold (Fold.drainBy print) $ Stream.chunksOfTimeout 5 1 Fold.toList s
244{-# INLINE chunksOfTimeout #-}
245chunksOfTimeout :: MonadAsync m
246 => Int -> Double -> Fold m a b -> Stream m a -> Stream m b
247chunksOfTimeout n timeout f =
250 0.1 False (const (return False)) timeout (Fold.take n f)
255------------------------------------------------------------------------------
256-- Windowed classification
257------------------------------------------------------------------------------
259-- TODO: To mark the position in space or time we can have Indexed or
260-- TimeStamped types. That can make it easy to deal with the position indices
263------------------------------------------------------------------------------
264-- Keyed Sliding Windows
265------------------------------------------------------------------------------
268{-# INLINABLE classifySlidingChunks #-}
270 :: (IsStream t, MonadAsync m, Ord k)
271 => Int -- ^ window size
272 -> Int -- ^ window slide
273 -> Fold m a b -- ^ Fold to be applied to window events
274 -> t m (k, a, Bool) -- ^ window key, data, close event
276classifySlidingChunks wsize wslide (Fold step initial extract) str
279-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
280-- slide the send window when an ack is received and we slide the receive
281-- window when a sequence is complete. Sliding is stateful in case of TCP,
282-- sliding releases the send buffer or makes data available to the user from
283-- the receive buffer.
284{-# INLINABLE classifySlidingSessions #-}
285classifySlidingSessions
286 :: (IsStream t, MonadAsync m, Ord k)
287 => Double -- ^ timer tick in seconds
288 -> Double -- ^ time window size
289 -> Double -- ^ window slide
290 -> Fold m a b -- ^ Fold to be applied to window events
291 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
293classifySlidingSessions tick interval slide (Fold step initial extract) str
297------------------------------------------------------------------------------
298-- Sliding Window Buffers
299------------------------------------------------------------------------------
301-- These buffered versions could be faster than concurrent incremental folds of
302-- all overlapping windows as in many cases we may not need all the values to
303-- compute the fold, we can just compute the result using the old value and new
304-- value. However, we may need the buffer once in a while, for example for
305-- string search we usually compute the hash incrementally but when the hash
306-- matches the hash of the pattern we need to compare the whole string.
308-- XXX we should be able to implement sequence based splitting combinators
309-- using this combinator.
312-- | Buffer n elements of the input in a ring buffer. When t new elements are
313-- collected, slide the window to remove the same number of oldest elements,
314-- insert the new elements, and apply an incremental fold on the sliding
315-- window, supplying the outgoing elements, the new ring buffer as arguments.
317 :: (IsStream t, Monad m, Ord a, Unboxed a)
318 => Int -- window size
319 -> Int -- window slide
320 -> Fold m (Ring a, Array a) b
323slidingChunkBuffer = undefined
325-- Buffer n seconds worth of stream elements of the input in a radix tree.
326-- Every t seconds, remove the items that are older than n seconds, and apply
327-- an incremental fold on the sliding window, supplying the outgoing elements,
328-- and the new radix tree buffer as arguments.
330 :: (IsStream t, Monad m, Ord a, Unboxed a)
331 => Int -- window size
333 -> Fold m (RTree a, Array a) b
336slidingSessionBuffer = undefined
339------------------------------------------------------------------------------
340-- Keyed Session Windows
341------------------------------------------------------------------------------
344-- | Keyed variable size space windows. Close the window if we do not receive a
345-- window event in the next "spaceout" elements.
346{-# INLINABLE classifyChunksBy #-}
348 :: (IsStream t, MonadAsync m, Ord k)
349 => Int -- ^ window spaceout (spread)
350 -> Bool -- ^ reset the spaceout when a chunk window element is received
351 -> Fold m a b -- ^ Fold to be applied to chunk window elements
352 -> t m (k, a, Bool) -- ^ chunk key, data, last element
354classifyChunksBy spanout reset (Fold step initial extract) str = undefined
356-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
357-- received within the chunk size window. The chunk gets closed only if no
358-- element is received within the chunk window.
360{-# INLINABLE classifyKeepAliveChunks #-}
361classifyKeepAliveChunks
362 :: (IsStream t, MonadAsync m, Ord k)
363 => Int -- ^ window spaceout (spread)
364 -> Fold m a b -- ^ Fold to be applied to chunk window elements
365 -> t m (k, a, Bool) -- ^ chunk key, data, last element
367classifyKeepAliveChunks spanout = classifyChunksBy spanout True
370data SessionState t m f s b = SessionState
371 { sessionCurTime :: !AbsTime -- ^ time since last event
372 , sessionEventTime :: !AbsTime -- ^ time as per last event
373 -- We can use the Map size instead of maintaining a count, but if we have
374 -- to switch to HashMap then it can be useful.
375 , sessionCount :: !Int -- ^ total number of sessions in progress
376 , sessionTimerHeap :: H.Heap (H.Entry AbsTime (Key f)) -- ^ heap for timeouts
377 , sessionKeyValueMap :: f s -- ^ Stored sessions for keys
378 , sessionOutputStream :: t (m :: Type -> Type) (Key f, b) -- ^ Completed sessions
381data SessionEntry s = LiveSession !AbsTime !s | ZombieSession
383-- delete from map and output the fold accumulator
384ejectEntry :: (Monad m, IsMap f) =>
388 -> Stream m (Key f, b)
392 -> m (heap, f entry, Stream m (Key f, b), Int)
393ejectEntry extract hp mp out cnt acc key = do
395 let out1 = (key, sess) `Stream.cons` out
396 let mp1 = IsMap.mapDelete key mp
397 return (hp, mp1, out1, cnt - 1)
399{-# NOINLINE flush #-}
400flush :: (Monad m, IsMap f) =>
402 -> SessionState Stream m f (SessionEntry s) b
403 -> m (SessionState Stream m f (SessionEntry s) b)
404flush extract session@SessionState{..} = do
405 (hp', mp', out, count) <-
413 { sessionCount = count
414 , sessionTimerHeap = hp'
415 , sessionKeyValueMap = mp'
416 , sessionOutputStream = out
421 ejectAll (hp, mp, out, !cnt) = do
422 let hres = H.uncons hp
424 Just (Entry _ key, hp1) -> do
425 r <- case IsMap.mapLookup key mp of
426 Nothing -> return (hp1, mp, out, cnt)
427 Just ZombieSession ->
428 return (hp1, IsMap.mapDelete key mp, out, cnt)
429 Just (LiveSession _ acc) ->
430 ejectEntry extract hp1 mp out cnt acc key
433 assert (IsMap.mapNull mp) (return ())
434 return (hp, mp, out, cnt)
436{-# NOINLINE ejectOne #-}
437ejectOne :: (IsMap f, Monad m) =>
440 -> ( H.Heap (Entry AbsTime (Key f))
441 , f (SessionEntry acc)
442 , Stream m (Key f, b)
445 -> m ( H.Heap (Entry AbsTime (Key f))
446 , f (SessionEntry acc)
447 , Stream m (Key f, b), Int
449ejectOne reset extract = go
453 go (hp, mp, out, !cnt) = do
454 let hres = H.uncons hp
456 Just (Entry expiry key, hp1) ->
457 case IsMap.mapLookup key mp of
458 Nothing -> go (hp1, mp, out, cnt)
459 Just ZombieSession ->
460 go (hp1, IsMap.mapDelete key mp, out, cnt)
461 Just (LiveSession expiry1 acc) -> do
462 if not reset || expiry1 <= expiry
463 then ejectEntry extract hp1 mp out cnt acc key
465 -- reset the session timeout and continue
466 let hp2 = H.insert (Entry expiry1 key) hp1
467 in go (hp2, mp, out, cnt)
469 assert (IsMap.mapNull mp) (return ())
470 return (hp, mp, out, cnt)
472{-# NOINLINE ejectExpired #-}
473ejectExpired :: (IsMap f, Monad m) =>
477 -> SessionState Stream m f (SessionEntry acc) b
479 -> m (SessionState Stream m f (SessionEntry acc) b)
480ejectExpired reset ejectPred extract session@SessionState{..} curTime = do
481 (hp', mp', out, count) <-
483 sessionTimerHeap sessionKeyValueMap Stream.nil sessionCount
485 { sessionCurTime = curTime
486 , sessionCount = count
487 , sessionTimerHeap = hp'
488 , sessionKeyValueMap = mp'
489 , sessionOutputStream = out
494 ejectLoop hp mp out !cnt = do
495 let hres = H.uncons hp
497 Just (Entry expiry key, hp1) -> do
500 then return (True, False)
506 case IsMap.mapLookup key mp of
507 Nothing -> ejectLoop hp1 mp out cnt
508 Just ZombieSession ->
509 ejectLoop hp1 (IsMap.mapDelete key mp) out cnt
510 Just (LiveSession expiry1 acc) -> do
511 if expiry1 <= curTime || not reset || force
513 (hp2,mp1,out1,cnt1) <-
514 ejectEntry extract hp1 mp out cnt acc key
515 ejectLoop hp2 mp1 out1 cnt1
517 -- reset the session timeout and continue
518 let hp2 = H.insert (Entry expiry1 key) hp1
519 in ejectLoop hp2 mp out cnt
520 else return (hp, mp, out, cnt)
522 assert (IsMap.mapNull mp) (return ())
523 return (hp, mp, out, cnt)
525-- XXX Use mutable IORef in accumulator
526{-# INLINE classifySessionsByGeneric #-}
527classifySessionsByGeneric
528 :: forall m f a b. (MonadAsync m, IsMap f)
529 => Proxy (f :: (Type -> Type))
530 -> Double -- ^ timer tick in seconds
531 -> Bool -- ^ reset the timer when an event is received
532 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
533 -> Double -- ^ session timeout in seconds
534 -> Fold m a b -- ^ Fold to be applied to session data
535 -> Stream m (AbsTime, (Key f, a)) -- ^ timestamp, (session key, session
537 -> Stream m (Key f, b) -- ^ session key, fold result
538classifySessionsByGeneric _ tick reset ejectPred tmout
539 (Fold step initial extract) input =
540 Stream.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)
541 $ Stream.scanlMAfter' sstep (return szero) (flush extract)
542 $ interjectSuffix tick (return Nothing)
547 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
548 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
550 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
551 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
553 , sessionTimerHeap = H.empty
554 , sessionKeyValueMap = IsMap.mapEmpty :: f s
555 , sessionOutputStream = Stream.nil
558 -- We can eject sessions based on the current session count to limit
559 -- memory consumption. There are two possible strategies:
561 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
562 -- threshold even before timeout, effectively reduce the timeout.
563 -- 2) Drop creation of new sessions but keep accepting new events for the
566 -- We use the first strategy as of now.
568 -- Got a new stream input element
569 sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
570 -- XXX instead of a heap we could use a timer wheel.
572 -- XXX if the key is an Int, we can also use an IntMap for slightly
573 -- better performance.
577 -- Values for each key are collected in a map using the supplied fold.
578 -- When we insert a key in the Map we insert an entry into the heap as
579 -- well with the session expiry as the sort key. The Map entry
580 -- consists of the fold result, and the expiry time of the session. If
581 -- "reset" is True the expiry time is readjusted whenever a new event
582 -- is processed. If the fold terminates and a new session is started
583 -- for the same key the expiry time is set to the first timestamp of
586 -- The heap must have at most one entry for any given key. The heap is
587 -- processed periodically to remove the expired entries. We pick up an
588 -- expired entry from the top of the heap and if the session has
589 -- expired based on the expiry time in the Map entry then we remove the
590 -- session from the Map and yield its fold output. Otherwise, we
591 -- reinsert the entry into the heap based on the current expiry in the
594 -- If an entry is removed from the Map and not removed from the heap
595 -- and in the meantime it is inserted again in the Map (using the same
596 -- key) then how do we avoid inserting multiple entries in the heap?
597 -- For this reason we maintain the invariant that the Map entry is
598 -- removed only when the heap entry is removed. Even if the fold has
599 -- finished we still keep a dummy Map entry (ZombieSession) until the
600 -- heap entry is removed. That way if we have a Map entry we do not
601 -- insert a heap entry because we know it is already there.
602 -- XXX The ZombieSession mechanism does not work as expected as we
603 -- ignore ZombieSession when inserting a new entry. Anyway, we can
604 -- remove this mechanism as at most only two heap entries may be
605 -- created and they will be ultimately cleaned up.
607 -- Heap processing needs the map and map processing needs the heap,
608 -- therefore we cannot separate the two for modularity unless we have a
609 -- way to achieve mutual recursion.
611 let curTime = max sessionEventTime timestamp
612 mOld = IsMap.mapLookup key sessionKeyValueMap
614 -- deleting a key from the heap is expensive, so we never
615 -- delete a key from heap, we just purge it from the Map and it
616 -- gets purged from the heap on timeout. We just need an extra
617 -- lookup in the Map when the key is purged from the heap, that
618 -- should not be expensive.
620 let (mp, cnt) = case mOld of
621 Just (LiveSession _ _) ->
623 key ZombieSession sessionKeyValueMap
626 _ -> (sessionKeyValueMap, sessionCount)
628 { sessionCurTime = curTime
629 , sessionEventTime = curTime
631 , sessionKeyValueMap = mp
632 , sessionOutputStream = Stream.fromPure (key, fb)
635 let expiry = addToAbsTime timestamp timeoutMs
636 (hp1, mp1, out1, cnt1) <- do
637 let vars = (sessionTimerHeap, sessionKeyValueMap,
638 Stream.nil, sessionCount)
640 -- inserting new entry
642 -- Eject a session from heap and map if needed
643 eject <- ejectPred sessionCount
644 (hp, mp, out, cnt) <-
646 then ejectOne reset extract vars
649 -- Insert the new session in heap
650 let hp' = H.insert (Entry expiry key) hp
651 in return (hp', mp, out, cnt + 1)
652 -- updating old entry
653 Just _ -> return vars
655 let acc = LiveSession expiry fs1
656 mp2 = IsMap.mapInsert key acc mp1
657 return $ SessionState
658 { sessionCurTime = curTime
659 , sessionEventTime = curTime
660 , sessionCount = cnt1
661 , sessionTimerHeap = hp1
662 , sessionKeyValueMap = mp2
663 , sessionOutputStream = out1
667 Just (LiveSession _ acc) -> return $ Fold.Partial acc
671 error $ "classifySessionsBy: "
672 ++ "The supplied fold must consume at least one input"
673 Fold.Partial fs -> do
676 Fold.Done fb -> done fb
677 Fold.Partial fs1 -> partial fs1
679 -- Got a timer tick event
680 sstep sessionState@SessionState{..} Nothing =
681 let curTime = addToAbsTime sessionCurTime tickMs
682 in ejectExpired reset ejectPred extract sessionState curTime
684-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
685-- classifies an input event @stream@ consisting of @(timestamp, (key,
686-- value))@ into sessions based on the @key@, folding all the values
687-- corresponding to the same key into a session using the supplied @fold@.
689-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
690-- session key and the folded value is emitted in the output stream. The
691-- timeout is measured from the first event in the session. If the @keepalive@
692-- option is set to 'True' the timeout is reset to 0 whenever an event is
695-- The @timestamp@ in the input stream is an absolute time from some epoch,
696-- characterizing the time when the input event was generated. The notion of
697-- current time is maintained by a monotonic event time clock using the
698-- timestamps seen in the input stream. The latest timestamp seen till now is
699-- used as the base for the current time. When no new events are seen, a timer
700-- is started with a clock resolution of @tick@ seconds. This timer is used to
701-- detect session timeouts in the absence of new events.
703-- To ensure an upper bound on the memory used the number of sessions can be
704-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
705-- oldest session is ejected before inserting a new session.
707-- When the stream ends any buffered sessions are ejected immediately.
709-- If a session key is received even after a session has finished, another
710-- session is created for that key.
713-- Stream.fold (Fold.drainBy print)
714-- $ Concur.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
715-- $ Stream.timestamped
717-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
724{-# INLINE classifySessionsBy #-}
726 :: (MonadAsync m, Ord k)
727 => Double -- ^ timer tick in seconds
728 -> Bool -- ^ reset the timer when an event is received
729 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
730 -> Double -- ^ session timeout in seconds
731 -> Fold m a b -- ^ Fold to be applied to session data
732 -> Stream m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
733 -> Stream m (k, b) -- ^ session key, fold result
734classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k))
736-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
737-- option set to 'True'.
740-- classifyKeepAliveSessions = classifySessionsBy 1 True
745{-# INLINE classifyKeepAliveSessions #-}
746classifyKeepAliveSessions ::
747 (MonadAsync m, Ord k)
748 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
749 -> Double -- ^ session inactive timeout
750 -> Fold m a b -- ^ Fold to be applied to session payload data
751 -> Stream m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
753classifyKeepAliveSessions = classifySessionsBy 1 True
755------------------------------------------------------------------------------
756-- Keyed tumbling windows
757------------------------------------------------------------------------------
759-- Tumbling windows is a special case of sliding windows where the window slide
760-- is the same as the window size. Or it can be a special case of session
761-- windows where the reset flag is set to False.
763-- XXX instead of using the early termination flag in the stream, we can use an
764-- early terminating fold instead.
767-- | Split the stream into fixed size chunks of specified size. Within each
768-- such chunk fold the elements in buckets identified by the keys. A particular
769-- bucket fold can be terminated early if a closing flag is encountered in an
770-- element for that key.
773{-# INLINABLE classifyChunksOf #-}
775 :: (IsStream t, MonadAsync m, Ord k)
776 => Int -- ^ window size
777 -> Fold m a b -- ^ Fold to be applied to window events
778 -> t m (k, a, Bool) -- ^ window key, data, close event
780classifyChunksOf wsize = classifyChunksBy wsize False
783-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
784-- option set to 'False'.
787-- classifySessionsOf = classifySessionsBy 1 False
792{-# INLINE classifySessionsOf #-}
794 (MonadAsync m, Ord k)
795 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
796 -> Double -- ^ time window size
797 -> Fold m a b -- ^ Fold to be applied to session data
798 -> Stream m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
800classifySessionsOf = classifySessionsBy 1 False
802------------------------------------------------------------------------------
804------------------------------------------------------------------------------
806-- | Continuously evaluate the input stream and sample the last event in time
807-- window of @n@ seconds.
809-- This is also known as @throttle@ in some libraries.
812-- sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
817{-# INLINE sampleIntervalEnd #-}
818sampleIntervalEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
819sampleIntervalEnd n = Stream.catMaybes . intervalsOf n Fold.last
821-- | Like 'sampleInterval' but samples at the beginning of the time window.
824-- sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
829{-# INLINE sampleIntervalStart #-}
830sampleIntervalStart :: MonadAsync m => Double -> Stream m a -> Stream m a
831sampleIntervalStart n = Stream.catMaybes . intervalsOf n Fold.one
837 | BurstDoneNext !x !t !x
839{-# INLINE sampleBurst #-}
840sampleBurst :: MonadAsync m => Bool -> Double -> Stream m a -> Stream m a
841sampleBurst sampleAtEnd gap xs =
842 -- XXX Ideally we should schedule a timer event exactly after gap time,
843 -- but the tick stream should work well as long as the timer
844 -- granularity is small enough compared to the gap.
845 Stream.mapMaybe extract
846 $ Stream.scan (Fold.foldl' step BurstNone)
848 $ interjectSuffix 0.01 (return Nothing) (fmap Just xs)
852 gap1 = toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
855 step BurstNone (t1, Just x1) = BurstWait t1 x1
856 step BurstNone _ = BurstNone
858 step (BurstDone _) (t1, Just x1) = BurstWait t1 x1
859 step (BurstDone _) _ = BurstNone
861 step old@(BurstWait t0 x0) (t1, Nothing)
862 | t1 - t0 >= gap1 = BurstDone x0
864 -- This can happen due to scheduling delays, if we received back to
865 -- back events spaced by more than the timeout without an
866 -- intervening timeout event then we emit the old event instead of
867 -- replacing it by the new.
868 step (BurstWait t0 x0) (t1, Just x1)
869 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
870 | sampleAtEnd = BurstWait t1 x1
871 | otherwise = BurstWait t1 x0
873 step (BurstDoneNext _ t0 x0) (t1, Nothing)
874 | t1 - t0 >= gap1 = BurstDone x0
875 | otherwise = BurstWait t0 x0
876 step (BurstDoneNext _ t0 x0) (t1, Just x1)
877 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
878 | sampleAtEnd = BurstWait t1 x1
879 | otherwise = BurstWait t1 x0
881 {-# INLINE extract #-}
882 extract (BurstDoneNext x _ _) = Just x
883 extract (BurstDone x) = Just x
886-- | Sample one event at the end of each burst of events. A burst is a group
887-- of events close together in time, it ends when an event is spaced by more
888-- than the specified time interval (in seconds) from the previous event.
890-- This is known as @debounce@ in some libraries.
892-- The clock granularity is 10 ms.
896{-# INLINE sampleBurstEnd #-}
897sampleBurstEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
898sampleBurstEnd = sampleBurst True
900-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
901-- instead of at the end of it.
905{-# INLINE sampleBurstStart #-}
906sampleBurstStart :: MonadAsync m => Double -> Stream m a -> Stream m a
907sampleBurstStart = sampleBurst False
909------------------------------------------------------------------------------
911------------------------------------------------------------------------------
913-- XXX We could use 'maxBuffer Block/Drop/Rotate/Sample' instead. However we
914-- may want to have the evaluation rate independent of the sampling rate. To
915-- support that we can decouple evaluation and sampling in independent stages.
916-- The sampling stage would strictly evaluate and sample, the evaluation stage
917-- would control the evaluation rate.
919-- | Evaluate the input stream continuously and keep only the oldest @n@
920-- elements in the buffer, discard the new ones when the buffer is full. When
921-- the output stream is evaluated it consumes the values from the buffer in a
926{-# INLINE sampleOld #-}
927sampleOld :: -- MonadAsync m =>
928 Int -> Stream m a -> Stream m a
931-- | Evaluate the input stream continuously and keep only the latest @n@
932-- elements in a ring buffer, keep discarding the older ones to make space for
933-- the new ones. When the output stream is evaluated it consumes the values
934-- from the buffer in a FIFO manner.
938{-# INLINE sampleNew #-}
939sampleNew :: -- MonadAsync m =>
940 Int -> Stream m a -> Stream m a
943-- | Like 'sampleNew' but samples at uniform intervals to match the consumer
944-- rate. Note that 'sampleNew' leads to non-uniform sampling depending on the
949{-# INLINE sampleRate #-}
950sampleRate :: -- MonadAsync m =>
951 Double -> Stream m a -> Stream m a
952sampleRate = 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