@@ -336,9 +336,10 @@ import Streamly.Fold.Types (Fold(..))
336336import Streamly.Mem.Array (Array)
337337import Streamly.Streams.Serial (SerialT)
338338import Streamly.Streams.StreamK (IsStream())
339−import Streamly.Time.Clock (Clock(..), getTime)
339+-- import Streamly.Time.Clock (Clock(..), getTime)
340340import Streamly.Time.Units
341− (AbsTime, MilliSecond64(..), diffAbsTime, fromRelTime)
341+ (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
342+ toAbsTime)
342343
343344import Streamly.Strict
344345
@@ -2113,24 +2114,22 @@ lchunksOf n f1 f2 = undefined
21132114-- | @sessionsByLifeTime lifetime tick k f stream@ groups together all input
21142115-- stream elements that belong to the same session. @lifetime@ is the maximum
21152116-- 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−-- timestamp of the first message seen for that session. The timestamp is
2118−-- compared with the current system time, we assume that the system time is in
2119−-- sync with the timestamps in the input stream. @tick@ is the timer
2120−-- granularity specified in seconds to check and purge the sessions that have
2121−-- timed out.
2117+-- purged after this duration. Session duration is measured using the
2118+-- timestamp of the first element seen for that session. To detect session
2119+-- timeouts, a monotonic event time clock is maintained using the timestamps
2120+-- seen in the inputs and a timer with a tick duration specified by @tick@.
21222121--
21232122-- The function @k@ returns a 3-tuple @(session key, timestamp, session close)@
21242123-- from an input element in the stream. @session key@ is a key that uniquely
21252124-- 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
2127−-- the Unix @Epoch@. @session close@ is a boolean indicating whether this
2128−-- element marks the closing of the session. When a message with
2125+-- time when the input element was generated, this is an absolute time measured
2126+-- from some @Epoch@. @session close@ is a boolean indicating whether this
2127+-- element marks the closing of the session. When an input element with
21292128-- @session close@ set to @True@ is seen the session is purged immediately.
21302129--
2131−-- All the messages belonging to a session are collected using the fold @f@.
2132−-- The fold result is emitted when the session is purged either via the session
2133−-- close event or via the session liftime timeout.
2130+-- All the input elements belonging to a session are collected using the fold
2131+-- @f@. The fold result is emitted when the session is purged either via the
2132+-- session close event or via the session liftime timeout.
21342133--
21352134-- @since 0.7.0
21362135{-# INLINABLE sessionsByLifeTime #-}
@@ -2143,19 +2142,27 @@ sessionsByLifeTime
21432142 -> t m a
21442143 -> t m b
21452144sessionsByLifeTime 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
21472146
21482147 where
21492148
21502149 -- XXX what if the timestamps and the system time are too far off, we may
21512150 -- accumulate too many entries, have a protection in that case? If the
21522151 -- 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.
2154+ --
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.
2158+ --
21542159 -- XXX we have to use a heap in pinned memory to scale it to a large size
2155− szero = Tuple3' H.empty Map.empty S.nil
2160+ lifetimeMs = toRelTime (round (lifetime * 1000) :: MilliSecond64)
2161+ tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
2162+ szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty S.nil
21562163
2157− -- Got a new input element
2158− sstep (Tuple3' hp mp _) (Right a) =
2164+ -- Got a new stream input element
2165+ sstep (Tuple4' evTime hp mp _) (Just a) =
21592166 -- deleting a key from the heap is expensive, so we never delete a
21602167 -- key, we just purge it from the Map and it gets purged from the
21612168 -- 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 =
21632170 let (key, ts, closed) = getKey a
21642171 accumulate v = maybe initial return v >>= \r -> step r a
21652172 in if closed
2166− then do
2167− let (v, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
2168− r <- accumulate v
2169− res <- extract r
2170− return $ Tuple3' hp mp' (S.yield res)
2171− else do
2172− let v = Map.lookup key mp
2173− acc <- accumulate v
2173+ then do
2174+ let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
2175+ acc <- accumulate r
2176+ res <- extract acc
2177+ return $ Tuple4' evTime hp mp' (S.yield res)
2178+ else do
2179+ let r = Map.lookup key mp
2180+ acc <- accumulate r
21742181 let mp' = Map.insert key acc mp
2175− let hp' = case v of
2176− Nothing -> H.insert (Entry ts key) hp
2177− Just _ -> hp
2178− return $ Tuple3' hp' mp' S.nil
2179−
2180− -- Got timer tick event
2182+ let expiry = addToAbsTime ts lifetimeMs
2183+ let hp' =
2184+ case r of
2185+ Nothing -> H.insert (Entry expiry key) 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.
2191+ return $ Tuple4' (max evTime ts) hp' mp' S.nil
2192+
2193+ -- Got a timer tick event
21812194 -- XXX can we yield the entries without accumulating them?
2182− sstep (Tuple3' heap mmp _) (Left t) = do
2183− (hp', mp', out) <- go heap mmp S.nil
2184− return $ Tuple3' hp' mp' out
2195+ sstep (Tuple4' evTime heap sessions _) Nothing = do
2196+ (hp', mp', out) <- go heap sessions S.nil
2197+ return $ Tuple4' curTime hp' mp' out
21852198
21862199 where
21872200
2201+ curTime = addToAbsTime evTime tickMs
21882202 go hp mp out = do
2189− let r = H.uncons hp
2190− case r of
2191− Just (Entry ts k, hp') -> do
2192− let duration = diffAbsTime ts t
2193− lt = round (lifetime * 1000) :: MilliSecond64
2194− if fromRelTime duration >= lt
2203+ let hres = H.uncons hp
2204+ case hres of
2205+ Just (Entry ts key, hp') -> do
2206+ let duration = diffAbsTime curTime ts
2207+ if duration >= lifetimeMs
21952208 then do
2196− let (v, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) k mp
2197− case v of
2209+ let (r, mp') = Map.updateLookupWithKey
2210+ (\_ _ -> Nothing) key mp
2211+ case r of
21982212 Nothing -> go hp' mp out
21992213 Just x -> do
2200− res <- extract x
2201− go hp' mp' (res `S.cons` out)
2214+ sess <- extract x
2215+ go hp' mp' (sess `S.cons` out)
22022216 else return (hp, mp, out)
22032217 Nothing -> return (hp, mp, out)
22042218
22052219 -- 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
22072221 timer = do
22082222 liftIO $ threadDelay (round $ tick * 1000000)
2209− now <- liftIO $ getTime Realtime
2210− return $ Left now
2223+ return Nothing