@@ -1,5 +1,7 @@
11{-# LANGUAGE CPP #-}
22{-# LANGUAGE RankNTypes #-}
3+{-# LANGUAGE RecordWildCards #-}
4+{-# LANGUAGE KindSignatures #-}
35{-# LANGUAGE FlexibleContexts #-}
46
57#if __GLASGOW_HASKELL__ >= 800
@@ -449,6 +451,9 @@ import Control.Monad.State.Strict (StateT)
449451import Control.Monad.Trans (MonadTrans(..))
450452import Control.Monad.Trans.Control (MonadBaseControl)
451453import Data.Functor.Identity (Identity (..))
454+#if __GLASGOW_HASKELL__ >= 800
455+import Data.Kind (Type)
456+#endif
452457import Data.Heap (Entry(..))
453458import Data.Maybe (isJust, fromJust, isNothing)
454459import Foreign.Storable (Storable)
@@ -480,7 +485,7 @@ import Streamly.Internal.Data.Stream.Serial (SerialT, WSerialT)
480485import Streamly.Internal.Data.Stream.Zip (ZipSerialM)
481486import Streamly.Internal.Data.Pipe.Types (Pipe (..))
482487import Streamly.Internal.Data.Time.Units
483− (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
488+ (AbsTime, MilliSecond64(..), addToAbsTime, toRelTime,
484489 toAbsTime, TimeUnit64)
485490import Streamly.Internal.Mutable.Prim.Var (Prim, Var)
486491
@@ -3967,49 +3972,83 @@ classifyKeepAliveChunks
39673972classifyKeepAliveChunks spanout = classifyChunksBy spanout True
39683973-}
39693974
3970−-- | @classifySessionsBy tick timeout reset f stream@ groups together all input
3971−-- stream elements that belong to the same session. @timeout@ is the maximum
3972−-- lifetime of a session in seconds. All elements belonging to a session are
3973−-- purged after this duration. If "reset" is 'Ture' then the timeout is reset
3974−-- after every event received in the session. Session duration is measured
3975−-- using the timestamp of the first element seen for that session. To detect
3976−-- session timeouts, a monotonic event time clock is maintained using the
3977−-- timestamps seen in the inputs and a timer with a tick duration specified by
3978−-- @tick@.
3979−--
3980−-- @session key@ is a key that uniquely identifies the session for the given
3981−-- element, @timestamp@ characterizes the time when the input element was
3982−-- generated, this is an absolute time measured from some @Epoch@. @session
3983−-- close@ is a boolean indicating whether this element marks the closing of the
3984−-- session. When an input element with @session close@ set to @True@ is seen
3985−-- the session is purged immediately.
3986−--
3987−-- All the input elements belonging to a session are collected using the fold
3988−-- @f@. The session key and the fold result are emitted in the output stream
3989−-- when the session is purged either via the session close event or via the
3990−-- session lifetime timeout.
3975+#if __GLASGOW_HASKELL__ < 800
3976+#define Type *
3977+#endif
3978+
3979+data SessionState t m k a b = SessionState
3980+ { sessionCurTime :: !AbsTime -- ^ time since last event
3981+ , sessionEventTime :: !AbsTime -- ^ time as per last event
3982+ , sessionCount :: !Int -- ^ total number sessions in progress
3983+ , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts
3984+ , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys
3985+ , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
3986+ }
3987+
3988+#undef Type
3989+
3990+-- | @classifySessionsBy tick timeout reset f stream@ groups timed events in an
3991+-- input event stream into sessions based on a session key. Each element in the
3992+-- stream is an event consisting of a 4-tuple @(session key, sesssion data,
3993+-- isClosingEvent, timestamp)@. @session key@ is a key that uniquely
3994+-- identifies a session. All the events belonging to a session are folded using
3995+-- the fold @f@ until an event with @isClosingEvent@ flag set arrives or a
3996+-- timeout has occurred. The session key and the result of the fold are
3997+-- emitted in the output stream when the session is purged.
3998+--
3999+-- When @reset@ is 'False', @timeout@ is the maximum lifetime of a session in
4000+-- seconds, measured from the @timestamp@ of the first event in that session.
4001+-- When "reset" is 'True' then the timeout is reset after every event received
4002+-- in the session, in other words timeout is measured from the timestamp of the
4003+-- last event in the session.
4004+--
4005+-- @timestamp@ in an event characterizes the time when the input event was
4006+-- generated, this is an absolute time measured from some @Epoch@. The notion
4007+-- of current time is maintained by a monotonic event time clock using the
4008+-- timestamps seen in the input stream. The latest timestamp seen till now is
4009+-- used as the base for the current time. When no new events are seen, a timer
4010+-- is started with a tick duration specified by @tick@. This timer is used to
4011+-- detect session timeouts in the absence of new events.
4012+--
4013+-- /Internal/
39914014--
3992−-- @since 0.7.0
39934015{-# INLINABLE classifySessionsBy #-}
39944016classifySessionsBy
39954017 :: (IsStream t, MonadAsync m, Ord k)
39964018 => Double -- ^ timer tick in seconds
3997− -> Double -- ^ session timeout
4019+ -> Double -- ^ session timeout in seconds
39984020 -> Bool -- ^ reset the timeout when an event is received
39994021 -> Fold m a b -- ^ Fold to be applied to session events
4000− -> t m (k, a, Bool, AbsTime) -- ^ session key, timestamp, close event, data
4001− -> t m (k, b)
4022+ -> t m (k, a, Bool, AbsTime) -- ^ session key, data, timestamp, close event
4023+ -> t m (k, b) -- ^ session key, fold result
40024024classifySessionsBy tick timeout reset (Fold step initial extract) str =
4003− concatMap (\(Tuple4' _ _ _ s) -> s) $ scanlM' sstep szero stream
4025+ concatMap (\session -> sessionOutputStream session)
4026+ $ scanlM' sstep szero stream
40044027
40054028 where
40064029
40074030 timeoutMs = toRelTime (round (timeout * 1000) :: MilliSecond64)
40084031 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
4009− szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty K.nil
4010−
4032+ szero = SessionState
4033+ { sessionCurTime = toAbsTime (0 :: MilliSecond64)
4034+ , sessionEventTime = toAbsTime (0 :: MilliSecond64)
4035+ , sessionCount = 0
4036+ , sessionTimerHeap = H.empty
4037+ , sessionKeyValueMap = Map.empty
4038+ , sessionOutputStream = K.nil
4039+ }
4040+
4041+ -- XXX If there are too many sessions in progress we may want to put a
4042+ -- limit on the number of sessions because of memory consumption limits. We
4043+ -- can use two different strategies in this case:
4044+ --
4045+ -- 1) Eject old sessions or sessions beyond a certain/lower timeout
4046+ -- threshold even before timeout, effectively reduce the timeout.
4047+ -- 2) Drop creation of new sessions but keep accepting new events for the
4048+ -- old ones.
4049+ --
40114050 -- Got a new stream input element
4012− sstep (Tuple4' evTime hp mp _) (Just (key, a, closing, ts)) =
4051+ sstep (session@SessionState{..}) (Just (key, a, closing, ts)) =
40134052 -- XXX we should use a heap in pinned memory to scale it to a large
40144053 -- size
40154054 --
@@ -4027,66 +4066,90 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =
40274066 -- XXX if the key is an Int, we can also use an IntMap for slightly
40284067 -- better performance.
40294068 --
4030− let accumulate v = do
4069+ let curTime = max sessionEventTime ts
4070+ accumulate v = do
40314071 Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
40324072 new <- step old a
40334073 return $ Tuple' ts new
40344074 in if closing
40354075 then do
4036− let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
4076+ let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key
4077+ sessionKeyValueMap
40374078 Tuple' _ acc <- accumulate r
40384079 res <- extract acc
4039− return $ Tuple4' evTime hp mp' (yield (key, res))
4080+ return $ session
4081+ { sessionCurTime = curTime
4082+ , sessionEventTime = curTime
4083+ , sessionCount = sessionCount - 1
4084+ , sessionKeyValueMap = mp'
4085+ , sessionOutputStream = yield (key, res)
4086+ }
40404087 else do
4041− let r = Map.lookup key mp
4088+ let r = Map.lookup key sessionKeyValueMap
40424089 acc <- accumulate r
4043− let mp' = Map.insert key acc mp
4044− let hp' =
4090+ let mp' = Map.insert key acc sessionKeyValueMap
4091+ let (hp', count) =
40454092 case r of
40464093 Nothing ->
40474094 let expiry = addToAbsTime ts timeoutMs
4048− in H.insert (Entry expiry key) hp
4049− Just _ -> hp
4095+ in (H.insert (Entry expiry key)
4096+ sessionTimerHeap, sessionCount + 1)
4097+ Just _ -> (sessionTimerHeap, sessionCount)
40504098 -- Event time is maintained as monotonically increasing
40514099 -- time. If we have lagged behind any of the timestamps
40524100 -- seen then we increase it to match the latest time seen
40534101 -- in the timestamps. We also increase it on timer ticks.
4054− return $ Tuple4' (max evTime ts) hp' mp' K.nil
4102+ return $ SessionState
4103+ { sessionCurTime = curTime
4104+ , sessionEventTime = curTime
4105+ , sessionCount = count
4106+ , sessionTimerHeap = hp'
4107+ , sessionKeyValueMap = mp'
4108+ , sessionOutputStream = K.nil
4109+ }
40554110
40564111 -- Got a timer tick event
40574112 -- XXX can we yield the entries without accumulating them?
4058− sstep (Tuple4' evTime heap sessions _) Nothing = do
4059− (hp', mp', out) <- go heap sessions K.nil
4060− return $ Tuple4' curTime hp' mp' out
4113+ sstep (session@SessionState{..}) Nothing = do
4114+ (hp', mp', out, count) <- go sessionTimerHeap sessionKeyValueMap K.nil 0
4115+ return $ session
4116+ { sessionCurTime = curTime
4117+ , sessionCount = sessionCount - count
4118+ , sessionTimerHeap = hp'
4119+ , sessionKeyValueMap = mp'
4120+ , sessionOutputStream = out
4121+ }
40614122
40624123 where
40634124
4064− curTime = addToAbsTime evTime tickMs
4065− go hp mp out = do
4125+ curTime = addToAbsTime sessionCurTime tickMs
4126+ go hp mp out cnt = do
40664127 let hres = H.uncons hp
40674128 case hres of
4068− Just (Entry ts key, hp') -> do
4069− let duration = diffAbsTime curTime ts
4070− if duration >= timeoutMs
4129+ Just (Entry expiry key, hp') -> do
4130+ if curTime >= expiry
40714131 then do
40724132 let (r, mp') = Map.updateLookupWithKey
40734133 (\_ _ -> Nothing) key mp
40744134 case r of
4075− Nothing -> go hp' mp' out
4135+ Nothing -> go hp' mp' out cnt
40764136 Just (Tuple' latestTS acc) -> do
4077− let dur = diffAbsTime curTime latestTS
4078− if dur >= timeoutMs || not reset
4137+ let expiry' = addToAbsTime latestTS timeoutMs
4138+ if curTime >= expiry' || not reset
40794139 then do
40804140 sess <- extract acc
40814141 go hp' mp' ((key, sess) `K.cons` out)
4142+ (cnt + 1)
40824143 else
40834144 -- reset the session timeout
4084− let expiry = addToAbsTime latestTS timeoutMs
4085− hp'' = H.insert (Entry expiry key) hp'
4086− mp'' = Map.insert key (Tuple' latestTS acc) mp'
4087− in go hp'' mp'' out
4088− else return (hp, mp, out)
4089− Nothing -> return (hp, mp, out)
4145+ -- XXX purge anyway if the session size
4146+ -- goes beyond maximum session size.
4147+ let hp'' = H.insert (Entry expiry' key) hp'
4148+ mp'' = Map.insert key
4149+ (Tuple' latestTS acc) mp'
4150+ in go hp'' mp'' out cnt
4151+ else return (hp, mp, out, cnt)
4152+ Nothing -> return (hp, mp, out, cnt)
40904153
40914154 -- merge timer events in the stream
40924155 stream = Serial.map Just str `Par.parallel` repeatM timer
@@ -4098,7 +4161,12 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =
40984161-- received within the session window. The session times out and gets closed
40994162-- only if no event is received within the specified session window size.
41004163--
4101−-- @since 0.7.0
4164+-- @
4165+-- classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
4166+-- @
4167+--
4168+-- /Internal/
4169+--
41024170{-# INLINABLE classifyKeepAliveSessions #-}
41034171classifyKeepAliveSessions
41044172 :: (IsStream t, MonadAsync m, Ord k)
@@ -4147,7 +4215,12 @@ classifyChunksOf wsize = classifyChunksBy wsize False
41474215-- session window end, a monotonic event time clock is maintained synced with
41484216-- the timestamps with a clock resolution of 1 second.
41494217--
4150−-- @since 0.7.0
4218+-- @
4219+-- classifySessionsOf interval = classifySessionsBy 1 interval False
4220+-- @
4221+--
4222+-- /Internal/
4223+--
41514224{-# INLINABLE classifySessionsOf #-}
41524225classifySessionsOf
41534226 :: (IsStream t, MonadAsync m, Ord k)