f1d30a1 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Use the fold result to identify session termination

f1d30a1Harendra Kumar7 years ago

src/Streamly/Internal/Prelude.hs+58−54

compare with master

unifiedsplit

@@ -3987,20 +3987,19 @@ data SessionState t m k a b = SessionState

3987

3988#undef Type

3989

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

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 4-tuple @(session key, sesssion data,

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

3993−-- isClosingEvent, timestamp)@. @session key@ is a key that uniquely

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

3994−-- identifies a session. All the events belonging to a session are folded using

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

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

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

3997−-- emitted in the output stream when the session is purged.

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

3997+-- is purged.

3998−--

3998+--

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

3999+-- 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−-- When "reset" is 'True' then the timeout is reset after every event received

4001+-- When "idle" is 'True' then the timeout is an idle timeout, it is reset after

4002−-- in the session, in other words timeout is measured from the timestamp of the

4003−-- last event in the session.

4002+-- every event received in the session.

4004--

4003--

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

4004-- @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

4005-- generated, this is an absolute time measured from some @Epoch@. The notion

@@ -4012,14 +4011,16 @@ data SessionState t m k a b = SessionState

4012--

4011--

4013-- /Internal/

4012-- /Internal/

4014--

4013--

4014+

4015+-- XXX Perhaps we should use an "Event a" type to represent timestamped data.

