@@ -215,7 +215,7 @@ module Streamly.Fold
216 -- *** By Chunks
216 -- *** By Chunks
217 , chunksOf
217 , chunksOf
218− , spellsOf
218+ , sessionsOf
220 -- *** By Elements
220 -- *** By Elements
221 , splitBy
221 , splitBy
@@ -244,9 +244,6 @@ module Streamly.Fold
244 , groupsBy
244 , groupsBy
245 , groupsRollingBy
245 , groupsRollingBy
248− , sessionsByLifeTime
250 -- * Distributing
247 -- * Distributing
252 -- The 'Applicative' instance of a distributing 'Fold' distributes one copy
249 -- The 'Applicative' instance of a distributing 'Fold' distributes one copy
@@ -285,29 +282,60 @@ module Streamly.Fold
286 -- Direct items in the input stream to different folds using a function to
283 -- Direct items in the input stream to different folds using a function to
287 -- select the fold. This is useful to demultiplex the input stream.
284 -- select the fold. This is useful to demultiplex the input stream.
288− , partitionByM
285+ -- , partitionByM
289− , partitionBy
286+ -- , partitionBy
290 , partition
287 , partition
292 -- * Demultiplexing
289 -- * Demultiplexing
294− , demuxWith
291+ -- , demuxWith
296− , demuxWith_
293+ -- , demuxWith_
298 -- * Classifying
295 -- * Classifying
299 , classify
296 , classify
300− , classifyWith
297+ -- , classifyWith
302 -- * Unzipping
299 -- * Unzipping
301+ -- These can be expressed using lmap/lmapM and unzip
304− , unzipWith
302+ -- , unzipWith
305− , unzipWithM
303+ -- , unzipWithM
307 -- * Nested Folds
305 -- * Nested Folds
308 -- , concatMap
306 -- , concatMap
309 -- , chunksOf
307 -- , chunksOf
310 , duplicate -- experimental
308 , duplicate -- experimental
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.
317+ -- The term "chunk" is used for a space window and the term "session" is
318+ -- used for a time window.
320+ -- ** Tumbling Windows
321+ -- | A new window starts after the previous window is finished.
322+ -- , classifyChunksOf
323+ , classifySessionsOf
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
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
@@ -1039,18 +1067,18 @@ lcatMaybes = lfilter isJust . lmap fromJust
1040-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
1068-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
1041-- XXX add this example after fixing the serial stream rate control
1069-- 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
1043-- > [3,7,11,15,19]
1071-- > [3,7,11,15,19]
1045-- | Group the input stream into windows of @n@ second each and then fold each
1073-- | Group the input stream into windows of @n@ second each and then fold each
1046-- group using the provided fold function.
1074-- group using the provided fold function.
1048-- @since 0.7.0
1076-- @since 0.7.0
1049−{-# INLINE spellsOf #-}
1077+{-# INLINE sessionsOf #-}
1050−spellsOf
1078+sessionsOf
1051 :: (IsStream t, MonadAsync m)
1079 :: (IsStream t, MonadAsync m)
1052 => Double -> Fold m a b -> t m a -> t m b
1080 => 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
1055 s = S.map Just m `parallel` S.repeatM timeout
1083 s = S.map Just m `parallel` S.repeatM timeout
1056 timeout = do
1084 timeout = do
@@ -1886,7 +1914,7 @@ demuxWith f kv = Fold step initial extract
1886 Just (Fold step' acc extract') -> do
1914 Just (Fold step' acc extract') -> do
1887 !r <- acc >>= \x -> step' x a
1915 !r <- acc >>= \x -> step' x a
1888 return $ Map.insert k (Fold step' (return r) extract') mp
1916 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)
1891-- | Fold a stream of key value pairs using a map of specific folds for each
1919-- | Fold a stream of key value pairs using a map of specific folds for each
1892-- key into a map from keys to the results of fold outputs of the corresponding
1920-- 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
1937 -- up never fails
1965 -- up never fails
1938 case Map.lookup (f a) mp of
1966 case Map.lookup (f a) mp of
1939 Nothing -> return mp
1967 Nothing -> return mp
1940− Just (Fold step' acc extract') -> do
1968+ Just (Fold step' acc _) -> do
1941− acc >>= \x -> step' x a
1969+ _ <- acc >>= \x -> step' x a
1942 return mp
1970 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 ()
1945-- | Given a stream of key value pairs and a map from keys to folds, fold the
1973-- | Given a stream of key value pairs and a map from keys to folds, fold the
1946-- values for each key using the corresponding folds, discarding the outputs.
1974-- values for each key using the corresponding folds, discarding the outputs.
@@ -2108,81 +2136,167 @@ lchunksOf n f1 f2 = undefined
2110------------------------------------------------------------------------------
2138------------------------------------------------------------------------------
2111−-- Session Windows
2139+-- Windowed classification
2112------------------------------------------------------------------------------
2140------------------------------------------------------------------------------
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.
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".
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
2156+------------------------------------------------------------------------------
2157+-- Keyed Sliding Windows
2158+------------------------------------------------------------------------------
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
2169+classifySlidingChunks wsize wslide (Fold step initial extract) str
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
2186+classifySlidingSessions tick interval slide (Fold step initial extract) str
2190+------------------------------------------------------------------------------
2191+-- Keyed Session Windows
2192+------------------------------------------------------------------------------
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 #-}
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
2205+classifyChunksBy spanout reset (Fold step initial extract) str = undefined
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.
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
2218+classifyKeepAliveChunks spanout = classifyChunksBy spanout True
2114−-- | @sessionsByLifeTime lifetime tick k f stream@ groups together all input
2221+-- | @classifySessionsBy tick timeout reset f stream@ groups together all input
2115−-- stream elements that belong to the same session. @lifetime@ is the maximum
2222+-- stream elements that belong to the same session. @timeout@ is the maximum
2116-- lifetime of a session in seconds. All elements belonging to a session are
2223-- lifetime of a session in seconds. All elements belonging to a session are
2224+-- purged after this duration. If "reset" is 'Ture' then the timeout is reset
2117−-- purged after this duration. Session duration is measured using the
2225+-- after every event received in the session. Session duration is measured
2118−-- timestamp of the first element seen for that session. To detect session
2226+-- using the timestamp of the first element seen for that session. To detect
2119−-- timeouts, a monotonic event time clock is maintained using the timestamps
2227+-- session timeouts, a monotonic event time clock is maintained using the
2120−-- seen in the inputs and a timer with a tick duration specified by @tick@.
2228+-- timestamps seen in the inputs and a timer with a tick duration specified by
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
2231+-- @session key@ is a key that uniquely identifies the session for the given
2124−-- identifies the session for the given element, @timestamp@ characterizes the
2232+-- element, @timestamp@ characterizes the time when the input element was
2125−-- time when the input element was generated, this is an absolute time measured
2233+-- generated, this is an absolute time measured from some @Epoch@. @session
2126−-- from some @Epoch@. @session close@ is a boolean indicating whether this
2234+-- close@ is a boolean indicating whether this element marks the closing of the
2127−-- element marks the closing of the session. When an input element with
2235+-- session. When an input element with @session close@ set to @True@ is seen
2128−-- @session close@ set to @True@ is seen the session is purged immediately.
2236+-- the session is purged immediately.
2130-- All the input elements belonging to a session are collected using the fold
2238-- All the input elements belonging to a session are collected using the fold
2239+-- @f@. The session key and the fold result are emitted in the output stream
2131−-- @f@. The fold result is emitted when the session is purged either via the
2240+-- when the session is purged either via the session close event or via the
2132−-- session close event or via the session liftime timeout.
2241+-- session liftime timeout.
2134-- @since 0.7.0
2243-- @since 0.7.0
2135−{-# INLINABLE sessionsByLifeTime #-}
2244+{-# INLINABLE classifySessionsBy #-}
2136−sessionsByLifeTime
2245+classifySessionsBy
2137 :: (IsStream t, MonadAsync m, Ord k)
2246 :: (IsStream t, MonadAsync m, Ord k)
2138− => Double -- lifetime of a session
2139− -> Double -- timer tick in secoonds
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
2140− -> (a -> (k, AbsTime, Bool)) -- (key, timestamp, session close)
2251+ -> t m (k, a, Bool, AbsTime) -- ^ session key, timestamp, close event, data
2142− -> t m a
2252+ -> t m (k, b)
2144−sessionsByLifeTime lifetime tick getKey (Fold step initial extract) str =
2253+classifySessionsBy tick timeout reset (Fold step initial extract) str =
2145 S.concatMap (\(Tuple4' _ _ _ s) -> s) $ S.scanlM' sstep szero stream
2254 S.concatMap (\(Tuple4' _ _ _ s) -> s) $ S.scanlM' sstep szero stream
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.
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.
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)
2161 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
2259 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
2162 szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty S.nil
2260 szero = Tuple4' (toAbsTime (0 :: MilliSecond64)) H.empty Map.empty S.nil
2164 -- Got a new stream input element
2262 -- 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
2166 -- deleting a key from the heap is expensive, so we never delete a
2267 -- deleting a key from the heap is expensive, so we never delete a
2167 -- key, we just purge it from the Map and it gets purged from the
2268 -- key, we just purge it from the Map and it gets purged from the
2168 -- heap on timeout. We just need an extra lookup in the Map when
2269 -- heap on timeout. We just need an extra lookup in the Map when
2169 -- the key is purged from the heap, that should not be expensive.
2270 -- the key is purged from the heap, that should not be expensive.
2170− let (key, ts, closed) = getKey a
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.
2278+ -- XXX if the key is an Int, we can also use an IntMap for slightly
2279+ -- better performance.
2171− accumulate v = maybe initial return v >>= \r -> step r a
2281+ let accumulate v = do
2172− in if closed
2282+ Tuple' _ old <- maybe (initial >>= return . Tuple' ts) return v
2284+ return $ Tuple' ts new
2174 let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
2287 let (r, mp') = Map.updateLookupWithKey (\_ _ -> Nothing) key mp
2175− acc <- accumulate r
2288+ Tuple' _ acc <- accumulate r
2176 res <- extract acc
2289 res <- extract acc
2177− return $ Tuple4' evTime hp mp' (S.yield res)
2290+ return $ Tuple4' evTime hp mp' (S.yield (key, res))
2179 let r = Map.lookup key mp
2292 let r = Map.lookup key mp
2180 acc <- accumulate r
2293 acc <- accumulate r
2181 let mp' = Map.insert key acc mp
2294 let mp' = Map.insert key acc mp
2182− let expiry = addToAbsTime ts lifetimeMs
2183 let hp' =
2295 let hp' =
2184 case r of
2296 case r of
2298+ let expiry = addToAbsTime ts timeoutMs
2185− Nothing -> H.insert (Entry expiry key) hp
2299+ in H.insert (Entry expiry key) hp
2186 Just _ -> hp
2300 Just _ -> hp
2187 -- Event time is maintained as monotonically increasing
2301 -- Event time is maintained as monotonically increasing
2188 -- time. If we have lagged behind any of the timestamps
2302 -- time. If we have lagged behind any of the timestamps
@@ -2204,15 +2318,24 @@ sessionsByLifeTime lifetime tick getKey (Fold step initial extract) str =
2204 case hres of
2318 case hres of
2205 Just (Entry ts key, hp') -> do
2319 Just (Entry ts key, hp') -> do
2206 let duration = diffAbsTime curTime ts
2320 let duration = diffAbsTime curTime ts
2207− if duration >= lifetimeMs
2321+ if duration >= timeoutMs
2209 let (r, mp') = Map.updateLookupWithKey
2323 let (r, mp') = Map.updateLookupWithKey
2210 (\_ _ -> Nothing) key mp
2324 (\_ _ -> Nothing) key mp
2211 case r of
2325 case r of
2212− Nothing -> go hp' mp out
2326+ Nothing -> go hp' mp' out
2213− Just x -> do
2327+ Just (Tuple' latestTS acc) -> do
2328+ let dur = diffAbsTime curTime latestTS
2329+ if dur >= timeoutMs || not reset
2214− sess <- extract x
2331+ sess <- extract acc
2215− go hp' mp' (sess `S.cons` out)
2332+ go hp' mp' ((key, sess) `S.cons` out)
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
2216 else return (hp, mp, out)
2339 else return (hp, mp, out)
2217 Nothing -> return (hp, mp, out)
2340 Nothing -> return (hp, mp, out)
@@ -2221,3 +2344,66 @@ sessionsByLifeTime lifetime tick getKey (Fold step initial extract) str =
2221 timer = do
2344 timer = do
2222 liftIO $ threadDelay (round $ tick * 1000000)
2345 liftIO $ threadDelay (round $ tick * 1000000)
2223 return Nothing
2346 return Nothing
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.
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
2360+classifyKeepAliveSessions timeout = classifySessionsBy 1 timeout True
2362+------------------------------------------------------------------------------
2363+-- Keyed tumbling windows
2364+------------------------------------------------------------------------------
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.
2370+-- XXX instead of using the early termination flag in the stream, we can use an
2371+-- early terminating fold instead.
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.
2380+{-# INLINABLE 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
2387+classifyChunksOf wsize = classifyChunksBy wsize False
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.
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.
2402+{-# INLINABLE 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
2409+classifySessionsOf interval = classifySessionsBy 1 interval False