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
20 , classifySessionsByGeneric
23 , classifyKeepAliveSessions
34import Control.Concurrent (threadDelay)
35import Control.Exception (assert)
36import Control.Monad.IO.Class (MonadIO(..))
37import Data.Heap (Entry(..))
38import Data.Kind (Type)
40import Data.Maybe (isNothing)
41import Data.Proxy (Proxy(..))
42import Streamly.Data.Fold (Fold)
43import Streamly.Internal.Data.Fold.Type (Fold (..))
44import Streamly.Internal.Data.IsMap (IsMap(..))
45import Streamly.Internal.Data.Stream.Type (Stream)
46import Streamly.Internal.Data.Time.Units
53import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)
55import qualified Data.Heap as H
56import qualified Streamly.Data.Fold as Fold
57import qualified Streamly.Data.Stream as Stream
58import qualified Streamly.Data.Unfold as Unfold
59import qualified Streamly.Internal.Data.Fold as Fold (Step(..))
60import qualified Streamly.Internal.Data.IsMap as IsMap
61import qualified Streamly.Internal.Data.Stream as Stream
68import Streamly.Internal.Data.Stream.Concurrent
72-- Imports for example snippets in this module.
75-- >>> import Control.Concurrent (threadDelay)
76-- >>> import qualified Streamly.Data.Array.Unboxed as Array
77-- >>> import qualified Streamly.Data.Fold as Fold
78-- >>> import qualified Streamly.Data.Parser as Parser
79-- >>> import qualified Streamly.Data.Stream as Stream
80-- >>> import qualified Streamly.Internal.Data.Stream as Stream (timestamped)
81-- >>> import qualified Streamly.Internal.Data.Stream.Concurrent as Concur
82-- >>> import qualified Streamly.Internal.Data.Stream.Time as Concur
83-- >>> import Prelude hiding (concatMap, concat)
86-- threadDelay (n * 1000000) -- sleep for n seconds
87-- putStrLn (show n ++ " sec") -- print "n sec"
91--------------------------------------------------------------------------------
93--------------------------------------------------------------------------------
95-- | Intersperse a monadic action into the input stream after every @n@
98-- >>> Stream.fold Fold.drain $ Concur.interjectSuffix 1.05 (putChar ',') $ Stream.mapM (\x -> threadDelay 1000000 >> putChar x) $ Stream.fromList "hello"
102{-# INLINE interjectSuffix #-}
103interjectSuffix :: MonadAsync m => Double -> m a -> Stream m a -> Stream m a
104interjectSuffix n f xs = parallelFst [xs, repeatM timed]
108 timed = liftIO (threadDelay (round $ n * 1000000)) >> f
109 repeatM = Stream.sequence . Stream.repeat
111-- XXX Notes from D.takeByTime (which was removed)
112-- XXX using getTime in the loop can be pretty expensive especially for
113-- computations where iterations are lightweight. We have the following
116-- 1) Run a timeout thread updating a flag asynchronously and check that
117-- flag here, that way we can have a cheap termination check.
119-- 2) Use COARSE clock to get time with lower resolution but more efficiently.
121-- 3) Use rdtscp/rdtsc to get time directly from the processor, compute the
122-- termination value of rdtsc in the beginning and then in each iteration just
123-- get rdtsc and check if we should terminate.
126-- | @takeInterval duration@ yields stream elements upto specified time
127-- @duration@ in seconds. The duration starts when the stream is evaluated for
128-- the first time, before the first element is yielded. The time duration is
129-- checked before generating each element, if the duration has expired the
132-- The total time taken in executing the stream is guaranteed to be /at least/
133-- @duration@, however, because the duration is checked before generating an
134-- element, the upper bound is indeterminate and depends on the time taken in
135-- generating and processing the last element.
137-- No element is yielded if the duration is zero. At least one element is
138-- yielded if the duration is non-zero.
142{-# INLINE takeInterval #-}
143takeInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
146 . Stream.takeWhile isNothing
147 . interjectSuffix d (return Nothing) . fmap Just
149-- | @dropInterval duration@ drops stream elements until specified @duration@ in
150-- seconds has passed. The duration begins when the stream is evaluated for the
151-- first time. The time duration is checked /after/ generating a stream element,
152-- the element is yielded if the duration has expired otherwise it is dropped.
154-- The time elapsed before starting to generate the first element is /at most/
155-- @duration@, however, because the duration expiry is checked after the
156-- element is generated, the lower bound is indeterminate and depends on the
157-- time taken in generating an element.
159-- All elements are yielded if the duration is zero.
163{-# INLINE dropInterval #-}
164dropInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
167 . Stream.dropWhile isNothing
168 . interjectSuffix d (return Nothing) . fmap Just
170-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
171-- XXX add this example after fixing the serial stream rate control
173-- | Group the input stream into windows of @n@ second each and then fold each
174-- group using the provided fold function.
176-- >>> Stream.toList $ Stream.take 5 $ Stream.intervalsOf 1 Fold.sum $ Stream.constRate 2 $ Stream.enumerateFrom 1
177-- [...,...,...,...,...]
180{-# INLINE intervalsOf #-}
181intervalsOf :: MonadAsync m => Double -> Fold m a b -> Stream m a -> Stream m b
184 (Fold.takeEndBy isNothing (Fold.catMaybes f))
185 (interjectSuffix n (return Nothing) (fmap Just xs))
187-- XXX This can be implemented more efficiently by sharing a Clock.Timer across
188-- parallel threads and resetting it whenever a span is emitted.
190-- | Like 'chunksOf' but if the chunk is not completed within the specified
191-- time interval then emit whatever we have collected till now. The chunk
192-- timeout is reset whenever a chunk is emitted. The granularity of the clock
195-- >>> s = Stream.delayPost 0.3 $ Stream.fromList [1..1000]
196-- >>> f = Stream.fold (Fold.drainBy print) $ Stream.chunksOfTimeout 5 1 Fold.toList s
199{-# INLINE chunksOfTimeout #-}
200chunksOfTimeout :: MonadAsync m
201 => Int -> Double -> Fold m a b -> Stream m a -> Stream m b
202chunksOfTimeout n timeout f =
205 0.1 False (const (return False)) timeout (Fold.take n f)
210------------------------------------------------------------------------------
211-- Windowed classification
212------------------------------------------------------------------------------
214-- TODO: To mark the position in space or time we can have Indexed or
215-- TimeStamped types. That can make it easy to deal with the position indices
218------------------------------------------------------------------------------
219-- Keyed Sliding Windows
220------------------------------------------------------------------------------
223{-# INLINABLE classifySlidingChunks #-}
225 :: (IsStream t, MonadAsync m, Ord k)
226 => Int -- ^ window size
227 -> Int -- ^ window slide
228 -> Fold m a b -- ^ Fold to be applied to window events
229 -> t m (k, a, Bool) -- ^ window key, data, close event
231classifySlidingChunks wsize wslide (Fold step initial extract) str
234-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
235-- slide the send window when an ack is received and we slide the receive
236-- window when a sequence is complete. Sliding is stateful in case of TCP,
237-- sliding releases the send buffer or makes data available to the user from
238-- the receive buffer.
239{-# INLINABLE classifySlidingSessions #-}
240classifySlidingSessions
241 :: (IsStream t, MonadAsync m, Ord k)
242 => Double -- ^ timer tick in seconds
243 -> Double -- ^ time window size
244 -> Double -- ^ window slide
245 -> Fold m a b -- ^ Fold to be applied to window events
246 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
248classifySlidingSessions tick interval slide (Fold step initial extract) str
252------------------------------------------------------------------------------
253-- Sliding Window Buffers
254------------------------------------------------------------------------------
256-- These buffered versions could be faster than concurrent incremental folds of
257-- all overlapping windows as in many cases we may not need all the values to
258-- compute the fold, we can just compute the result using the old value and new
259-- value. However, we may need the buffer once in a while, for example for
260-- string search we usually compute the hash incrementally but when the hash
261-- matches the hash of the pattern we need to compare the whole string.
263-- XXX we should be able to implement sequence based splitting combinators
264-- using this combinator.
267-- | Buffer n elements of the input in a ring buffer. When t new elements are
268-- collected, slide the window to remove the same number of oldest elements,
269-- insert the new elements, and apply an incremental fold on the sliding
270-- window, supplying the outgoing elements, the new ring buffer as arguments.
272 :: (IsStream t, Monad m, Ord a, Unboxed a)
273 => Int -- window size
274 -> Int -- window slide
275 -> Fold m (Ring a, Array a) b
278slidingChunkBuffer = undefined
280-- Buffer n seconds worth of stream elements of the input in a radix tree.
281-- Every t seconds, remove the items that are older than n seconds, and apply
282-- an incremental fold on the sliding window, supplying the outgoing elements,
283-- and the new radix tree buffer as arguments.
285 :: (IsStream t, Monad m, Ord a, Unboxed a)
286 => Int -- window size
288 -> Fold m (RTree a, Array a) b
291slidingSessionBuffer = undefined
294------------------------------------------------------------------------------
295-- Keyed Session Windows
296------------------------------------------------------------------------------
299-- | Keyed variable size space windows. Close the window if we do not receive a
300-- window event in the next "spaceout" elements.
301{-# INLINABLE classifyChunksBy #-}
303 :: (IsStream t, MonadAsync m, Ord k)
304 => Int -- ^ window spaceout (spread)
305 -> Bool -- ^ reset the spaceout when a chunk window element is received
306 -> Fold m a b -- ^ Fold to be applied to chunk window elements
307 -> t m (k, a, Bool) -- ^ chunk key, data, last element
309classifyChunksBy spanout reset (Fold step initial extract) str = undefined
311-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
312-- received within the chunk size window. The chunk gets closed only if no
313-- element is received within the chunk window.
315{-# INLINABLE classifyKeepAliveChunks #-}
316classifyKeepAliveChunks
317 :: (IsStream t, MonadAsync m, Ord k)
318 => Int -- ^ window spaceout (spread)
319 -> Fold m a b -- ^ Fold to be applied to chunk window elements
320 -> t m (k, a, Bool) -- ^ chunk key, data, last element
322classifyKeepAliveChunks spanout = classifyChunksBy spanout True
325data SessionState t m f s b = SessionState
326 { sessionCurTime :: !AbsTime -- ^ time since last event
327 , sessionEventTime :: !AbsTime -- ^ time as per last event
328 -- We can use the Map size instead of maintaining a count, but if we have
329 -- to switch to HashMap then it can be useful.
330 , sessionCount :: !Int -- ^ total number of sessions in progress
331 , sessionTimerHeap :: H.Heap (H.Entry AbsTime (Key f)) -- ^ heap for timeouts
332 , sessionKeyValueMap :: f s -- ^ Stored sessions for keys
333 , sessionOutputStream :: t (m :: Type -> Type) (Key f, b) -- ^ Completed sessions
336data SessionEntry s = LiveSession !AbsTime !s | ZombieSession
338-- delete from map and output the fold accumulator
339ejectEntry :: (Monad m, IsMap f) =>
343 -> Stream m (Key f, b)
347 -> m (heap, f entry, Stream m (Key f, b), Int)
348ejectEntry extract hp mp out cnt acc key = do
350 let out1 = (key, sess) `Stream.cons` out
351 let mp1 = IsMap.mapDelete key mp
352 return (hp, mp1, out1, cnt - 1)
354{-# NOINLINE flush #-}
355flush :: (Monad m, IsMap f) =>
357 -> SessionState Stream m f (SessionEntry s) b
358 -> m (SessionState Stream m f (SessionEntry s) b)
359flush extract session@SessionState{..} = do
360 (hp', mp', out, count) <-
368 { sessionCount = count
369 , sessionTimerHeap = hp'
370 , sessionKeyValueMap = mp'
371 , sessionOutputStream = out
376 ejectAll (hp, mp, out, !cnt) = do
377 let hres = H.uncons hp
379 Just (Entry _ key, hp1) -> do
380 r <- case IsMap.mapLookup key mp of
381 Nothing -> return (hp1, mp, out, cnt)
382 Just ZombieSession ->
383 return (hp1, IsMap.mapDelete key mp, out, cnt)
384 Just (LiveSession _ acc) ->
385 ejectEntry extract hp1 mp out cnt acc key
388 assert (IsMap.mapNull mp) (return ())
389 return (hp, mp, out, cnt)
391{-# NOINLINE ejectOne #-}
392ejectOne :: (IsMap f, Monad m) =>
395 -> ( H.Heap (Entry AbsTime (Key f))
396 , f (SessionEntry acc)
397 , Stream m (Key f, b)
400 -> m ( H.Heap (Entry AbsTime (Key f))
401 , f (SessionEntry acc)
402 , Stream m (Key f, b), Int
404ejectOne reset extract = go
408 go (hp, mp, out, !cnt) = do
409 let hres = H.uncons hp
411 Just (Entry expiry key, hp1) ->
412 case IsMap.mapLookup key mp of
413 Nothing -> go (hp1, mp, out, cnt)
414 Just ZombieSession ->
415 go (hp1, IsMap.mapDelete key mp, out, cnt)
416 Just (LiveSession expiry1 acc) -> do
417 if not reset || expiry1 <= expiry
418 then ejectEntry extract hp1 mp out cnt acc key
420 -- reset the session timeout and continue
421 let hp2 = H.insert (Entry expiry1 key) hp1
422 in go (hp2, mp, out, cnt)
424 assert (IsMap.mapNull mp) (return ())
425 return (hp, mp, out, cnt)
427{-# NOINLINE ejectExpired #-}
428ejectExpired :: (IsMap f, Monad m) =>
432 -> SessionState Stream m f (SessionEntry acc) b
434 -> m (SessionState Stream m f (SessionEntry acc) b)
435ejectExpired reset ejectPred extract session@SessionState{..} curTime = do
436 (hp', mp', out, count) <-
438 sessionTimerHeap sessionKeyValueMap Stream.nil sessionCount
440 { sessionCurTime = curTime
441 , sessionCount = count
442 , sessionTimerHeap = hp'
443 , sessionKeyValueMap = mp'
444 , sessionOutputStream = out
449 ejectLoop hp mp out !cnt = do
450 let hres = H.uncons hp
452 Just (Entry expiry key, hp1) -> do
455 then return (True, False)
461 case IsMap.mapLookup key mp of
462 Nothing -> ejectLoop hp1 mp out cnt
463 Just ZombieSession ->
464 ejectLoop hp1 (IsMap.mapDelete key mp) out cnt
465 Just (LiveSession expiry1 acc) -> do
466 if expiry1 <= curTime || not reset || force
468 (hp2,mp1,out1,cnt1) <-
469 ejectEntry extract hp1 mp out cnt acc key
470 ejectLoop hp2 mp1 out1 cnt1
472 -- reset the session timeout and continue
473 let hp2 = H.insert (Entry expiry1 key) hp1
474 in ejectLoop hp2 mp out cnt
475 else return (hp, mp, out, cnt)
477 assert (IsMap.mapNull mp) (return ())
478 return (hp, mp, out, cnt)
480-- XXX Use mutable IORef in accumulator
481{-# INLINE classifySessionsByGeneric #-}
482classifySessionsByGeneric
483 :: forall m f a b. (MonadAsync m, IsMap f)
484 => Proxy (f :: (Type -> Type))
485 -> Double -- ^ timer tick in seconds
486 -> Bool -- ^ reset the timer when an event is received
487 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
488 -> Double -- ^ session timeout in seconds
489 -> Fold m a b -- ^ Fold to be applied to session data
490 -> Stream m (AbsTime, (Key f, a)) -- ^ timestamp, (session key, session
492 -> Stream m (Key f, b) -- ^ session key, fold result
493classifySessionsByGeneric _ tick reset ejectPred tmout
494 (Fold step initial extract) input =
495 Stream.unfoldMany (Unfold.lmap sessionOutputStream Unfold.fromStream)
496 $ Stream.scanlMAfter' sstep (return szero) (flush extract)
497 $ interjectSuffix tick (return Nothing)
502 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
503 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
505 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
506 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
508 , sessionTimerHeap = H.empty
509 , sessionKeyValueMap = IsMap.mapEmpty :: f s
510 , sessionOutputStream = Stream.nil
513 -- We can eject sessions based on the current session count to limit
514 -- memory consumption. There are two possible strategies:
516 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
517 -- threshold even before timeout, effectively reduce the timeout.
518 -- 2) Drop creation of new sessions but keep accepting new events for the
521 -- We use the first strategy as of now.
523 -- Got a new stream input element
524 sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
525 -- XXX instead of a heap we could use a timer wheel.
527 -- XXX if the key is an Int, we can also use an IntMap for slightly
528 -- better performance.
532 -- Values for each key are collected in a map using the supplied fold.
533 -- When we insert a key in the Map we insert an entry into the heap as
534 -- well with the session expiry as the sort key. The Map entry
535 -- consists of the fold result, and the expiry time of the session. If
536 -- "reset" is True the expiry time is readjusted whenever a new event
537 -- is processed. If the fold terminates and a new session is started
538 -- for the same key the expiry time is set to the first timestamp of
541 -- The heap must have at most one entry for any given key. The heap is
542 -- processed periodically to remove the expired entries. We pick up an
543 -- expired entry from the top of the heap and if the session has
544 -- expired based on the expiry time in the Map entry then we remove the
545 -- session from the Map and yield its fold output. Otherwise, we
546 -- reinsert the entry into the heap based on the current expiry in the
549 -- If an entry is removed from the Map and not removed from the heap
550 -- and in the meantime it is inserted again in the Map (using the same
551 -- key) then how do we avoid inserting multiple entries in the heap?
552 -- For this reason we maintain the invariant that the Map entry is
553 -- removed only when the heap entry is removed. Even if the fold has
554 -- finished we still keep a dummy Map entry (ZombieSession) until the
555 -- heap entry is removed. That way if we have a Map entry we do not
556 -- insert a heap entry because we know it is already there.
557 -- XXX The ZombieSession mechanism does not work as expected as we
558 -- ignore ZombieSession when inserting a new entry. Anyway, we can
559 -- remove this mechanism as at most only two heap entries may be
560 -- created and they will be ultimately cleaned up.
562 -- Heap processing needs the map and map processing needs the heap,
563 -- therefore we cannot separate the two for modularity unless we have a
564 -- way to achieve mutual recursion.
566 let curTime = max sessionEventTime timestamp
567 mOld = IsMap.mapLookup key sessionKeyValueMap
569 -- deleting a key from the heap is expensive, so we never
570 -- delete a key from heap, we just purge it from the Map and it
571 -- gets purged from the heap on timeout. We just need an extra
572 -- lookup in the Map when the key is purged from the heap, that
573 -- should not be expensive.
575 let (mp, cnt) = case mOld of
576 Just (LiveSession _ _) ->
578 key ZombieSession sessionKeyValueMap
581 _ -> (sessionKeyValueMap, sessionCount)
583 { sessionCurTime = curTime
584 , sessionEventTime = curTime
586 , sessionKeyValueMap = mp
587 , sessionOutputStream = Stream.fromPure (key, fb)
590 let expiry = addToAbsTime timestamp timeoutMs
591 (hp1, mp1, out1, cnt1) <- do
592 let vars = (sessionTimerHeap, sessionKeyValueMap,
593 Stream.nil, sessionCount)
595 -- inserting new entry
597 -- Eject a session from heap and map if needed
598 eject <- ejectPred sessionCount
599 (hp, mp, out, cnt) <-
601 then ejectOne reset extract vars
604 -- Insert the new session in heap
605 let hp' = H.insert (Entry expiry key) hp
606 in return (hp', mp, out, cnt + 1)
607 -- updating old entry
608 Just _ -> return vars
610 let acc = LiveSession expiry fs1
611 mp2 = IsMap.mapInsert key acc mp1
612 return $ SessionState
613 { sessionCurTime = curTime
614 , sessionEventTime = curTime
615 , sessionCount = cnt1
616 , sessionTimerHeap = hp1
617 , sessionKeyValueMap = mp2
618 , sessionOutputStream = out1
622 Just (LiveSession _ acc) -> return $ Fold.Partial acc
626 error $ "classifySessionsBy: "
627 ++ "The supplied fold must consume at least one input"
628 Fold.Partial fs -> do
631 Fold.Done fb -> done fb
632 Fold.Partial fs1 -> partial fs1
634 -- Got a timer tick event
635 sstep sessionState@SessionState{..} Nothing =
636 let curTime = addToAbsTime sessionCurTime tickMs
637 in ejectExpired reset ejectPred extract sessionState curTime
639-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
640-- classifies an input event @stream@ consisting of @(timestamp, (key,
641-- value))@ into sessions based on the @key@, folding all the values
642-- corresponding to the same key into a session using the supplied @fold@.
644-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
645-- session key and the folded value is emitted in the output stream. The
646-- timeout is measured from the first event in the session. If the @keepalive@
647-- option is set to 'True' the timeout is reset to 0 whenever an event is
650-- The @timestamp@ in the input stream is an absolute time from some epoch,
651-- characterizing the time when the input event was generated. The notion of
652-- current time is maintained by a monotonic event time clock using the
653-- timestamps seen in the input stream. The latest timestamp seen till now is
654-- used as the base for the current time. When no new events are seen, a timer
655-- is started with a clock resolution of @tick@ seconds. This timer is used to
656-- detect session timeouts in the absence of new events.
658-- To ensure an upper bound on the memory used the number of sessions can be
659-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
660-- oldest session is ejected before inserting a new session.
662-- When the stream ends any buffered sessions are ejected immediately.
664-- If a session key is received even after a session has finished, another
665-- session is created for that key.
668-- Stream.fold (Fold.drainBy print)
669-- $ Concur.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
670-- $ Stream.timestamped
672-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
679{-# INLINE classifySessionsBy #-}
681 :: (MonadAsync m, Ord k)
682 => Double -- ^ timer tick in seconds
683 -> Bool -- ^ reset the timer when an event is received
684 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
685 -> Double -- ^ session timeout in seconds
686 -> Fold m a b -- ^ Fold to be applied to session data
687 -> Stream m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
688 -> Stream m (k, b) -- ^ session key, fold result
689classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k))
691-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
692-- option set to 'True'.
695-- classifyKeepAliveSessions = classifySessionsBy 1 True
700{-# INLINE classifyKeepAliveSessions #-}
701classifyKeepAliveSessions ::
702 (MonadAsync m, Ord k)
703 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
704 -> Double -- ^ session inactive timeout
705 -> Fold m a b -- ^ Fold to be applied to session payload data
706 -> Stream m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
708classifyKeepAliveSessions = classifySessionsBy 1 True
710------------------------------------------------------------------------------
711-- Keyed tumbling windows
712------------------------------------------------------------------------------
714-- Tumbling windows is a special case of sliding windows where the window slide
715-- is the same as the window size. Or it can be a special case of session
716-- windows where the reset flag is set to False.
718-- XXX instead of using the early termination flag in the stream, we can use an
719-- early terminating fold instead.
722-- | Split the stream into fixed size chunks of specified size. Within each
723-- such chunk fold the elements in buckets identified by the keys. A particular
724-- bucket fold can be terminated early if a closing flag is encountered in an
725-- element for that key.
728{-# INLINABLE classifyChunksOf #-}
730 :: (IsStream t, MonadAsync m, Ord k)
731 => Int -- ^ window size
732 -> Fold m a b -- ^ Fold to be applied to window events
733 -> t m (k, a, Bool) -- ^ window key, data, close event
735classifyChunksOf wsize = classifyChunksBy wsize False
738-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
739-- option set to 'False'.
742-- classifySessionsOf = classifySessionsBy 1 False
747{-# INLINE classifySessionsOf #-}
749 (MonadAsync m, Ord k)
750 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
751 -> Double -- ^ time window size
752 -> Fold m a b -- ^ Fold to be applied to session data
753 -> Stream m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
755classifySessionsOf = classifySessionsBy 1 False
757------------------------------------------------------------------------------
759------------------------------------------------------------------------------
761-- | Continuously evaluate the input stream and sample the last event in time
762-- window of @n@ seconds.
764-- This is also known as @throttle@ in some libraries.
767-- sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
772{-# INLINE sampleIntervalEnd #-}
773sampleIntervalEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
774sampleIntervalEnd n = Stream.catMaybes . intervalsOf n Fold.last
776-- | Like 'sampleInterval' but samples at the beginning of the time window.
779-- sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
784{-# INLINE sampleIntervalStart #-}
785sampleIntervalStart :: MonadAsync m => Double -> Stream m a -> Stream m a
786sampleIntervalStart n = Stream.catMaybes . intervalsOf n Fold.one
792 | BurstDoneNext !x !t !x
794{-# INLINE sampleBurst #-}
795sampleBurst :: MonadAsync m => Bool -> Double -> Stream m a -> Stream m a
796sampleBurst sampleAtEnd gap xs =
797 -- XXX Ideally we should schedule a timer event exactly after gap time,
798 -- but the tick stream should work well as long as the timer
799 -- granularity is small enough compared to the gap.
800 Stream.mapMaybe extract
801 $ Stream.scan (Fold.foldl' step BurstNone)
803 $ interjectSuffix 0.01 (return Nothing) (fmap Just xs)
807 gap1 = toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
810 step BurstNone (t1, Just x1) = BurstWait t1 x1
811 step BurstNone _ = BurstNone
813 step (BurstDone _) (t1, Just x1) = BurstWait t1 x1
814 step (BurstDone _) _ = BurstNone
816 step old@(BurstWait t0 x0) (t1, Nothing)
817 | t1 - t0 >= gap1 = BurstDone x0
819 -- This can happen due to scheduling delays, if we received back to
820 -- back events spaced by more than the timeout without an
821 -- intervening timeout event then we emit the old event instead of
822 -- replacing it by the new.
823 step (BurstWait t0 x0) (t1, Just x1)
824 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
825 | sampleAtEnd = BurstWait t1 x1
826 | otherwise = BurstWait t1 x0
828 step (BurstDoneNext _ t0 x0) (t1, Nothing)
829 | t1 - t0 >= gap1 = BurstDone x0
830 | otherwise = BurstWait t0 x0
831 step (BurstDoneNext _ t0 x0) (t1, Just x1)
832 | t1 - t0 >= gap1 = BurstDoneNext x0 t1 x1
833 | sampleAtEnd = BurstWait t1 x1
834 | otherwise = BurstWait t1 x0
836 {-# INLINE extract #-}
837 extract (BurstDoneNext x _ _) = Just x
838 extract (BurstDone x) = Just x
841-- | Sample one event at the end of each burst of events. A burst is a group
842-- of events close together in time, it ends when an event is spaced by more
843-- than the specified time interval (in seconds) from the previous event.
845-- This is known as @debounce@ in some libraries.
847-- The clock granularity is 10 ms.
851{-# INLINE sampleBurstEnd #-}
852sampleBurstEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
853sampleBurstEnd = sampleBurst True
855-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
856-- instead of at the end of it.
860{-# INLINE sampleBurstStart #-}
861sampleBurstStart :: MonadAsync m => Double -> Stream m a -> Stream m a
862sampleBurstStart = sampleBurst False
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