@@ -215,7 +215,7 @@ module Streamly.Fold
215215
216216 -- *** By Chunks
217217 , chunksOf
218− , spellsOf
218+ , sessionsOf
219219
220220 -- *** By Elements
221221 , splitBy
@@ -244,9 +244,6 @@ module Streamly.Fold
244244 , groupsBy
245245 , groupsRollingBy
246246
247− -- ** Sessions
248− , sessionsByLifeTime
249−
250247 -- * Distributing
251248 -- |
252249 -- The 'Applicative' instance of a distributing 'Fold' distributes one copy
@@ -285,29 +282,60 @@ module Streamly.Fold
285282 -- |
286283 -- Direct items in the input stream to different folds using a function to
287284 -- select the fold. This is useful to demultiplex the input stream.
288− , partitionByM
289− , partitionBy
285+ -- , partitionByM
286+ -- , partitionBy
290287 , partition
291288
292289 -- * Demultiplexing
293290 , demux
294− , demuxWith
291+ -- , demuxWith
295292 , demux_
296− , demuxWith_
293+ -- , demuxWith_
297294
298295 -- * Classifying
299296 , classify
300− , classifyWith
297+ -- , classifyWith
301298
302299 -- * Unzipping
303300 , unzip
304− , unzipWith
305− , unzipWithM
301+ -- These can be expressed using lmap/lmapM and unzip
302+ -- , unzipWith
303+ -- , unzipWithM
306304
307305 -- * Nested Folds
308306 -- , concatMap
309307 -- , chunksOf
310308 , duplicate -- experimental
309+
310+ -- * Windowed Classification
311+ -- | Split the stream into windows or chunks in space or time. Each window
312+ -- can be associated with a key, all events associated with a particular
313+ -- key in the window can be folded to a single result. The stream is split
314+ -- into windows of specified size, the window can be terminated early if
315+ -- the closing flag is specified in the input stream.
316+ --
317+ -- The term "chunk" is used for a space window and the term "session" is
318+ -- used for a time window.
319+
320+ -- ** Tumbling Windows
321+ -- | A new window starts after the previous window is finished.
322+ -- , classifyChunksOf
323+ , classifySessionsOf
324+
325+ -- ** Keep Alive Windows
326+ -- | The window size is extended if an event arrives within the specified
327+ -- window size. This can represent sessions with idle or inactive timeout.
328+ -- , classifyKeepAliveChunks
329+ , classifyKeepAliveSessions
330+
331+ {-
332+ -- ** Sliding Windows
333+ -- | A new window starts after the specified slide from the previous
334+ -- window. Therefore windows can overlap.
335+ , classifySlidingChunks
336+ , classifySlidingSessions
337+ -}
338+
311339 )
312340where
313341
@@ -1039,18 +1067,18 @@ lcatMaybes = lfilter isJust . lmap fromJust
10391067
10401068-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
10411069-- XXX add this example after fixing the serial stream rate control
1042−-- >>> S.toList $ S.take 5 $ spellsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
1070+-- >>> S.toList $ S.take 5 $ sessionsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
10431071-- > [3,7,11,15,19]
10441072--
10451073-- | Group the input stream into windows of @n@ second each and then fold each
10461074-- group using the provided fold function.
10471075--
10481076-- @since 0.7.0
1049−{-# INLINE spellsOf #-}
1050−spellsOf
1077+{-# INLINE sessionsOf #-}
1078+sessionsOf
10511079 :: (IsStream t, MonadAsync m)
10521080 => Double -> Fold m a b -> t m a -> t m b
1053−spellsOf n f m = splitSuffixBy' isNothing (lcatMaybes f) s
1081+sessionsOf n f m = splitSuffixBy' isNothing (lcatMaybes f) s
10541082 where
10551083 s = S.map Just m `parallel` S.repeatM timeout
10561084 timeout = do
@@ -1886,7 +1914,7 @@ demuxWith f kv = Fold step initial extract
18861914 Just (Fold step' acc extract') -> do
18871915 !r <- acc >>= \x -> step' x a
18881916 return $ Map.insert k (Fold step' (return r) extract') mp
1889− extract = mapM (\(Fold s acc e) -> acc >>= e)
1917+ extract = mapM (\(Fold _ acc e) -> acc >>= e)
18901918
18911919-- | Fold a stream of key value pairs using a map of specific folds for each
18921920-- key into a map from keys to the results of fold outputs of the corresponding
@@ -1937,10 +1965,10 @@ demuxWith_ f kv = Fold step initial extract
19371965 -- up never fails
19381966 case Map.lookup (f a) mp of
19391967 Nothing -> return mp
1940− Just (Fold step' acc extract') -> do
1941− acc >>= \x -> step' x a
1968+ Just (Fold step' acc _) -> do
1969+ _ <- acc >>= \x -> step' x a
19421970 return mp
1943− extract mp = mapM (\(Fold s acc e) -> acc >>= e) mp >> return ()
1971+ extract mp = mapM (\(Fold _ acc e) -> acc >>= e) mp >> return ()
19441972
19451973-- | Given a stream of key value pairs and a map from keys to folds, fold the
19461974-- values for each key using the corresponding folds, discarding the outputs.
@@ -2108,81 +2136,167 @@ lchunksOf n f1 f2 = undefined
21082136-}
21092137
21102138------------------------------------------------------------------------------
2111−-- Session Windows
2139+-- Windowed classification
21122140------------------------------------------------------------------------------
21132141
2114−-- | @sessionsByLifeTime lifetime tick k f stream@ groups together all input
2115−-- stream elements that belong to the same session. @lifetime@ is the maximum
2142+-- We divide the stream into windows or chunks in space or time and each window
2143+-- can be associated with a key, all events associated with a particular key in
2144+-- the window can be folded to a single result. The stream can be split into
2145+-- windows by size or by using a split predicate on the elements in the stream.
2146+-- For example, when we receive a closing flag, we can close the window.
2147+--
2148+-- A "chunk" is a space window and a "session" is a time window. Are there any
2149+-- other better short words to describe them. An alternative is to use
2150+-- "swindow" and "twindow". Another word for "session" could be "spell".
2151+--
2152+-- TODO: To mark the position in space or time we can have Indexed or
2153+-- TimeStamped types. That can make it easy to deal with the position indices
2154+-- or timestamps.
2155+
2156+------------------------------------------------------------------------------
2157+-- Keyed Sliding Windows
2158+------------------------------------------------------------------------------
2159+
2160+{-
2161+{-# INLINABLE classifySlidingChunks #-}
2162+classifySlidingChunks
2163+ :: (IsStream t, MonadAsync m, Ord k)
2164+ => Int -- ^ window size
2165+ -> Int -- ^ window slide
2166+ -> Fold m a b -- ^ Fold to be applied to window events
2167+ -> t m (k, a, Bool) -- ^ window key, data, close event
2168+ -> t m (k, b)
2169+classifySlidingChunks wsize wslide (Fold step initial extract) str
2170+ = undefined
2171+
2172+-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
2173+-- slide the send window when an ack is received and we slide the receive
2174+-- window when a sequence is complete. Sliding is stateful in case of TCP,
2175+-- sliding releases the send buffer or makes data available to the user from
2176+-- the receive buffer.
2177+{-# INLINABLE classifySlidingSessions #-}
2178+classifySlidingSessions
2179+ :: (IsStream t, MonadAsync m, Ord k)
2180+ => Double -- ^ timer tick in seconds
2181+ -> Double -- ^ time window size
2182+ -> Double -- ^ window slide
2183+ -> Fold m a b -- ^ Fold to be applied to window events
2184+ -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
2185+ -> t m (k, b)
2186+classifySlidingSessions tick interval slide (Fold step initial extract) str
2187+ = undefined
2188+-}
2189+
2190+------------------------------------------------------------------------------
2191+-- Keyed Session Windows
2192+------------------------------------------------------------------------------
2193+
2194+{-
2195+-- | Keyed variable size space windows. Close the window if we do not receive a
2196+-- window event in the next "spaceout" elements.
2197+{-# INLINABLE classifyChunksBy #-}
2198+classifyChunksBy
2199+ :: (IsStream t, MonadAsync m, Ord k)
2200+ => Int -- ^ window spaceout (spread)
2201+ -> Bool -- ^ reset the spaceout when a chunk window element is received
2202+ -> Fold m a b -- ^ Fold to be applied to chunk window elements
2203+ -> t m (k, a, Bool) -- ^ chunk key, data, last element
2204+ -> t m (k, b)
2205+classifyChunksBy spanout reset (Fold step initial extract) str = undefined
2206+
2207+-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
2208+-- received within the chunk size window. The chunk gets closed only if no
2209+-- element is received within the chunk window.
2210+--
2211+{-# INLINABLE classifyKeepAliveChunks #-}
2212+classifyKeepAliveChunks
2213+ :: (IsStream t, MonadAsync m, Ord k)
2214+ => Int -- ^ window spaceout (spread)
2215+ -> Fold m a b -- ^ Fold to be applied to chunk window elements
2216+ -> t m (k, a, Bool) -- ^ chunk key, data, last element
2217+ -> t m (k, b)
2218+classifyKeepAliveChunks spanout = classifyChunksBy spanout True
2219+-}
2220+
2221+-- | @classifySessionsBy tick timeout reset f stream@ groups together all input
2222+-- stream elements that belong to the same session. @timeout@ is the maximum
21162223-- lifetime of a session in seconds. All elements belonging to a session are
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@.
2121−--
2122−-- The function @k@ returns a 3-tuple @(session key, timestamp, session close)@
2123−-- from an input element in the stream. @session key@ is a key that uniquely
2124−-- identifies the session for the given element, @timestamp@ characterizes the
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
2128−-- @session close@ set to @True@ is seen the session is purged immediately.
2224+-- purged after this duration. If "reset" is 'Ture' then the timeout is reset
2225+-- after every event received in the session. Session duration is measured
2226+-- using the timestamp of the first element seen for that session. To detect
2227+-- session timeouts, a monotonic event time clock is maintained using the
2228+-- timestamps seen in the inputs and a timer with a tick duration specified by
2229+-- @tick@.
2230+--
2231+-- @session key@ is a key that uniquely identifies the session for the given
2232+-- element, @timestamp@ characterizes the time when the input element was
2233+-- generated, this is an absolute time measured from some @Epoch@. @session
2234+-- close@ is a boolean indicating whether this element marks the closing of the
2235+-- session. When an input element with @session close@ set to @True@ is seen
2236+-- the session is purged immediately.
21292237--
21302238-- 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.
2239+-- @f@. The session key and the fold result are emitted in the output stream
2240+-- when the session is purged either via the session close event or via the
2241+-- session liftime timeout.
21332242--
21342243-- @since 0.7.0
2135−{-# INLINABLE sessionsByLifeTime #-}
2136−sessionsByLifeTime
2244+{-# INLINABLE classifySessionsBy #-}
2245+classifySessionsBy
21372246 :: (IsStream t, MonadAsync m, Ord k)
2138− => Double -- lifetime of a session
2139− -> Double -- timer tick in secoonds
2140− -> (a -> (k, AbsTime, Bool)) -- (key, timestamp, session close)
2141− -> Fold m a b
2142− -> t m a
2143− -> t m b
2144−sessionsByLifeTime lifetime tick getKey (Fold step initial extract) str =
2247+ => Double -- ^ timer tick in seconds
2248+ -> Double -- ^ session timeout
2249+ -> Bool -- ^ reset the timeout when an event is received
2250+ -> Fold m a b -- ^ Fold to be applied to session events
2251+ -> t m (k, a, Bool, AbsTime) -- ^ session key, timestamp, close event, data
2252+ -> t m (k, b)
2253+classifySessionsBy tick timeout reset (Fold step initial extract) str =
21452254 S.concatMap (\(Tuple4' _ _ _ s) -> s) $ S.scanlM' sstep szero stream
21462255
21472256 where
21482257
2149− -- XXX what if the timestamps and the system time are too far off, we may
2150− -- accumulate too many entries, have a protection in that case? If the
2151− -- incoming logs seem too new or too old than the current system time then
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− --
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)
2258+ timeoutMs = toRelTime (round (timeout * 1000) :: MilliSecond64)
21612259 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
21622260 szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty S.nil
21632261
21642262 -- Got a new stream input element
2165− sstep (Tuple4' evTime hp mp _) (Just a) =
2263+ sstep (Tuple4' evTime hp mp _) (Just (key, a, closing, ts)) =
2264+ -- XXX we should use a heap in pinned memory to scale it to a large
2265+ -- size
2266+ --
21662267 -- deleting a key from the heap is expensive, so we never delete a
21672268 -- key, we just purge it from the Map and it gets purged from the
21682269 -- heap on timeout. We just need an extra lookup in the Map when
21692270 -- the key is purged from the heap, that should not be expensive.
2170− let (key, ts, closed) = getKey a
2171− accumulate v = maybe initial return v >>= \r -> step r a
2172− in if closed
2271+ --
2272+ -- To detect session inactivity we keep a timestamp of the latest event
2273+ -- in the Map along with the fold result. When we purge the session
2274+ -- from the heap we match the timestamp in the heap with the timestamp
2275+ -- in the Map, if the latest timestamp is newer and has not expired we
2276+ -- reinsert the key in the heap.
2277+ --
2278+ -- XXX if the key is an Int, we can also use an IntMap for slightly
2279+ -- better performance.
2280+ --
2281+ let accumulate v = do
2282+ Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
2283+ new <- step old a
2284+ return $ Tuple' ts new
2285+ in if closing
21732286 then do
21742287 let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
2175− acc <- accumulate r
2288+ Tuple' _ acc <- accumulate r
21762289 res <- extract acc
2177− return $ Tuple4' evTime hp mp' (S.yield res)
2290+ return $ Tuple4' evTime hp mp' (S.yield (key, res))
21782291 else do
21792292 let r = Map.lookup key mp
21802293 acc <- accumulate r
21812294 let mp' = Map.insert key acc mp
2182− let expiry = addToAbsTime ts lifetimeMs
21832295 let hp' =
21842296 case r of
2185− Nothing -> H.insert (Entry expiry key) hp
2297+ Nothing ->
2298+ let expiry = addToAbsTime ts timeoutMs
2299+ in H.insert (Entry expiry key) hp
21862300 Just _ -> hp
21872301 -- Event time is maintained as monotonically increasing
21882302 -- time. If we have lagged behind any of the timestamps
@@ -2204,15 +2318,24 @@ sessionsByLifeTime lifetime tick getKey (Fold step initial extract) str =
22042318 case hres of
22052319 Just (Entry ts key, hp') -> do
22062320 let duration = diffAbsTime curTime ts
2207− if duration >= lifetimeMs
2321+ if duration >= timeoutMs
22082322 then do
22092323 let (r, mp') = Map.updateLookupWithKey
22102324 (\_ _ -> Nothing) key mp
22112325 case r of
2212− Nothing -> go hp' mp out
2213− Just x -> do
2214− sess <- extract x
2215− go hp' mp' (sess `S.cons` out)
2326+ Nothing -> go hp' mp' out
2327+ Just (Tuple' latestTS acc) -> do
2328+ let dur = diffAbsTime curTime latestTS
2329+ if dur >= timeoutMs || not reset
2330+ then do
2331+ sess <- extract acc
2332+ go hp' mp' ((key, sess) `S.cons` out)
2333+ else
2334+ -- reset the session timeout
2335+ let expiry = addToAbsTime latestTS timeoutMs
2336+ hp'' = H.insert (Entry expiry key) hp'
2337+ mp'' = Map.insert key (Tuple' latestTS acc) mp'
2338+ in go hp'' mp'' out
22162339 else return (hp, mp, out)
22172340 Nothing -> return (hp, mp, out)
22182341
@@ -2221,3 +2344,66 @@ sessionsByLifeTime lifetime tick getKey (Fold step initial extract) str =
22212344 timer = do
22222345 liftIO $ threadDelay (round $ tick * 1000000)
22232346 return Nothing
2347+
2348+-- | Like 'classifySessionsOf' but the session is kept alive if an event is
2349+-- received within the session window. The session times out and gets closed
2350+-- only if no event is received within the specified session window size.
2351+--
2352+-- @since 0.7.0
2353+{-# INLINABLE classifyKeepAliveSessions #-}
2354+classifyKeepAliveSessions
2355+ :: (IsStream t, MonadAsync m, Ord k)
2356+ => Double -- ^ session inactive timeout
2357+ -> Fold m a b -- ^ Fold to be applied to session payload data
2358+ -> t m (k, a, Bool, AbsTime) -- ^ session key, data, close flag, timestamp
2359+ -> t m (k, b)
2360+classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
2361+
2362+------------------------------------------------------------------------------
2363+-- Keyed tumbling windows
2364+------------------------------------------------------------------------------
2365+
2366+-- Tumbling windows is a special case of sliding windows where the window slide
2367+-- is the same as the window size. Or it can be a special case of session
2368+-- windows where the reset flag is set to False.
2369+
2370+-- XXX instead of using the early termination flag in the stream, we can use an
2371+-- early terminating fold instead.
2372+
2373+{-
2374+-- | Split the stream into fixed size chunks of specified size. Within each
2375+-- such chunk fold the elements in buckets identified by the keys. A particular
2376+-- bucket fold can be terminated early if a closing flag is encountered in an
2377+-- element for that key.
2378+--
2379+-- @since 0.7.0
2380+{-# INLINABLE classifyChunksOf #-}
2381+classifyChunksOf
2382+ :: (IsStream t, MonadAsync m, Ord k)
2383+ => Int -- ^ window size
2384+ -> Fold m a b -- ^ Fold to be applied to window events
2385+ -> t m (k, a, Bool) -- ^ window key, data, close event
2386+ -> t m (k, b)
2387+classifyChunksOf wsize = classifyChunksBy wsize False
2388+-}
2389+
2390+-- | Split the stream into fixed size time windows of specified interval in
2391+-- seconds. Within each such window, fold the elements in buckets identified by
2392+-- the keys. A particular bucket fold can be terminated early if a closing flag
2393+-- is encountered in an element for that key. Once a fold is terminated the key
2394+-- and value for that bucket are emitted in the output stream.
2395+--
2396+-- Session @timestamp@ in the input stream is an absolute time from some epoch,
2397+-- characterizing the time when the input element was generated. To detect
2398+-- session window end, a monotonic event time clock is maintained synced with
2399+-- the timestamps with a clock resolution of 1 second.
2400+--
2401+-- @since 0.7.0
2402+{-# INLINABLE classifySessionsOf #-}
2403+classifySessionsOf
2404+ :: (IsStream t, MonadAsync m, Ord k)
2405+ => Double -- ^ time window size
2406+ -> Fold m a b -- ^ Fold to be applied to window events
2407+ -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
2408+ -> t m (k, b)
2409+classifySessionsOf interval = classifySessionsBy 1 interval False