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
9module Streamly.Internal.Data.Stream.Time
11 -- * Imports for Examples
15 periodic -- XXX Should go to Streamly.Data.Stream
16 , ticks -- XXX Should go to Streamly.Data.Stream
37 -- * Windowed Sessions
38 , classifySessionsByGeneric
41 , classifyKeepAliveSessions
43 -- XXX This should go in the concurrent module
45 -- | Evaluate strictly using a buffer of results. When the buffer becomes
46 -- full we can block, drop the new elements, drop the oldest element and
47 -- insert the new at the end.
54import Control.Concurrent (threadDelay)
55import Control.Exception (assert)
56import Control.Monad.IO.Class (MonadIO(..))
57import Data.Heap (Entry(..))
58import Data.Kind (Type)
60import Data.Maybe (isNothing)
61import Data.Proxy (Proxy(..))
62import Streamly.Data.Fold (Fold)
63import Streamly.Internal.Data.Fold (Fold (..))
64import Streamly.Internal.Data.IsMap (IsMap(..))
65import Streamly.Internal.Data.Stream (Stream)
66import Streamly.Internal.Data.Time.Units
73import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)
75import qualified Data.Heap as H
76import qualified Streamly.Data.Fold as Fold
77import qualified Streamly.Data.Stream as Stream
78import qualified Streamly.Data.Unfold as Unfold
79import qualified Streamly.Internal.Data.Fold as Fold (Step(..))
80import qualified Streamly.Internal.Data.IsMap as IsMap
81import qualified Streamly.Internal.Data.Stream as Stream
87import Streamly.Internal.Data.Stream.Concurrent
91-- Imports for example snippets in this module.
94-- >>> import Control.Concurrent (threadDelay)
95-- >>> import qualified Streamly.Data.Array as Array
96-- >>> import qualified Streamly.Data.Fold as Fold
97-- >>> import qualified Streamly.Data.Parser as Parser
98-- >>> import qualified Streamly.Data.Stream as Stream
99-- >>> import qualified Streamly.Data.Stream.Prelude as Stream
100-- >>> import qualified Streamly.Internal.Data.Stream as Stream (delayPost, timestamped)
101-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Stream (parListEagerFst)
102-- >>> import qualified Streamly.Internal.Data.Stream.Time as Stream
103-- >>> import Prelude hiding (concatMap, concat)
106-- threadDelay (n * 1000000) -- sleep for n seconds
107-- putStrLn (show n ++ " sec") -- print "n sec"
111--------------------------------------------------------------------------------
113--------------------------------------------------------------------------------
115-- | Generate a stream by running an action periodically at the specified time
118{-# INLINE periodic #-}
119periodic :: MonadIO m => m a -> Double -> Stream m a
120periodic action n = Stream.repeatM timed
124 timed = liftIO (threadDelay (round $ n * 1000000)) >> action
126-- | Generate a tick stream consisting of '()' elements, each tick is generated
127-- after the specified time delay given in seconds.
129-- >>> ticks = Stream.periodic (return ())
132ticks :: MonadIO m => Double -> Stream m ()
133ticks = periodic (return ())
135-- | Generate a tick stream, ticks are generated at the specified 'Rate'. The
136-- rate is adaptive, the tick generation speed can be increased or decreased at
137-- different times to achieve the specified rate. The specific behavior for
138-- different styles of 'Rate' specifications is documented under 'Rate'. The
139-- effective maximum rate achieved by a stream is governed by the processor
142-- >>> tickStream = Stream.repeatM (return ())
143-- >>> ticksRate r = Stream.parEval (Stream.rate (Just r)) tickStream
145{-# INLINE ticksRate #-}
146ticksRate :: MonadAsync m => Rate -> Stream m ()
147ticksRate r = parEval (rate (Just r)) $ Stream.repeatM (return ())
149-- XXX The case when the interval is 0, we should run only the stream being
152-- | Intersperse a monadic action into the input stream after every @n@
157-- >>> interject n f xs = Stream.parListEagerFst [xs, Stream.periodic f n]
161-- >>> s = Stream.fromList "hello"
162-- >>> input = Stream.mapM (\x -> threadDelay 1000000 >> putChar x) s
163-- >>> Stream.fold Fold.drain $ Stream.interject (putChar ',') 1.05 input
166{-# INLINE interject #-}
167interject :: MonadAsync m => m a -> Double -> Stream m a -> Stream m a
168interject f n xs = parListEagerFst [xs, periodic f n]
170-- XXX No element should be yielded if the duration is zero.
172-- | @takeInterval interval@ runs the stream only upto the specified time
173-- @interval@ in seconds.
175-- The interval starts when the stream is evaluated for the first time.
177{-# INLINE takeInterval #-}
178takeInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
181 . Stream.takeWhile isNothing
182 . interject (return Nothing) d . fmap Just
184-- | Take time interval @i@ seconds at the end of the stream.
186-- O(n) space, where n is the number elements taken.
189{-# INLINE takeLastInterval #-}
190takeLastInterval :: -- MonadAsync m =>
191 Double -> Stream m a -> Stream m a
192takeLastInterval = undefined
194-- XXX All elements should be yielded if the duration is zero.
196-- | @dropInterval interval@ drops all the stream elements that are generated
197-- before the specified @interval@ in seconds has passed.
199-- The interval begins when the stream is evaluated for the first time.
201{-# INLINE dropInterval #-}
202dropInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
205 . Stream.dropWhile isNothing
206 . interject (return Nothing) d . fmap Just
208-- | Drop time interval @i@ seconds at the end of the stream.
210-- O(n) space, where n is the number elements dropped.
213{-# INLINE dropLastInterval #-}
214dropLastInterval :: -- MonadAsync m =>
215 Int -> Stream m a -> Stream m a
216dropLastInterval = undefined
218-- | Group the input stream into windows of @n@ second each and then fold each
219-- group using the provided fold function.
221-- >>> twoPerSec = Stream.parEval (Stream.constRate 2) $ Stream.enumerateFrom 1
222-- >>> intervals = Stream.intervalsOf 1 Fold.toList twoPerSec
223-- >>> Stream.fold Fold.toList $ Stream.take 2 intervals
226{-# INLINE intervalsOf #-}
227intervalsOf :: MonadAsync m => Double -> Fold m a b -> Stream m a -> Stream m b
230 (Fold.takeEndBy isNothing (Fold.catMaybes f))
231 (interject (return Nothing) n (fmap Just xs))
233-- XXX This can be implemented more efficiently by sharing a Clock.Timer across
234-- parallel threads and resetting it whenever a span is emitted.
236-- | Like 'chunksOf' but if the chunk is not completed within the specified
237-- time interval then emit whatever we have collected till now. The chunk
238-- timeout is reset whenever a chunk is emitted. The granularity of the clock
241-- >>> s = Stream.delayPost 0.3 $ Stream.fromList [1..1000]
242-- >>> f = Stream.fold (Fold.drainMapM print) $ Stream.groupsOfTimeout 5 1 Fold.toList s
245{-# INLINE groupsOfTimeout #-}
246groupsOfTimeout :: MonadAsync m
247 => Int -> Double -> Fold m a b -> Stream m a -> Stream m b
248groupsOfTimeout n timeout f =
251 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 = Stream.cons (key, sess) 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 $ interject (return Nothing) tick
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.drainMapM print)
714-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
715-- $ Stream.timestamped
717-- $ Stream.fromList ((,) <$> [1,2,3] <*> ['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'.
786-- >>> classifySessionsOf = Stream.classifySessionsBy 1 False
790{-# INLINE classifySessionsOf #-}
792 (MonadAsync m, Ord k)
793 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
794 -> Double -- ^ time window size
795 -> Fold m a b -- ^ Fold to be applied to session data
796 -> Stream m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
798classifySessionsOf = classifySessionsBy 1 False
800------------------------------------------------------------------------------
802------------------------------------------------------------------------------
804-- | Continuously evaluate the input stream and sample the last event in each
805-- time window of @n@ seconds.
807-- This is also known as @throttle@ in some libraries.
809-- >>> sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.latest
811{-# INLINE sampleIntervalEnd #-}
812sampleIntervalEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
813sampleIntervalEnd n = Stream.catMaybes . intervalsOf n Fold.latest
815-- | Like 'sampleInterval' but samples at the beginning of the time window.
817-- >>> sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
819{-# INLINE sampleIntervalStart #-}
820sampleIntervalStart :: MonadAsync m => Double -> Stream m a -> Stream m a
821sampleIntervalStart n = Stream.catMaybes . intervalsOf n Fold.one
827 | BurstDoneNext !x !t !x
829{-# INLINE sampleBurst #-}
830sampleBurst :: MonadAsync m => Bool -> Double -> Stream m a -> Stream m a
831sampleBurst sampleAtEnd gap xs =
832 -- XXX Ideally we should schedule a timer event exactly after gap time,
833 -- but the tick stream should work well as long as the timer
834 -- granularity is small enough compared to the gap.
835 Stream.mapMaybe extract
836 $ Stream.scan (Fold.foldl' step BurstNone)
838 $ interject (return Nothing) 0.01 (fmap Just xs)
842 gap1 = toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
845 step BurstNone (t1, Just x1) = BurstWait t1 x1
846 step BurstNone _ = BurstNone
848 step (BurstDone _) (t1, Just x1) = BurstWait t1 x1
849 step (BurstDone _) _ = BurstNone
851 step old@(BurstWait t0 x0) (t1, Nothing)
852 | t1 - t0 >= gap1 = BurstDone x0
854 -- This can happen due to scheduling delays, if we received back to
855 -- back events spaced by more than the timeout without an
856 -- intervening timeout event then we emit the old event instead of
857 -- replacing it by the new.
858 step (BurstWait t0 x0) (t1, Just x1)
859 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
860 | sampleAtEnd = BurstWait t1 x1
861 | otherwise = BurstWait t1 x0
863 step (BurstDoneNext _ t0 x0) (t1, Nothing)
864 | t1 - t0 >= gap1 = BurstDone x0
865 | otherwise = BurstWait t0 x0
866 step (BurstDoneNext _ t0 x0) (t1, Just x1)
867 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
868 | sampleAtEnd = BurstWait t1 x1
869 | otherwise = BurstWait t1 x0
871 {-# INLINE extract #-}
872 extract (BurstDoneNext x _ _) = Just x
873 extract (BurstDone x) = Just x
876-- | Sample one event at the end of each burst of events. A burst is a group
877-- of events close together in time, it ends when an event is spaced by more
878-- than the specified time interval (in seconds) from the previous event.
880-- This is known as @debounce@ in some libraries.
882-- The clock granularity is 10 ms.
884{-# INLINE sampleBurstEnd #-}
885sampleBurstEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
886sampleBurstEnd = sampleBurst True
888-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
889-- instead of at the end of it.
891{-# INLINE sampleBurstStart #-}
892sampleBurstStart :: MonadAsync m => Double -> Stream m a -> Stream m a
893sampleBurstStart = sampleBurst False
895------------------------------------------------------------------------------
897------------------------------------------------------------------------------
899-- XXX We could use 'maxBuffer Block/Drop/Rotate/Sample' instead. However we
900-- may want to have the evaluation rate independent of the sampling rate. To
901-- support that we can decouple evaluation and sampling in independent stages.
902-- The sampling stage would strictly evaluate and sample, the evaluation stage
903-- would control the evaluation rate.
905-- bufferLatest - evaluate retaining only the latest element. Fork a thread
906-- that keeps writing the stream to an IORef, another parallel thread reads
909-- bufferLatestN - Fork a thread that keeps writing to a sliding ring buffer,
910-- and the stream unfolds the entire buffer when it is read, and the empty
911-- buffer starts filling again in the same way.
913-- bufferOldestN - Fork a thread that keeps writing to a buffer until it is
914-- filled and keeps dropping once it is filled. When the stream is read the
915-- 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@
918-- elements in the buffer, discard the new ones when the buffer is full. When
919-- the output stream is evaluated the collected buffer is streamed and the
920-- buffer starts filling again.
924{-# INLINE bufferOldestN #-}
925bufferOldestN :: -- MonadAsync m =>
926 Int -> Stream m a -> Stream m a
927bufferOldestN = undefined
929-- | 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
931-- the new ones. When the output stream is evaluated the buffer collected till
932-- now is streamed and it starts filling again.
936{-# INLINE bufferLatestN #-}
937bufferLatestN :: -- MonadAsync m =>
938 Int -> Stream m a -> Stream m a
939bufferLatestN = undefined
941-- | Always produce the latest available element from the stream without any
942-- delay. The stream is continuously evaluated at the highest possible rate and
943-- only the latest element is retained for sampling.
947{-# INLINE bufferLatest #-}
948bufferLatest :: -- MonadAsync m =>
949 Stream m a -> Stream m (Maybe a)
950bufferLatest = 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