@@ -28,6 +28,12 @@
28-- Similarly, a partitioning applicative can partition the input among the
28-- Similarly, a partitioning applicative can partition the input among the
29-- constituent folds. There are many other ways to combine folds.
29-- constituent folds. There are many other ways to combine folds.
31+-- All combinators in this module are of true streaming nature. For example,
32+-- the splitting combinators accept a fold which consumes the elements in the
33+-- split segments as soon as they are generated by the producer, segments are
34+-- not buffered in memory, resulting in constant memory usage. The folds used
35+-- in these combinators can be considered as continuations.
31-- > import qualified Streamly.Fold as FL
37-- > import qualified Streamly.Fold as FL
@@ -192,6 +198,7 @@ module Streamly.Fold
192 , break -- breakPre
198 , break -- breakPre
193 -- , breakPost
199 -- , breakPost
194 -- , breakOn
200 -- , breakOn
195 -- , spanBy
202 -- , spanBy
196 -- , spanRollingBy
203 -- , spanRollingBy
@@ -199,19 +206,16 @@ module Streamly.Fold
199 -- breakOnSeq
206 -- breakOnSeq
201 -- ** Splitting
208 -- ** Splitting
209+ -- | Streams can be split into segments in space or in time. We use the
210+ -- term @chunk@ to refer to a spatial length of the stream (spatial window)
211+ -- and the term @session@ to refer to a length in time (time window).
203− -- In imperative terms grouped folding can be considered as a nested loop
213+ -- In imperative terms, grouped folding can be considered as a nested loop
204 -- where we loop over the stream to group elements and then loop over
214 -- where we loop over the stream to group elements and then loop over
205 -- individual groups to fold them to a single value that is yielded in the
215 -- individual groups to fold them to a single value that is yielded in the
206 -- output stream.
216 -- output stream.
208− -- Note that these grouping folds are true streaming folds that never
218+ -- , groupScan
209− -- accumulate the group in memory before folding, i.e. the group elements
210− -- are consumed by the folds as they are yielded by the stream. Therefore,
211− -- the whole computation runs in constant space.
212− -- In contrast, we can simply use a scan on the stream to buffer the whole
213− -- groups in memory and then map a fold on it to fold the groups. This kind
214− -- of grouping and folding would not work well when the group size is big.
216 -- *** By Chunks
220 -- *** By Chunks
217 , chunksOf
221 , chunksOf
@@ -335,7 +339,9 @@ module Streamly.Fold
335 , classifySlidingChunks
339 , classifySlidingChunks
336 , classifySlidingSessions
340 , classifySlidingSessions
338−
342+ -- ** Sliding Window Buffers
343+ -- , slidingChunkBuffer
344+ -- , slidingSessionBuffer
@@ -362,9 +368,9 @@ import qualified Data.Map.Strict as Map
362import Streamly (MonadAsync, parallel)
368import Streamly (MonadAsync, parallel)
363import Streamly.Fold.Types (Fold(..))
369import Streamly.Fold.Types (Fold(..))
364import Streamly.Mem.Array (Array)
370import Streamly.Mem.Array (Array)
371+-- import Streamly.Mem.Ring (Ring)
365import Streamly.Streams.Serial (SerialT)
372import Streamly.Streams.Serial (SerialT)
366import Streamly.Streams.StreamK (IsStream())
373import Streamly.Streams.StreamK (IsStream())
367−-- import Streamly.Time.Clock (Clock(..), getTime)
368import Streamly.Time.Units
374import Streamly.Time.Units
369 (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
375 (AbsTime, MilliSecond64(..), addToAbsTime, diffAbsTime, toRelTime,
370 toAbsTime)
376 toAbsTime)
@@ -1044,6 +1050,45 @@ splitAt n (Fold stepL initialL extractL) (Fold stepR initialR extractR) =
1044-- N-ary APIs
1050-- N-ary APIs
1045------------------------------------------------------------------------------
1051------------------------------------------------------------------------------
1053+------------------------------------------------------------------------------
1054+-- Generalized grouping
1055+------------------------------------------------------------------------------
1057+-- This combinator is the most general grouping combinator and can be used to
1058+-- implement all other grouping combinators.
1060+-- XXX check if this can implement the splitOn combinator i.e. we can slide in
1061+-- new elements, slide out old elements and incrementally compute the hash.
1062+-- Also, can we implement the windowed classification combinators using this?
1064+-- In fact this is a parse. Instead of using a special return value in the fold
1065+-- we are using a mapping function.
1067+-- Note that 'scanl'' (usually followed by a map to extract the desired value
1068+-- from the accumulator) can be used to realize many implementations e.g. a
1069+-- sliding window implementation. A scan followed by a mapMaybe is also a good
1070+-- pattern to express many problems where we want to emit a filtered output and
1071+-- not emit an output on every input.
1073+-- Passing on of the initial accumulator value to the next fold is equivalent
1074+-- to returning the leftover concept.
1077+-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
1078+-- @splitter@ is applied on the accumulator of the fold every time an item is
1079+-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
1080+-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
1081+-- in the output stream and the initial value of the accumulator for the next
1082+-- group's fold. This allows us to control whether to start fresh for the next
1083+-- fold or to continue from the previous fold's output.
1085+{-# INLINE groupScan #-}
1087+ :: (IsStream t, Monad m)
1088+ => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
1089+groupScan split fold m = undefined
1047-- | Group the input stream into groups of @n@ elements each and then fold each
1092-- | Group the input stream into groups of @n@ elements each and then fold each
1048-- group using the provided fold function.
1093-- group using the provided fold function.
@@ -1688,21 +1733,21 @@ splitSuffixOnAny subseq f m = undefined
1690------------------------------------------------------------------------------
1735------------------------------------------------------------------------------
1691−-- Grouped by order
1736+-- Reorder in sequence
1692------------------------------------------------------------------------------
1737------------------------------------------------------------------------------
1695-- Buffer until the next element in sequence arrives. The function argument
1740-- Buffer until the next element in sequence arrives. The function argument
1696-- determines the difference in sequence numbers. This could be useful in
1741-- determines the difference in sequence numbers. This could be useful in
1697-- implementing sequenced streams, for example, TCP reassembly.
1742-- implementing sequenced streams, for example, TCP reassembly.
1698−{-# INLINE foldOrderedBy #-}
1743+{-# INLINE reassembleBy #-}
1699−foldOrderedBy
1744+reassembleBy
1700 :: (IsStream t, Monad m)
1745 :: (IsStream t, Monad m)
1701− => (forall n. Monad n => Fold n a b)
1746+ => Fold m a b
1702 -> (a -> a -> Int)
1747 -> (a -> a -> Int)
1703 -> t m a
1748 -> t m a
1704 -> t m b
1749 -> t m b
1705−foldOrderedBy = undefined
1750+reassembleBy = undefined
1708------------------------------------------------------------------------------
1753------------------------------------------------------------------------------
@@ -2187,6 +2232,48 @@ classifySlidingSessions tick interval slide (Fold step initial extract) str
2187 = undefined
2232 = undefined
2235+------------------------------------------------------------------------------
2236+-- Sliding Window Buffers
2237+------------------------------------------------------------------------------
2239+-- These buffered versions could be faster than concurrent incremental folds of
2240+-- all overlapping windows as in many cases we may not need all the values to
2241+-- compute the fold, we can just compute the result using the old value and new
2242+-- value. However, we may need the buffer once in a while, for example for
2243+-- string search we usually compute the hash incrementally but when the hash
2244+-- matches the hash of the pattern we need to compare the whole string.
2246+-- XXX we should be able to implement sequence based splitting combinators
2247+-- using this combinator.
2250+-- | Buffer n elements of the input in a ring buffer. When t new elements are
2251+-- collected, slide the window to remove the same number of oldest elements,
2252+-- insert the new elements, and apply an incremental fold on the sliding
2253+-- window, supplying the outgoing elements, the new ring buffer as arguments.
2255+ :: (IsStream t, Monad m, Ord a, Storable a)
2256+ => Int -- window size
2257+ -> Int -- window slide
2258+ -> Fold m (Ring a, Array a) b
2261+slidingChunkBuffer = undefined
2263+-- Buffer n seconds worth of stream elements of the input in a radix tree.
2264+-- Every t seconds, remove the items that are older than n seconds, and apply
2265+-- an incremental fold on the sliding window, supplying the outgoing elements,
2266+-- and the new radix tree buffer as arguments.
2267+slidingSessionBuffer
2268+ :: (IsStream t, Monad m, Ord a, Storable a)
2269+ => Int -- window size
2270+ -> Int -- tick size
2271+ -> Fold m (RTree a, Array a) b
2274+slidingSessionBuffer = undefined
2190------------------------------------------------------------------------------
2277------------------------------------------------------------------------------
2191-- Keyed Session Windows
2278-- Keyed Session Windows
2192------------------------------------------------------------------------------
2279------------------------------------------------------------------------------