f6fd95a · streamly · git.iko.soy

1+{-# LANGUAGE BangPatterns #-}

1{-# LANGUAGE CPP #-}

2{-# LANGUAGE CPP #-}

2{-# LANGUAGE RankNTypes #-}

3{-# LANGUAGE RankNTypes #-}

3{-# LANGUAGE RecordWildCards #-}

4{-# LANGUAGE RecordWildCards #-}

442where

443where

443

444

444import Control.Concurrent (threadDelay)

445import Control.Concurrent (threadDelay)

445−import Control.Exception (Exception)

446+import Control.Exception (Exception, assert)

446import Control.Monad (void)

447import Control.Monad (void)

447import Control.Monad.Catch (MonadCatch)

448import Control.Monad.Catch (MonadCatch)

448import Control.Monad.IO.Class (MonadIO(..))

449import Control.Monad.IO.Class (MonadIO(..))

3987

3988

3988#undef Type

3989#undef Type

3989

3990

3990−-- | @classifySessionsBy tick timeout idle f stream@ groups timed events in an

3991+-- | @classifySessionsBy tick timeout idle pred f stream@ groups timestamped

3991−-- input event stream into sessions based on a session key. Each element in the

3992+-- events in an input event stream into sessions based on a session key. Each

3992−-- stream is an event consisting of a triple @(session key, sesssion data,

3993+-- element in the stream is an event consisting of a triple @(session key,

3993−-- timestamp)@. @session key@ is a key that uniquely identifies the session.

3994+-- sesssion data, timestamp)@. @session key@ is a key that uniquely identifies

3994−-- All the events belonging to a session are folded using the fold @f@ until

3995+-- the session. All the events belonging to a session are folded using the

3995−-- the fold returns a 'Left' result or a timeout has occurred. The session key

3996+-- fold @f@ until the fold returns a 'Left' result or a timeout has occurred.

3996−-- and the result of the fold are emitted in the output stream when the session

3997+-- The session key and the result of the fold are emitted in the output stream

3997−-- is purged.

3998+-- when the session is purged.

3998--

3999--

3999-- When @idle@ is 'False', @timeout@ is the maximum lifetime of a session in

4000-- When @idle@ 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-- 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

4002-- every event received in the session.

4003-- every event received in the session.

4003--

4004--

4004-- @timestamp@ in an event characterizes the time when the input event was

4005-- @timestamp@ in an event characterizes the time when the input event was

4009-- is started with a tick duration specified by @tick@. This timer is used to

4010-- is started with a tick duration specified by @tick@. This timer is used to

4010-- detect session timeouts in the absence of new events.

4011-- detect session timeouts in the absence of new events.

4011--

4012--

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.

4013--

4017--

4014-- /Internal/

4018-- /Internal/

4015--

4019--

4021 => Double -- ^ timer tick in seconds

4025 => Double -- ^ timer tick in seconds

4022 -> Double -- ^ session timeout in seconds

4026 -> Double -- ^ session timeout in seconds

4023 -> Bool -- ^ reset the timeout when an event is received

4027 -> 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

4025 -> Fold m a (Either b b) -- ^ Fold to be applied to session events

4029 -> Fold m a (Either b b) -- ^ Fold to be applied to session events

4026 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp

4030 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp

4027 -> t m (k, b) -- ^ session key, fold result

4031 -> t m (k, b) -- ^ session key, fold result

4028−classifySessionsBy tick timeout reset maxSessions

4032+classifySessionsBy tick timeout reset ejectPred

4029 (Fold step initial extract) str =

4033 (Fold step initial extract) str =

4030 concatMap (\session -> sessionOutputStream session)

4034 concatMap (\session -> sessionOutputStream session)

4031 $ scanlM' sstep szero stream

4035 $ scanlM' sstep szero stream

4043 , sessionOutputStream = K.nil

4047 , sessionOutputStream = K.nil

4044 }

4048 }

4045

4049

4046− -- XXX If there are too many sessions in progress we may want to put a

4050+ -- We can eject sessions based on the current session count to limit

4047− -- limit on the number of sessions because of memory consumption limits. We

4051+ -- memory consumption. There are two possible strategies:

4048− -- can use two different strategies in this case:

4049 --

4052 --

4050 -- 1) Eject old sessions or sessions beyond a certain/lower timeout

4053 -- 1) Eject old sessions or sessions beyond a certain/lower timeout

4051 -- threshold even before timeout, effectively reduce the timeout.

4054 -- threshold even before timeout, effectively reduce the timeout.

4052 -- 2) Drop creation of new sessions but keep accepting new events for the

4055 -- 2) Drop creation of new sessions but keep accepting new events for the

4053 -- old ones.

4056 -- old ones.

4054 --

4057 --

4058+ -- We use the first strategy as of now.

4059+

4055 -- Got a new stream input element

4060 -- Got a new stream input element

4056− sstep (session@SessionState{..}) (Just (key, a, ts)) = do

4061+ sstep (session@SessionState{..}) (Just (key, value, timestamp)) = do

4057 -- XXX we should use a heap in pinned memory to scale it to a large

4062 -- XXX we should use a heap in pinned memory to scale it to a large

4058 -- size

4063 -- size

4059 --

4064 --

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− --

4065 -- To detect session inactivity we keep a timestamp of the latest event

4066 -- in the Map along with the fold result. When we purge the session

4067 -- from the heap we match the timestamp in the heap with the timestamp

4071 -- XXX if the key is an Int, we can also use an IntMap for slightly

4072 -- better performance.

4073 --

4074− let curTime = max sessionEventTime ts

4074+ let curTime = max sessionEventTime timestamp

4075 accumulate v = do

4076− Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v

4076+ old <- case v of

4077+ Nothing -> initial

4078+ Just (Tuple' _ acc) -> return acc

4077− new <- step old a

4079+ new <- step old value

4078− return $ Tuple' ts new

4080+ return $ Tuple' timestamp new

4079 mOld = Map.lookup key sessionKeyValueMap

4081 mOld = Map.lookup key sessionKeyValueMap

4080

4082

4081 acc@(Tuple' _ fres) <- accumulate mOld

4083 acc@(Tuple' _ fres) <- accumulate mOld

4082 res <- extract fres

4084 res <- extract fres

4083 case res of

4085 case res of

4084 Left x -> do

4086 Left x -> do

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+ --

4085− let mp' = case mOld of

4093+ let (mp, cnt) = case mOld of

4086− Nothing -> sessionKeyValueMap

4094+ Nothing -> (sessionKeyValueMap, sessionCount)

4087− Just _ -> Map.delete key sessionKeyValueMap

4095+ Just _ -> (Map.delete key sessionKeyValueMap

4096+ , sessionCount - 1)

4088 return $ session

4097 return $ session

4089 { sessionCurTime = curTime

4098 { sessionCurTime = curTime

4090 , sessionEventTime = curTime

4099 , sessionEventTime = curTime

4091− , sessionCount = sessionCount - 1

4100+ , sessionCount = cnt

4092− , sessionKeyValueMap = mp'

4101+ , sessionKeyValueMap = mp

4093 , sessionOutputStream = yield (key, x)

4102 , sessionOutputStream = yield (key, x)

4094 }

4103 }

4095 Right _ -> do

4104 Right _ -> do

4105+ (hp1, mp1, out1, cnt1) <- do

4096− let mp' = Map.insert key acc sessionKeyValueMap

4106+ let vars = (sessionTimerHeap, sessionKeyValueMap,

4097− let (hp', count) =

4107+ K.nil, sessionCount)

4098 case mOld of

4108 case mOld of

4109+ -- inserting new entry

4099− Nothing ->

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

4100− let expiry = addToAbsTime ts timeoutMs

4119+ let expiry = addToAbsTime timestamp timeoutMs

4101− in (H.insert (Entry expiry key)

4120+ hp' = H.insert (Entry expiry key) hp

4102− sessionTimerHeap, sessionCount + 1)

4121+ in return $ (hp', mp, out, (cnt + 1))

4103− Just _ -> (sessionTimerHeap, sessionCount)

4122+ -- updating old entry

4104− -- Event time is maintained as monotonically increasing

4123+ Just _ -> return vars

4105− -- time. If we have lagged behind any of the timestamps

4124+

4106− -- seen then we increase it to match the latest time seen

4125+ let mp2 = Map.insert key acc mp1

4107− -- in the timestamps. We also increase it on timer ticks.

4108 return $ SessionState

4126 return $ SessionState

4109 { sessionCurTime = curTime

4127 { sessionCurTime = curTime

4110 , sessionEventTime = curTime

4128 , sessionEventTime = curTime

4111− , sessionCount = count

4129+ , sessionCount = cnt1

4112− , sessionTimerHeap = hp'

4130+ , sessionTimerHeap = hp1

4113− , sessionKeyValueMap = mp'

4131+ , sessionKeyValueMap = mp2

4114− , sessionOutputStream = K.nil

4132+ , sessionOutputStream = out1

4115 }

4133 }

4116

4134

4117 -- Got a timer tick event

4135 -- Got a timer tick event

4118− -- XXX can we yield the entries without accumulating them?

4136+ sstep (sessionState@SessionState{..}) Nothing =

4119− sstep (session@SessionState{..}) Nothing = do

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

4120 (hp', mp', out, count) <-

4171 (hp', mp', out, count) <-

4121− go sessionTimerHeap sessionKeyValueMap K.nil sessionCount

4172+ ejectLoop sessionTimerHeap sessionKeyValueMap K.nil sessionCount

4122 return $ session

4173 return $ session

4123 { sessionCurTime = curTime

4174 { sessionCurTime = curTime

4124 , sessionCount = count

4175 , sessionCount = count

4129

4180

4130 where

4181 where

4131

4182

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

4135 let hres = H.uncons hp

4184 let hres = H.uncons hp

4136 case hres of

4185 case hres of

4137− Just (Entry expiry key, hp') -> do

4186+ Just (Entry expiry key, hp1) -> do

4187+ (eject, force) <- do

4138− if expired expiry

4188+ if curTime >= expiry

4189+ then return (True, False)

4190+ else do

4191+ r <- ejectPred cnt

4192+ return (r, r)

4193+ if eject

4139 then do

4194 then do

4140− let (r, mp') = Map.updateLookupWithKey

4195+ case Map.lookup key mp of

4141− (\_ _ -> Nothing) key mp

4142− case r of

4143− Nothing -> go hp' mp' out cnt

4196+ Nothing -> ejectLoop hp1 mp out cnt

4144 Just (Tuple' latestTS acc) -> do

4197 Just (Tuple' latestTS acc) -> do

4145− let expiry' = addToAbsTime latestTS timeoutMs

4198+ let expiry1 = addToAbsTime latestTS timeoutMs

4146− if expired expiry' || not reset

4199+ if expiry1 <= curTime || not reset || force

4147 then do

4200 then do

4148− sess <- extract acc

4201+ (hp2,mp1,out1,cnt1) <-

4149− let val = case sess of

4202+ ejectEntry hp1 mp out cnt acc key

4150− Left x -> x

4203+ ejectLoop hp2 mp1 out1 cnt1

4151− Right x -> x

4152− go hp' mp' ((key, val) `K.cons` out)

4153− (cnt - 1)

4154 else

4204 else

4155− -- reset the session timeout

4205+ -- reset the session timeout and continue

4156− -- XXX purge anyway if the session size

4157− -- goes beyond maximum session size.

4158− let hp'' = H.insert (Entry expiry' key) hp'

4206+ let hp2 = H.insert (Entry expiry1 key) hp1

4159− mp'' = Map.insert key

4160− (Tuple' latestTS acc) mp'

4161− in go hp'' mp'' out cnt

4207+ in ejectLoop hp2 mp out cnt

4162 else return (hp, mp, out, cnt)

4208 else return (hp, mp, out, cnt)

4209+ Nothing -> do

4210+ assert (Map.null mp) (return ())

4163− Nothing -> return (hp, mp, out, cnt)

4211+ return (hp, mp, out, cnt)

4164

4212

4165 -- merge timer events in the stream

4213 -- merge timer events in the stream

4166 stream = Serial.map Just str `Par.parallel` repeatM timer

4214 stream = Serial.map Just str `Par.parallel` repeatM timer

4172-- received within the session window. The session times out and gets closed

4220-- received within the session window. The session times out and gets closed

4173-- only if no event is received within the specified session window size.

4221-- only if no event is received within the specified session window size.

4174--

4222--

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+--

4175-- @

4226-- @

4176−-- classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True

4227+-- classifyKeepAliveSessions timeout pred = classifySessionsBy 1 timeout True pred

4177-- @

4228-- @

4178--

4229--

4179-- /Internal/

4230-- /Internal/

4182classifyKeepAliveSessions

4233classifyKeepAliveSessions

4183 :: (IsStream t, MonadAsync m, Ord k)

4234 :: (IsStream t, MonadAsync m, Ord k)

4184 => Double -- ^ session inactive timeout

4235 => Double -- ^ session inactive timeout

4236+ -> (Int -> m Bool) -- ^ predicate to eject sessions on session count

4185 -> Fold m a (Either b b) -- ^ Fold to be applied to session payload data

4237 -> 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

4187 -> t m (k, b)

4239 -> t m (k, b)

4188−classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True

4240+classifyKeepAliveSessions timeout ejectPred =

4241+ classifySessionsBy 1 timeout True ejectPred

4189

4242

4190------------------------------------------------------------------------------

4243------------------------------------------------------------------------------

4191-- Keyed tumbling windows

4244-- Keyed tumbling windows

4225-- session window end, a monotonic event time clock is maintained synced with

4278-- session window end, a monotonic event time clock is maintained synced with

4226-- the timestamps with a clock resolution of 1 second.

4279-- the timestamps with a clock resolution of 1 second.

4227--

4280--

4281+-- If the ejection predicate returns 'True', the session with the longest

4282+-- lifetime is ejected before inserting a new session.

4283+--

4228-- @

4284-- @

4229−-- classifySessionsOf interval = classifySessionsBy 1 interval False

4285+-- classifySessionsOf interval pred = classifySessionsBy 1 interval False pred

4230-- @

4286-- @

4231--

4287--

4232-- /Internal/

4288-- /Internal/

4235classifySessionsOf

4291classifySessionsOf

4236 :: (IsStream t, MonadAsync m, Ord k)

4292 :: (IsStream t, MonadAsync m, Ord k)

4237 => Double -- ^ time window size

4293 => Double -- ^ time window size

4294+ -> (Int -> m Bool) -- ^ predicate to eject sessions on session count

4238− -> Fold m a (Either b b) -- ^ Fold to be applied to window events

4295+ -> Fold m a (Either b b) -- ^ Fold to be applied to session events

4239− -> t m (k, a, AbsTime) -- ^ window key, data, close flag, timestamp

4296+ -> t m (k, a, AbsTime) -- ^ session key, data, timestamp

4240 -> t m (k, b)

4297 -> t m (k, b)

4298+classifySessionsOf interval ejectPred =

4241−classifySessionsOf interval = classifySessionsBy 1 interval False

4299+ classifySessionsBy 1 interval False ejectPred

4242

4300

4243------------------------------------------------------------------------------

4301------------------------------------------------------------------------------

4244-- Exceptions

4302-- Exceptions