4015{-# INLINABLE classifySessionsBy #-}

4016{-# INLINABLE classifySessionsBy #-}

4016classifySessionsBy

4017classifySessionsBy

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

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

4018 => Double -- ^ timer tick in seconds

4019 => Double -- ^ timer tick in seconds

4019 -> Double -- ^ session timeout in seconds

4020 -> Double -- ^ session timeout in seconds

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

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

4021− -> Fold m a b -- ^ Fold to be applied to session events

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

4022− -> t m (k, a, Bool, AbsTime) -- ^ session key, data, timestamp, close event

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

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

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

4024classifySessionsBy tick timeout reset (Fold step initial extract) str =

4025classifySessionsBy tick timeout reset (Fold step initial extract) str =

4025 concatMap (\session -> sessionOutputStream session)

4026 concatMap (\session -> sessionOutputStream session)

@@ -4048,7 +4049,7 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =

4048 -- old ones.

4049 -- old ones.

4049 --

4050 --

4050 -- Got a new stream input element

4051 -- Got a new stream input element

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

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

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

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

4053 -- size

4054 -- size

4054 --

4055 --

@@ -4071,42 +4072,43 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =

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

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

4072 new <- step old a

4073 new <- step old a

4073 return $ Tuple' ts new

4074 return $ Tuple' ts new

4074− in if closing

4075+ mOld = Map.lookup key sessionKeyValueMap

4075− then do

4076+

4076− let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key

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

4078+ res <- extract fres

4079+ case res of

4080+ Left x -> do

4081+ let mp' = case mOld of

4077− sessionKeyValueMap

4082+ Nothing -> sessionKeyValueMap

4078− Tuple' _ acc <- accumulate r

4083+ Just _ -> Map.delete key sessionKeyValueMap

4079− res <- extract acc

4080 return $ session

4084 return $ session

4081 { sessionCurTime = curTime

4085 { sessionCurTime = curTime

4082 , sessionEventTime = curTime

4086 , sessionEventTime = curTime

4083 , sessionCount = sessionCount - 1

4087 , sessionCount = sessionCount - 1

4084 , sessionKeyValueMap = mp'

4088 , sessionKeyValueMap = mp'

4085− , sessionOutputStream = yield (key, res)

4089+ , sessionOutputStream = yield (key, x)

4090+ }

4091+ Right _ -> do

4092+ let mp' = Map.insert key acc sessionKeyValueMap

4093+ let (hp', count) =

4094+ case mOld of

4095+ Nothing ->

4096+ let expiry = addToAbsTime ts timeoutMs

4097+ in (H.insert (Entry expiry key)

4098+ sessionTimerHeap, sessionCount + 1)

4099+ Just _ -> (sessionTimerHeap, sessionCount)

4100+ -- Event time is maintained as monotonically increasing

4101+ -- time. If we have lagged behind any of the timestamps

4102+ -- seen then we increase it to match the latest time seen

4103+ -- in the timestamps. We also increase it on timer ticks.

4104+ return $ SessionState

4105+ { sessionCurTime = curTime

4106+ , sessionEventTime = curTime

4107+ , sessionCount = count

4108+ , sessionTimerHeap = hp'

4109+ , sessionKeyValueMap = mp'

4110+ , sessionOutputStream = K.nil

4086 }

4111 }

4087− else do

4088− let r = Map.lookup key sessionKeyValueMap

4089− acc <- accumulate r

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

4091− let (hp', count) =

4092− case r of

4093− Nothing ->

4094− let expiry = addToAbsTime ts timeoutMs

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

4096− sessionTimerHeap, sessionCount + 1)

4097− Just _ -> (sessionTimerHeap, sessionCount)

4098− -- Event time is maintained as monotonically increasing

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

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

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

4102− return $ SessionState

4103− { sessionCurTime = curTime

4104− , sessionEventTime = curTime

4105− , sessionCount = count

4106− , sessionTimerHeap = hp'

4107− , sessionKeyValueMap = mp'

4108− , sessionOutputStream = K.nil

4109− }

4110

4112

4111 -- Got a timer tick event

4113 -- Got a timer tick event

4112 -- XXX can we yield the entries without accumulating them?

4114 -- XXX can we yield the entries without accumulating them?

@@ -4138,7 +4140,10 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =

4138 if curTime >= expiry' || not reset

4140 if curTime >= expiry' || not reset

4139 then do

4141 then do

4140 sess <- extract acc

4142 sess <- extract acc

4143+ let val = case sess of

4144+ Left x -> x

4145+ Right x -> x

4141− go hp' mp' ((key, sess) `K.cons` out)

4146+ go hp' mp' ((key, val) `K.cons` out)

4142 (cnt + 1)

4147 (cnt + 1)

4143 else

4148 else

4144 -- reset the session timeout

4149 -- reset the session timeout

@@ -4171,8 +4176,8 @@ classifySessionsBy tick timeout reset (Fold step initial extract) str =

4171classifyKeepAliveSessions

4176classifyKeepAliveSessions

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

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

4173 => Double -- ^ session inactive timeout

4178 => Double -- ^ session inactive timeout

4174− -> Fold m a b -- ^ Fold to be applied to session payload data

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

4175− -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp

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

4176 -> t m (k, b)

4181 -> t m (k, b)

4177classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True

4182classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True

4178

4183

@@ -4205,10 +4210,9 @@ classifyChunksOf wsize = classifyChunksBy wsize False

4205-}

4210-}

4206

4211

4207-- | Split the stream into fixed size time windows of specified interval in

4212-- | Split the stream into fixed size time windows of specified interval in

4208−-- seconds. Within each such window, fold the elements in buckets identified by

4213+-- seconds. Within each such window, fold the elements in sessions identified

4209−-- the keys. A particular bucket fold can be terminated early if a closing flag

4210−-- is encountered in an element for that key. Once a fold is terminated the key

4211−-- and value for that bucket are emitted in the output stream.

4214+-- by the session keys. The fold result is emitted in the output stream if the

4215+-- fold returns a 'Left' result or if the time window ends.

4212--

4216--

4213-- Session @timestamp@ in the input stream is an absolute time from some epoch,

4217-- Session @timestamp@ in the input stream is an absolute time from some epoch,

4214-- characterizing the time when the input element was generated. To detect

4218-- characterizing the time when the input element was generated. To detect

@@ -4225,8 +4229,8 @@ classifyChunksOf wsize = classifyChunksBy wsize False

4225classifySessionsOf

4229classifySessionsOf

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

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

4227 => Double -- ^ time window size

4231 => Double -- ^ time window size

4228− -> Fold m a b -- ^ Fold to be applied to window events

4232+ -> Fold m a (Either b b) -- ^ Fold to be applied to window events

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

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

4230 -> t m (k, b)

4234 -> t m (k, b)

4231classifySessionsOf interval = classifySessionsBy 1 interval False

4235classifySessionsOf interval = classifySessionsBy 1 interval False

4232

4236