@@ -0,0 +1,862 @@
1+-- |
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
7+-- Portability : GHC
8+--
9+
10+module Streamly.Internal.Data.Stream.Time
11+ (
12+ -- Primitives
13+ interjectSuffix
14+ , takeInterval
15+ , dropInterval
16+ , intervalsOf
17+ , chunksOfTimeout
18+
19+ -- Sessions
20+ , classifySessionsByGeneric
21+ , classifySessionsBy
22+ , classifySessionsOf
23+ , classifyKeepAliveSessions
24+
25+ -- Sampling
26+ , sampleIntervalEnd
27+ , sampleIntervalStart
28+ , sampleBurst
29+ , sampleBurstEnd
30+ , sampleBurstStart
31+ )
32+where
33+
34+import Control.Concurrent (threadDelay)
35+import Control.Exception (assert)
36+import Control.Monad.IO.Class (MonadIO(..))
37+import Data.Heap (Entry(..))
38+import Data.Kind (Type)
39+import Data.Map (Map)
40+import Data.Maybe (isNothing)
41+import Data.Proxy (Proxy(..))
42+import Streamly.Data.Fold (Fold)
43+import Streamly.Internal.Data.Fold.Type (Fold (..))
44+import Streamly.Internal.Data.IsMap (IsMap(..))
45+import Streamly.Internal.Data.Stream.Type (Stream)
46+import Streamly.Internal.Data.Time.Units
47+ ( AbsTime
48+ , MilliSecond64(..)
49+ , addToAbsTime
50+ , toAbsTime
51+ , toRelTime
52+ )
53+import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)
54+
55+import qualified Data.Heap as H
56+import qualified Streamly.Data.Fold as Fold
57+import qualified Streamly.Data.Stream as Stream
58+import qualified Streamly.Data.Unfold as Unfold
59+import qualified Streamly.Internal.Data.Fold as Fold (Step(..))
60+import qualified Streamly.Internal.Data.IsMap as IsMap
61+import qualified Streamly.Internal.Data.Stream as Stream
62+ ( catMaybes
63+ , scanlMAfter'
64+ , timeIndexed
65+ , timestamped
66+ )
67+
68+import Streamly.Internal.Data.Stream.Concurrent
69+
70+-- $setup
71+--
72+-- Imports for example snippets in this module.
73+--
74+-- >>> :m
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)
84+-- >>> :{
85+-- delay n = do
86+-- threadDelay (n * 1000000) -- sleep for n seconds
87+-- putStrLn (show n ++ " sec") -- print "n sec"
88+-- return n -- IO Int
89+-- :}
90+
91+--------------------------------------------------------------------------------
92+-- Combinators
93+--------------------------------------------------------------------------------
94+
95+-- | Intersperse a monadic action into the input stream after every @n@
96+-- seconds.
97+--
98+-- >>> Stream.fold Fold.drain $ Concur.interjectSuffix 1.05 (putChar ',') $ Stream.mapM (\x -> threadDelay 1000000 >> putChar x) $ Stream.fromList "hello"
99+-- h,e,l,l,o
100+--
101+-- /Pre-release/
102+{-# INLINE interjectSuffix #-}
103+interjectSuffix :: MonadAsync m => Double -> m a -> Stream m a -> Stream m a
104+interjectSuffix n f xs = parallelFst [xs, repeatM timed]
105+
106+ where
107+
108+ timed = liftIO (threadDelay (round $ n * 1000000)) >> f
109+ repeatM = Stream.sequence . Stream.repeat
110+
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
114+-- options:
115+--
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.
118+--
119+-- 2) Use COARSE clock to get time with lower resolution but more efficiently.
120+--
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.
124+
125+
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
130+-- stream stops.
131+--
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.
136+--
137+-- No element is yielded if the duration is zero. At least one element is
138+-- yielded if the duration is non-zero.
139+--
140+-- /Pre-release/
141+--
142+{-# INLINE takeInterval #-}
143+takeInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
144+takeInterval d =
145+ Stream.catMaybes
146+ . Stream.takeWhile isNothing
147+ . interjectSuffix d (return Nothing) . fmap Just
148+
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.
153+--
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.
158+--
159+-- All elements are yielded if the duration is zero.
160+--
161+-- /Pre-release/
162+--
163+{-# INLINE dropInterval #-}
164+dropInterval :: MonadAsync m => Double -> Stream m a -> Stream m a
165+dropInterval d =
166+ Stream.catMaybes
167+ . Stream.dropWhile isNothing
168+ . interjectSuffix d (return Nothing) . fmap Just
169+
170+-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
171+-- XXX add this example after fixing the serial stream rate control
172+--
173+-- | Group the input stream into windows of @n@ second each and then fold each
174+-- group using the provided fold function.
175+--
176+-- >>> Stream.toList $ Stream.take 5 $ Stream.intervalsOf 1 Fold.sum $ Stream.constRate 2 $ Stream.enumerateFrom 1
177+-- [...,...,...,...,...]
178+--
179+-- @since 0.7.0
180+{-# INLINE intervalsOf #-}
181+intervalsOf :: MonadAsync m => Double -> Fold m a b -> Stream m a -> Stream m b
182+intervalsOf n f xs =
183+ Stream.foldMany
184+ (Fold.takeEndBy isNothing (Fold.catMaybes f))
185+ (interjectSuffix n (return Nothing) (fmap Just xs))
186+
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.
189+--
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
193+-- is 100 ms.
194+--
195+-- >>> s = Stream.delayPost 0.3 $ Stream.fromList [1..1000]
196+-- >>> f = Stream.fold (Fold.drainBy print) $ Stream.chunksOfTimeout 5 1 Fold.toList s
197+--
198+-- /Pre-release/
199+{-# INLINE chunksOfTimeout #-}
200+chunksOfTimeout :: MonadAsync m
201+ => Int -> Double -> Fold m a b -> Stream m a -> Stream m b
202+chunksOfTimeout n timeout f =
203+ fmap snd
204+ . classifySessionsBy
205+ 0.1 False (const (return False)) timeout (Fold.take n f)
206+ . Stream.timestamped
207+ . fmap ((),)
208+
209+
210+------------------------------------------------------------------------------
211+-- Windowed classification
212+------------------------------------------------------------------------------
213+
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
216+-- or timestamps.
217+
218+------------------------------------------------------------------------------
219+-- Keyed Sliding Windows
220+------------------------------------------------------------------------------
221+
222+{-
223+{-# INLINABLE classifySlidingChunks #-}
224+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
230+ -> t m (k, b)
231+classifySlidingChunks wsize wslide (Fold step initial extract) str
232+ = undefined
233+
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 #-}
240+classifySlidingSessions
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
247+ -> t m (k, b)
248+classifySlidingSessions tick interval slide (Fold step initial extract) str
249+ = undefined
250+-}
251+
252+------------------------------------------------------------------------------
253+-- Sliding Window Buffers
254+------------------------------------------------------------------------------
255+
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.
262+--
263+-- XXX we should be able to implement sequence based splitting combinators
264+-- using this combinator.
265+
266+{-
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.
271+slidingChunkBuffer
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
276+ -> t m a
277+ -> t m b
278+slidingChunkBuffer = undefined
279+
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.
284+slidingSessionBuffer
285+ :: (IsStream t, Monad m, Ord a, Unboxed a)
286+ => Int -- window size
287+ -> Int -- tick size
288+ -> Fold m (RTree a, Array a) b
289+ -> t m a
290+ -> t m b
291+slidingSessionBuffer = undefined
292+-}
293+
294+------------------------------------------------------------------------------
295+-- Keyed Session Windows
296+------------------------------------------------------------------------------
297+
298+{-
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 #-}
302+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
308+ -> t m (k, b)
309+classifyChunksBy spanout reset (Fold step initial extract) str = undefined
310+
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.
314+--
315+{-# INLINABLE classifyKeepAliveChunks #-}
316+classifyKeepAliveChunks
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
321+ -> t m (k, b)
322+classifyKeepAliveChunks spanout = classifyChunksBy spanout True
323+-}
324+
325+data 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
334+ }
335+
336+data SessionEntry s = LiveSession !AbsTime !s | ZombieSession
337+
338+-- delete from map and output the fold accumulator
339+ejectEntry :: (Monad m, IsMap f) =>
340+ (acc -> m b)
341+ -> heap
342+ -> f entry
343+ -> Stream m (Key f, b)
344+ -> Int
345+ -> acc
346+ -> Key f
347+ -> m (heap, f entry, Stream m (Key f, b), Int)
348+ejectEntry extract hp mp out cnt acc key = do
349+ sess <- extract acc
350+ let out1 = (key, sess) `Stream.cons` out
351+ let mp1 = IsMap.mapDelete key mp
352+ return (hp, mp1, out1, cnt - 1)
353+
354+{-# NOINLINE flush #-}
355+flush :: (Monad m, IsMap f) =>
356+ (s -> m b)
357+ -> SessionState Stream m f (SessionEntry s) b
358+ -> m (SessionState Stream m f (SessionEntry s) b)
359+flush extract session@SessionState{..} = do
360+ (hp', mp', out, count) <-
361+ ejectAll
362+ ( sessionTimerHeap
363+ , sessionKeyValueMap
364+ , Stream.nil
365+ , sessionCount
366+ )
367+ return $ session
368+ { sessionCount = count
369+ , sessionTimerHeap = hp'
370+ , sessionKeyValueMap = mp'
371+ , sessionOutputStream = out
372+ }
373+
374+ where
375+
376+ ejectAll (hp, mp, out, !cnt) = do
377+ let hres = H.uncons hp
378+ case hres of
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
386+ ejectAll r
387+ Nothing -> do
388+ assert (IsMap.mapNull mp) (return ())
389+ return (hp, mp, out, cnt)
390+
391+{-# NOINLINE ejectOne #-}
392+ejectOne :: (IsMap f, Monad m) =>
393+ Bool
394+ -> (acc -> m b)
395+ -> ( H.Heap (Entry AbsTime (Key f))
396+ , f (SessionEntry acc)
397+ , Stream m (Key f, b)
398+ , Int
399+ )
400+ -> m ( H.Heap (Entry AbsTime (Key f))
401+ , f (SessionEntry acc)
402+ , Stream m (Key f, b), Int
403+ )
404+ejectOne reset extract = go
405+
406+ where
407+
408+ go (hp, mp, out, !cnt) = do
409+ let hres = H.uncons hp
410+ case hres of
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
419+ else
420+ -- reset the session timeout and continue
421+ let hp2 = H.insert (Entry expiry1 key) hp1
422+ in go (hp2, mp, out, cnt)
423+ Nothing -> do
424+ assert (IsMap.mapNull mp) (return ())
425+ return (hp, mp, out, cnt)
426+
427+{-# NOINLINE ejectExpired #-}
428+ejectExpired :: (IsMap f, Monad m) =>
429+ Bool
430+ -> (Int -> m Bool)
431+ -> (acc -> m b)
432+ -> SessionState Stream m f (SessionEntry acc) b
433+ -> AbsTime
434+ -> m (SessionState Stream m f (SessionEntry acc) b)
435+ejectExpired reset ejectPred extract session@SessionState{..} curTime = do
436+ (hp', mp', out, count) <-
437+ ejectLoop
438+ sessionTimerHeap sessionKeyValueMap Stream.nil sessionCount
439+ return $ session
440+ { sessionCurTime = curTime
441+ , sessionCount = count
442+ , sessionTimerHeap = hp'
443+ , sessionKeyValueMap = mp'
444+ , sessionOutputStream = out
445+ }
446+
447+ where
448+
449+ ejectLoop hp mp out !cnt = do
450+ let hres = H.uncons hp
451+ case hres of
452+ Just (Entry expiry key, hp1) -> do
453+ (eject, force) <-
454+ if curTime >= expiry
455+ then return (True, False)
456+ else do
457+ r <- ejectPred cnt
458+ return (r, r)
459+ if eject
460+ then
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
467+ then do
468+ (hp2,mp1,out1,cnt1) <-
469+ ejectEntry extract hp1 mp out cnt acc key
470+ ejectLoop hp2 mp1 out1 cnt1
471+ else
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)
476+ Nothing -> do
477+ assert (IsMap.mapNull mp) (return ())
478+ return (hp, mp, out, cnt)
479+
480+-- XXX Use mutable IORef in accumulator
481+{-# INLINE classifySessionsByGeneric #-}
482+classifySessionsByGeneric
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
491+ -- data)
492+ -> Stream m (Key f, b) -- ^ session key, fold result
493+classifySessionsByGeneric _ 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)
498+ $ fmap Just input
499+
500+ where
501+
502+ timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
503+ tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
504+ szero = SessionState
505+ { sessionCurTime = toAbsTime (0 :: MilliSecond64)
506+ , sessionEventTime = toAbsTime (0 :: MilliSecond64)
507+ , sessionCount = 0
508+ , sessionTimerHeap = H.empty
509+ , sessionKeyValueMap = IsMap.mapEmpty :: f s
510+ , sessionOutputStream = Stream.nil
511+ }
512+
513+ -- We can eject sessions based on the current session count to limit
514+ -- memory consumption. There are two possible strategies:
515+ --
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
519+ -- old ones.
520+ --
521+ -- We use the first strategy as of now.
522+
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.
526+ --
527+ -- XXX if the key is an Int, we can also use an IntMap for slightly
528+ -- better performance.
529+ --
530+ -- How it works:
531+ --
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
539+ -- the new session.
540+ --
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
547+ -- Map entry.
548+ --
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.
561+ --
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.
565+ --
566+ let curTime = max sessionEventTime timestamp
567+ mOld = IsMap.mapLookup key sessionKeyValueMap
568+ let done fb = do
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.
574+ --
575+ let (mp, cnt) = case mOld of
576+ Just (LiveSession _ _) ->
577+ ( IsMap.mapInsert
578+ key ZombieSession sessionKeyValueMap
579+ , sessionCount - 1
580+ )
581+ _ -> (sessionKeyValueMap, sessionCount)
582+ return $ session
583+ { sessionCurTime = curTime
584+ , sessionEventTime = curTime
585+ , sessionCount = cnt
586+ , sessionKeyValueMap = mp
587+ , sessionOutputStream = Stream.fromPure (key, fb)
588+ }
589+ partial fs1 = do
590+ let expiry = addToAbsTime timestamp timeoutMs
591+ (hp1, mp1, out1, cnt1) <- do
592+ let vars = (sessionTimerHeap, sessionKeyValueMap,
593+ Stream.nil, sessionCount)
594+ case mOld of
595+ -- inserting new entry
596+ Nothing -> do
597+ -- Eject a session from heap and map if needed
598+ eject <- ejectPred sessionCount
599+ (hp, mp, out, cnt) <-
600+ if eject
601+ then ejectOne reset extract vars
602+ else return vars
603+
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
609+
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
619+ }
620+ res0 <- do
621+ case mOld of
622+ Just (LiveSession _ acc) -> return $ Fold.Partial acc
623+ _ -> initial
624+ case res0 of
625+ Fold.Done _ ->
626+ error $ "classifySessionsBy: "
627+ ++ "The supplied fold must consume at least one input"
628+ Fold.Partial fs -> do
629+ res <- step fs value
630+ case res of
631+ Fold.Done fb -> done fb
632+ Fold.Partial fs1 -> partial fs1
633+
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
638+
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@.
643+--
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
648+-- received.
649+--
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.
657+--
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.
661+--
662+-- When the stream ends any buffered sessions are ejected immediately.
663+--
664+-- If a session key is received even after a session has finished, another
665+-- session is created for that key.
666+--
667+-- >>> :{
668+-- Stream.fold (Fold.drainBy print)
669+-- $ Concur.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
670+-- $ Stream.timestamped
671+-- $ Stream.delay 0.1
672+-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
673+-- :}
674+-- (1,"abc")
675+-- (2,"abc")
676+-- (3,"abc")
677+--
678+-- /Pre-release/
679+{-# INLINE classifySessionsBy #-}
680+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
689+classifySessionsBy = classifySessionsByGeneric (Proxy :: Proxy (Map k))
690+
691+-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
692+-- option set to 'True'.
693+--
694+-- @
695+-- classifyKeepAliveSessions = classifySessionsBy 1 True
696+-- @
697+--
698+-- /Pre-release/
699+--
700+{-# INLINE classifyKeepAliveSessions #-}
701+classifyKeepAliveSessions ::
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)
707+ -> Stream m (k, b)
708+classifyKeepAliveSessions = classifySessionsBy 1 True
709+
710+------------------------------------------------------------------------------
711+-- Keyed tumbling windows
712+------------------------------------------------------------------------------
713+
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.
717+
718+-- XXX instead of using the early termination flag in the stream, we can use an
719+-- early terminating fold instead.
720+
721+{-
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.
726+--
727+-- @since 0.7.0
728+{-# INLINABLE classifyChunksOf #-}
729+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
734+ -> t m (k, b)
735+classifyChunksOf wsize = classifyChunksBy wsize False
736+-}
737+
738+-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
739+-- option set to 'False'.
740+--
741+-- @
742+-- classifySessionsOf = classifySessionsBy 1 False
743+-- @
744+--
745+-- /Pre-release/
746+--
747+{-# INLINE classifySessionsOf #-}
748+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)
754+ -> Stream m (k, b)
755+classifySessionsOf = classifySessionsBy 1 False
756+
757+------------------------------------------------------------------------------
758+-- Sampling
759+------------------------------------------------------------------------------
760+
761+-- | Continuously evaluate the input stream and sample the last event in time
762+-- window of @n@ seconds.
763+--
764+-- This is also known as @throttle@ in some libraries.
765+--
766+-- @
767+-- sampleIntervalEnd n = Stream.catMaybes . Stream.intervalsOf n Fold.last
768+-- @
769+--
770+-- /Pre-release/
771+--
772+{-# INLINE sampleIntervalEnd #-}
773+sampleIntervalEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
774+sampleIntervalEnd n = Stream.catMaybes . intervalsOf n Fold.last
775+
776+-- | Like 'sampleInterval' but samples at the beginning of the time window.
777+--
778+-- @
779+-- sampleIntervalStart n = Stream.catMaybes . Stream.intervalsOf n Fold.one
780+-- @
781+--
782+-- /Pre-release/
783+--
784+{-# INLINE sampleIntervalStart #-}
785+sampleIntervalStart :: MonadAsync m => Double -> Stream m a -> Stream m a
786+sampleIntervalStart n = Stream.catMaybes . intervalsOf n Fold.one
787+
788+data BurstState t x =
789+ BurstNone
790+ | BurstWait !t !x
791+ | BurstDone !x
792+ | BurstDoneNext !x !t !x
793+
794+{-# INLINE sampleBurst #-}
795+sampleBurst :: MonadAsync m => Bool -> Double -> Stream m a -> Stream m a
796+sampleBurst 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)
802+ $ Stream.timeIndexed
803+ $ interjectSuffix 0.01 (return Nothing) (fmap Just xs)
804+
805+ where
806+
807+ gap1 = toRelTime64 (NanoSecond64 (round (gap * 10^(9::Int))))
808+
809+ {-# INLINE step #-}
810+ step BurstNone (t1, Just x1) = BurstWait t1 x1
811+ step BurstNone _ = BurstNone
812+
813+ step (BurstDone _) (t1, Just x1) = BurstWait t1 x1
814+ step (BurstDone _) _ = BurstNone
815+
816+ step old@(BurstWait t0 x0) (t1, Nothing)
817+ | t1 - t0 >= gap1 = BurstDone x0
818+ | otherwise = old
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
827+
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
835+
836+ {-# INLINE extract #-}
837+ extract (BurstDoneNext x _ _) = Just x
838+ extract (BurstDone x) = Just x
839+ extract _ = Nothing
840+
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.
844+--
845+-- This is known as @debounce@ in some libraries.
846+--
847+-- The clock granularity is 10 ms.
848+--
849+-- /Pre-release/
850+--
851+{-# INLINE sampleBurstEnd #-}
852+sampleBurstEnd :: MonadAsync m => Double -> Stream m a -> Stream m a
853+sampleBurstEnd = sampleBurst True
854+
855+-- | Like 'sampleBurstEnd' but samples the event at the beginning of the burst
856+-- instead of at the end of it.
857+--
858+-- /Pre-release/
859+--
860+{-# INLINE sampleBurstStart #-}
861+sampleBurstStart :: MonadAsync m => Double -> Stream m a -> Stream m a
862+sampleBurstStart = sampleBurst False