@@ -336,9 +336,10 @@ import Streamly.Fold.Types (Fold(..))
336import Streamly.Mem.Array (Array)
336import Streamly.Mem.Array (Array)
337import Streamly.Streams.Serial (SerialT)
337import Streamly.Streams.Serial (SerialT)
338import Streamly.Streams.StreamK (IsStream())
338import Streamly.Streams.StreamK (IsStream())
339−import Streamly.Time.Clock (Clock(..), getTime)
339+-- import Streamly.Time.Clock (Clock(..), getTime)
340import Streamly.Time.Units
340import Streamly.Time.Units
341− (AbsTime, MilliSecond64(..), diffAbsTime, fromRelTime)
341+ (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
343import Streamly.Strict
344import Streamly.Strict
@@ -2113,24 +2114,22 @@ lchunksOf n f1 f2 = undefined
2113-- | @sessionsByLifeTime lifetime tick k f stream@ groups together all input
2114-- | @sessionsByLifeTime lifetime tick k f stream@ groups together all input
2114-- stream elements that belong to the same session. @lifetime@ is the maximum
2115-- stream elements that belong to the same session. @lifetime@ is the maximum
2115-- lifetime of a session in seconds. All elements belonging to a session are
2116-- lifetime of a session in seconds. All elements belonging to a session are
2116−-- purged after this duration. The session duration is measured using the
2117+-- purged after this duration. Session duration is measured using the
2117−-- timestamp of the first message seen for that session. The timestamp is
2118+-- timestamp of the first element seen for that session. To detect session
2118−-- compared with the current system time, we assume that the system time is in
2119+-- timeouts, a monotonic event time clock is maintained using the timestamps
2119−-- sync with the timestamps in the input stream. @tick@ is the timer
2120+-- seen in the inputs and a timer with a tick duration specified by @tick@.
2120−-- granularity specified in seconds to check and purge the sessions that have
2123-- The function @k@ returns a 3-tuple @(session key, timestamp, session close)@
2122-- The function @k@ returns a 3-tuple @(session key, timestamp, session close)@
2124-- from an input element in the stream. @session key@ is a key that uniquely
2123-- from an input element in the stream. @session key@ is a key that uniquely
2125-- identifies the session for the given element, @timestamp@ characterizes the
2124-- identifies the session for the given element, @timestamp@ characterizes the
2126−-- time when the input element was generated, this must be a time measured from
2125+-- time when the input element was generated, this is an absolute time measured
2127−-- the Unix @Epoch@. @session close@ is a boolean indicating whether this
2126+-- from some @Epoch@. @session close@ is a boolean indicating whether this
2128−-- element marks the closing of the session. When a message with
2127+-- element marks the closing of the session. When an input element with
2129-- @session close@ set to @True@ is seen the session is purged immediately.
2128-- @session close@ set to @True@ is seen the session is purged immediately.
2131−-- All the messages belonging to a session are collected using the fold @f@.
2130+-- All the input elements belonging to a session are collected using the fold
2132−-- The fold result is emitted when the session is purged either via the session
2131+-- @f@. The fold result is emitted when the session is purged either via the
2133−-- close event or via the session liftime timeout.
2132+-- session close event or via the session liftime timeout.
2135-- @since 0.7.0
2134-- @since 0.7.0
2136{-# INLINABLE sessionsByLifeTime #-}
2135{-# INLINABLE sessionsByLifeTime #-}
@@ -2143,19 +2142,27 @@ sessionsByLifeTime
2143 -> t m a
2142 -> t m a
2144 -> t m b
2143 -> t m b
2145sessionsByLifeTime lifetime tick getKey (Fold step initial extract) str =
2144sessionsByLifeTime lifetime tick getKey (Fold step initial extract) str =
2146− S.concatMap (\(Tuple3' _ _ s) -> s) $ S.scanlM' sstep szero stream
2145+ S.concatMap (\(Tuple4' _ _ _ s) -> s) $ S.scanlM' sstep szero stream
2150 -- XXX what if the timestamps and the system time are too far off, we may
2149 -- XXX what if the timestamps and the system time are too far off, we may
2151 -- accumulate too many entries, have a protection in that case? If the
2150 -- accumulate too many entries, have a protection in that case? If the
2152 -- incoming logs seem too new or too old than the current system time then
2151 -- incoming logs seem too new or too old than the current system time then
2153− -- send an alert.
2152+ -- send an alert. This should be detected earlier in the processing
2153+ -- pipeline, this function should not worry about it.
2155+ -- XXX what if the timestamps are delayed, we would expire the sessions
2156+ -- without waiting for all the messages. We should wait at least from the
2157+ -- first timestamp + lifetime.
2154 -- XXX we have to use a heap in pinned memory to scale it to a large size
2159 -- XXX we have to use a heap in pinned memory to scale it to a large size
2160+ lifetimeMs = toRelTime (round (lifetime * 1000) :: MilliSecond64)
2161+ tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
2155− szero = Tuple3' H.empty Map.empty S.nil
2162+ szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty S.nil
2157− -- Got a new input element
2164+ -- Got a new stream input element
2158− sstep (Tuple3' hp mp _) (Right a) =
2165+ sstep (Tuple4' evTime hp mp _) (Just a) =
2159 -- deleting a key from the heap is expensive, so we never delete a
2166 -- deleting a key from the heap is expensive, so we never delete a
2160 -- key, we just purge it from the Map and it gets purged from the
2167 -- key, we just purge it from the Map and it gets purged from the
2161 -- heap on timeout. We just need an extra lookup in the Map when
2168 -- heap on timeout. We just need an extra lookup in the Map when
@@ -2163,48 +2170,54 @@ sessionsByLifeTime lifetime tick getKey (Fold step initial extract) str =
2163 let (key, ts, closed) = getKey a
2170 let (key, ts, closed) = getKey a
2164 accumulate v = maybe initial return v >>= \r -> step r a
2171 accumulate v = maybe initial return v >>= \r -> step r a
2165 in if closed
2172 in if closed
2166− then do
2173+ then do
2167− let (v, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
2174+ let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
2168− r <- accumulate v
2175+ acc <- accumulate r
2169− res <- extract r
2176+ res <- extract acc
2170− return $ Tuple3' hp mp' (S.yield res)
2177+ return $ Tuple4' evTime hp mp' (S.yield res)
2171− else do
2178+ else do
2172− let v = Map.lookup key mp
2179+ let r = Map.lookup key mp
2173− acc <- accumulate v
2180+ acc <- accumulate r
2174 let mp' = Map.insert key acc mp
2181 let mp' = Map.insert key acc mp
2182+ let expiry = addToAbsTime ts lifetimeMs
2175− let hp' = case v of
2184+ case r of
2176− Nothing -> H.insert (Entry ts key) hp
2185+ Nothing -> H.insert (Entry expiry key) hp
2177− Just _ -> hp
2186+ Just _ -> hp
2187+ -- Event time is maintained as monotonically increasing
2188+ -- time. If we have lagged behind any of the timestamps
2189+ -- seen then we increase it to match the latest time seen
2190+ -- in the timestamps. We also increase it on timer ticks.
2178− return $ Tuple3' hp' mp' S.nil
2191+ return $ Tuple4' (max evTime ts) hp' mp' S.nil
2180− -- Got timer tick event
2193+ -- Got a timer tick event
2181 -- XXX can we yield the entries without accumulating them?
2194 -- XXX can we yield the entries without accumulating them?
2182− sstep (Tuple3' heap mmp _) (Left t) = do
2195+ sstep (Tuple4' evTime heap sessions _) Nothing = do
2183− (hp', mp', out) <- go heap mmp S.nil
2196+ (hp', mp', out) <- go heap sessions S.nil
2184− return $ Tuple3' hp' mp' out
2197+ return $ Tuple4' curTime hp' mp' out
2201+ curTime = addToAbsTime evTime tickMs
2188 go hp mp out = do
2202 go hp mp out = do
2189− let r = H.uncons hp
2203+ let hres = H.uncons hp
2190− case r of
2204+ case hres of
2191− Just (Entry ts k, hp') -> do
2205+ Just (Entry ts key, hp') -> do
2192− let duration = diffAbsTime ts t
2206+ let duration = diffAbsTime curTime ts
2193− lt = round (lifetime * 1000) :: MilliSecond64
2194− if fromRelTime duration >= lt
2207+ if duration >= lifetimeMs
2196− let (v, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) k mp
2209+ let (r, mp') = Map.updateLookupWithKey
2210+ (\_ _ -> Nothing) key mp
2197− case v of
2211+ case r of
2198 Nothing -> go hp' mp out
2212 Nothing -> go hp' mp out
2199 Just x -> do
2213 Just x -> do
2200− res <- extract x
2214+ sess <- extract x
2201− go hp' mp' (res `S.cons` out)
2215+ go hp' mp' (sess `S.cons` out)
2202 else return (hp, mp, out)
2216 else return (hp, mp, out)
2203 Nothing -> return (hp, mp, out)
2217 Nothing -> return (hp, mp, out)
2205 -- merge timer events in the stream
2219 -- merge timer events in the stream
2206− stream = S.map (\x -> Right x) str `parallel` S.repeatM timer
2220+ stream = S.map Just str `parallel` S.repeatM timer
2207 timer = do
2221 timer = do
2208 liftIO $ threadDelay (round $ tick * 1000000)
2222 liftIO $ threadDelay (round $ tick * 1000000)
2209− now <- liftIO $ getTime Realtime
2210− return $ Left now
2223+ return Nothing