@@ -1,3 +1,4 @@
1+{-# LANGUAGE BangPatterns #-}
12{-# LANGUAGE CPP #-}
23{-# LANGUAGE RankNTypes #-}
34{-# LANGUAGE RecordWildCards #-}
@@ -442,7 +443,7 @@ module Streamly.Internal.Prelude
442443where
443444
444445import Control.Concurrent (threadDelay)
445−import Control.Exception (Exception)
446+import Control.Exception (Exception, assert)
446447import Control.Monad (void)
447448import Control.Monad.Catch (MonadCatch)
448449import Control.Monad.IO.Class (MonadIO(..))
@@ -3987,18 +3988,18 @@ data SessionState t m k a b = SessionState
39873988
39883989#undef Type
39893990
3990−-- | @classifySessionsBy tick timeout idle 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 triple @(session key, sesssion data,
3993−-- timestamp)@. @session key@ is a key that uniquely identifies the session.
3994−-- All the events belonging to a session are folded using the fold @f@ until
3995−-- the fold returns a 'Left' result or a timeout has occurred. The session key
3996−-- and the result of the fold are emitted in the output stream when the session
3997−-- is purged.
3991+-- | @classifySessionsBy tick timeout idle pred f stream@ groups timestamped
3992+-- events in an input event stream into sessions based on a session key. Each
3993+-- element in the stream is an event consisting of a triple @(session key,
3994+-- sesssion data, timestamp)@. @session key@ is a key that uniquely identifies
3995+-- the session. All the events belonging to a session are folded using the
3996+-- fold @f@ until the fold returns a 'Left' result or a timeout has occurred.
3997+-- The session key and the result of the fold are emitted in the output stream
3998+-- when the session is purged.
39983999--
39994000-- When @idle@ is 'False', @timeout@ is the maximum lifetime of a session in
40004001-- seconds, measured from the @timestamp@ of the first event in that session.
4001−-- When "idle" is 'True' then the timeout is an idle timeout, it is reset after
4002+-- When @idle@ is 'True' then the timeout is an idle timeout, it is reset after
40024003-- every event received in the session.
40034004--
40044005-- @timestamp@ in an event characterizes the time when the input event was
@@ -4009,7 +4010,10 @@ data SessionState t m k a b = SessionState
40094010-- is started with a tick duration specified by @tick@. This timer is used to
40104011-- detect session timeouts in the absence of new events.
40114012--
4012−-- After @maxSessions@ limit is reached we start ejecting old sessions.
4013+-- The predicate @pred@ is invoked with the current session count, if it
4014+-- returns 'True' a session is ejected from the session cache before inserting
4015+-- a new session. This could be useful to alert or eject sessions when the
4016+-- number of sessions becomes too high.
40134017--
40144018-- /Internal/
40154019--
@@ -4021,11 +4025,11 @@ classifySessionsBy
40214025 => Double -- ^ timer tick in seconds
40224026 -> Double -- ^ session timeout in seconds
40234027 -> Bool -- ^ reset the timeout when an event is received
4024− -> Int -- ^ maximum number of sessions to cache
4028+ -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
40254029 -> Fold m a (Either b b) -- ^ Fold to be applied to session events
40264030 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
40274031 -> t m (k, b) -- ^ session key, fold result
4028−classifySessionsBy tick timeout reset maxSessions
4032+classifySessionsBy tick timeout reset ejectPred
40294033 (Fold step initial extract) str =
40304034 concatMap (\session -> sessionOutputStream session)
40314035 $ scanlM' sstep szero stream
@@ -4043,25 +4047,21 @@ classifySessionsBy tick timeout reset maxSessions
40434047 , sessionOutputStream = K.nil
40444048 }
40454049
4046− -- XXX If there are too many sessions in progress we may want to put a
4047− -- limit on the number of sessions because of memory consumption limits. We
4048− -- can use two different strategies in this case:
4050+ -- We can eject sessions based on the current session count to limit
4051+ -- memory consumption. There are two possible strategies:
40494052 --
40504053 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
40514054 -- threshold even before timeout, effectively reduce the timeout.
40524055 -- 2) Drop creation of new sessions but keep accepting new events for the
40534056 -- old ones.
40544057 --
4058+ -- We use the first strategy as of now.
4059+
40554060 -- Got a new stream input element
4056− sstep (session@SessionState{..}) (Just (key, a, ts)) = do
4061+ sstep (session@SessionState{..}) (Just (key, value, timestamp)) = do
40574062 -- XXX we should use a heap in pinned memory to scale it to a large
40584063 -- size
40594064 --
4060− -- deleting a key from the heap is expensive, so we never delete a
4061− -- key, we just purge it from the Map and it gets purged from the
4062− -- heap on timeout. We just need an extra lookup in the Map when
4063− -- the key is purged from the heap, that should not be expensive.
4064− --
40654065 -- To detect session inactivity we keep a timestamp of the latest event
40664066 -- in the Map along with the fold result. When we purge the session
40674067 -- from the heap we match the timestamp in the heap with the timestamp
@@ -4071,54 +4071,105 @@ classifySessionsBy tick timeout reset maxSessions
40714071 -- XXX if the key is an Int, we can also use an IntMap for slightly
40724072 -- better performance.
40734073 --
4074− let curTime = max sessionEventTime ts
4074+ let curTime = max sessionEventTime timestamp
40754075 accumulate v = do
4076− Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
4077− new <- step old a
4078− return $ Tuple' ts new
4076+ old <- case v of
4077+ Nothing -> initial
4078+ Just (Tuple' _ acc) -> return acc
4079+ new <- step old value
4080+ return $ Tuple' timestamp new
40794081 mOld = Map.lookup key sessionKeyValueMap
40804082
40814083 acc@(Tuple' _ fres) <- accumulate mOld
40824084 res <- extract fres
40834085 case res of
40844086 Left x -> do
4085− let mp' = case mOld of
4086− Nothing -> sessionKeyValueMap
4087− Just _ -> Map.delete key sessionKeyValueMap
4087+ -- deleting a key from the heap is expensive, so we never
4088+ -- delete a key from heap, we just purge it from the Map and it
4089+ -- gets purged from the heap on timeout. We just need an extra
4090+ -- lookup in the Map when the key is purged from the heap, that
4091+ -- should not be expensive.
4092+ --
4093+ let (mp, cnt) = case mOld of
4094+ Nothing -> (sessionKeyValueMap, sessionCount)
4095+ Just _ -> (Map.delete key sessionKeyValueMap
4096+ , sessionCount - 1)
40884097 return $ session
40894098 { sessionCurTime = curTime
40904099 , sessionEventTime = curTime
4091− , sessionCount = sessionCount - 1
4092− , sessionKeyValueMap = mp'
4100+ , sessionCount = cnt
4101+ , sessionKeyValueMap = mp
40934102 , sessionOutputStream = yield (key, x)
40944103 }
40954104 Right _ -> do
4096− let mp' = Map.insert key acc sessionKeyValueMap
4097− let (hp', count) =
4105+ (hp1, mp1, out1, cnt1) <- do
4106+ let vars = (sessionTimerHeap, sessionKeyValueMap,
4107+ K.nil, sessionCount)
40984108 case mOld of
4099− Nothing ->
4100− let expiry = addToAbsTime ts timeoutMs
4101− in (H.insert (Entry expiry key)
4102− sessionTimerHeap, sessionCount + 1)
4103− Just _ -> (sessionTimerHeap, sessionCount)
4104− -- Event time is maintained as monotonically increasing
4105− -- time. If we have lagged behind any of the timestamps
4106− -- seen then we increase it to match the latest time seen
4107− -- in the timestamps. We also increase it on timer ticks.
4109+ -- inserting new entry
4110+ Nothing -> do
4111+ -- Eject a session from heap and map is needed
4112+ eject <- ejectPred sessionCount
4113+ (hp, mp, out, cnt) <-
4114+ if eject
4115+ then ejectOne vars
4116+ else return vars
4117+
4118+ -- Insert the new session in heap
4119+ let expiry = addToAbsTime timestamp timeoutMs
4120+ hp' = H.insert (Entry expiry key) hp
4121+ in return $ (hp', mp, out, (cnt + 1))
4122+ -- updating old entry
4123+ Just _ -> return vars
4124+
4125+ let mp2 = Map.insert key acc mp1
41084126 return $ SessionState
41094127 { sessionCurTime = curTime
41104128 , sessionEventTime = curTime
4111− , sessionCount = count
4112− , sessionTimerHeap = hp'
4113− , sessionKeyValueMap = mp'
4114− , sessionOutputStream = K.nil
4129+ , sessionCount = cnt1
4130+ , sessionTimerHeap = hp1
4131+ , sessionKeyValueMap = mp2
4132+ , sessionOutputStream = out1
41154133 }
41164134
41174135 -- Got a timer tick event
4118− -- XXX can we yield the entries without accumulating them?
4119− sstep (session@SessionState{..}) Nothing = do
4136+ sstep (sessionState@SessionState{..}) Nothing =
4137+ let curTime = addToAbsTime sessionCurTime tickMs
4138+ in ejectExpired sessionState curTime
4139+
4140+ fromEither e =
4141+ case e of
4142+ Left x -> x
4143+ Right x -> x
4144+
4145+ -- delete from map and output the fold accumulator
4146+ ejectEntry hp mp out cnt acc key = do
4147+ sess <- extract acc
4148+ let out1 = (key, fromEither sess) `K.cons` out
4149+ let mp1 = Map.delete key mp
4150+ return (hp, mp1, out1, (cnt - 1))
4151+
4152+ ejectOne (hp, mp, out, !cnt) = do
4153+ let hres = H.uncons hp
4154+ case hres of
4155+ Just (Entry expiry key, hp1) -> do
4156+ case Map.lookup key mp of
4157+ Nothing -> ejectOne (hp1, mp, out, cnt)
4158+ Just (Tuple' latestTS acc) -> do
4159+ let expiry1 = addToAbsTime latestTS timeoutMs
4160+ if not reset || expiry1 <= expiry
4161+ then ejectEntry hp1 mp out cnt acc key
4162+ else
4163+ -- reset the session timeout and continue
4164+ let hp2 = H.insert (Entry expiry1 key) hp1
4165+ in ejectOne (hp2, mp, out, cnt)
4166+ Nothing -> do
4167+ assert (Map.null mp) (return ())
4168+ return (hp, mp, out, cnt)
4169+
4170+ ejectExpired (session@SessionState{..}) curTime = do
41204171 (hp', mp', out, count) <-
4121− go sessionTimerHeap sessionKeyValueMap K.nil sessionCount
4172+ ejectLoop sessionTimerHeap sessionKeyValueMap K.nil sessionCount
41224173 return $ session
41234174 { sessionCurTime = curTime
41244175 , sessionCount = count
@@ -4129,38 +4180,35 @@ classifySessionsBy tick timeout reset maxSessions
41294180
41304181 where
41314182
4132− curTime = addToAbsTime sessionCurTime tickMs
4133− expired etime cnt = curTime >= etime || cnt > maxSessions
4134− go hp mp out cnt = do
4183+ ejectLoop hp mp out !cnt = do
41354184 let hres = H.uncons hp
41364185 case hres of
4137− Just (Entry expiry key, hp') -> do
4138− if expired expiry
4186+ Just (Entry expiry key, hp1) -> do
4187+ (eject, force) <- do
4188+ if curTime >= expiry
4189+ then return (True, False)
4190+ else do
4191+ r <- ejectPred cnt
4192+ return (r, r)
4193+ if eject
41394194 then do
4140− let (r, mp') = Map.updateLookupWithKey
4141− (\_ _ -> Nothing) key mp
4142− case r of
4143− Nothing -> go hp' mp' out cnt
4195+ case Map.lookup key mp of
4196+ Nothing -> ejectLoop hp1 mp out cnt
41444197 Just (Tuple' latestTS acc) -> do
4145− let expiry' = addToAbsTime latestTS timeoutMs
4146− if expired expiry' || not reset
4198+ let expiry1 = addToAbsTime latestTS timeoutMs
4199+ if expiry1 <= curTime || not reset || force
41474200 then do
4148− sess <- extract acc
4149− let val = case sess of
4150− Left x -> x
4151− Right x -> x
4152− go hp' mp' ((key, val) `K.cons` out)
4153− (cnt - 1)
4201+ (hp2,mp1,out1,cnt1) <-
4202+ ejectEntry hp1 mp out cnt acc key
4203+ ejectLoop hp2 mp1 out1 cnt1
41544204 else
4155− -- reset the session timeout
4156− -- XXX purge anyway if the session size
4157− -- goes beyond maximum session size.
4158− let hp'' = H.insert (Entry expiry' key) hp'
4159− mp'' = Map.insert key
4160− (Tuple' latestTS acc) mp'
4161− in go hp'' mp'' out cnt
4205+ -- reset the session timeout and continue
4206+ let hp2 = H.insert (Entry expiry1 key) hp1
4207+ in ejectLoop hp2 mp out cnt
41624208 else return (hp, mp, out, cnt)
4163− Nothing -> return (hp, mp, out, cnt)
4209+ Nothing -> do
4210+ assert (Map.null mp) (return ())
4211+ return (hp, mp, out, cnt)
41644212
41654213 -- merge timer events in the stream
41664214 stream = Serial.map Just str `Par.parallel` repeatM timer
@@ -4172,8 +4220,11 @@ classifySessionsBy tick timeout reset maxSessions
41724220-- received within the session window. The session times out and gets closed
41734221-- only if no event is received within the specified session window size.
41744222--
4223+-- If the ejection predicate returns 'True', the session that was idle for
4224+-- the longest time is ejected before inserting a new session.
4225+--
41754226-- @
4176−-- classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
4227+-- classifyKeepAliveSessions timeout pred = classifySessionsBy 1 timeout True pred
41774228-- @
41784229--
41794230-- /Internal/
@@ -4182,10 +4233,12 @@ classifySessionsBy tick timeout reset maxSessions
41824233classifyKeepAliveSessions
41834234 :: (IsStream t, MonadAsync m, Ord k)
41844235 => Double -- ^ session inactive timeout
4236+ -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
41854237 -> Fold m a (Either b b) -- ^ Fold to be applied to session payload data
4186− -> t m (k, a, AbsTime) -- ^ session key, data, close flag, timestamp
4238+ -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
41874239 -> t m (k, b)
4188−classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
4240+classifyKeepAliveSessions timeout ejectPred =
4241+ classifySessionsBy 1 timeout True ejectPred
41894242
41904243------------------------------------------------------------------------------
41914244-- Keyed tumbling windows
@@ -4225,8 +4278,11 @@ classifyChunksOf wsize = classifyChunksBy wsize False
42254278-- session window end, a monotonic event time clock is maintained synced with
42264279-- the timestamps with a clock resolution of 1 second.
42274280--
4281+-- If the ejection predicate returns 'True', the session with the longest
4282+-- lifetime is ejected before inserting a new session.
4283+--
42284284-- @
4229−-- classifySessionsOf interval = classifySessionsBy 1 interval False
4285+-- classifySessionsOf interval pred = classifySessionsBy 1 interval False pred
42304286-- @
42314287--
42324288-- /Internal/
@@ -4235,10 +4291,12 @@ classifyChunksOf wsize = classifyChunksBy wsize False
42354291classifySessionsOf
42364292 :: (IsStream t, MonadAsync m, Ord k)
42374293 => Double -- ^ time window size
4238− -> Fold m a (Either b b) -- ^ Fold to be applied to window events
4239− -> t m (k, a, AbsTime) -- ^ window key, data, close flag, timestamp
4294+ -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
4295+ -> Fold m a (Either b b) -- ^ Fold to be applied to session events
4296+ -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
42404297 -> t m (k, b)
4241−classifySessionsOf interval = classifySessionsBy 1 interval False
4298+classifySessionsOf interval ejectPred =
4299+ classifySessionsBy 1 interval False ejectPred
42424300
42434301------------------------------------------------------------------------------
42444302-- Exceptions