2-- Module : Streamly.Internal.Data.Stream.IsStream.Reduce
3-- Copyright : (c) 2017 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
6-- Stability : experimental
9-- Reduce streams by streams, folds or parsers.
11module Streamly.Internal.Data.Stream.IsStream.Reduce
13 -- * Reduce By Streams
20 -- Reduce a stream by folding or parsing chunks of the stream. Functions
21 -- generally ending in these shapes:
24 -- f (Fold m a b) -> t m a -> t m b
25 -- f (Parser m a b) -> t m a -> t m b
29 -- | Apply folds on a stream.
36 -- | Element unaware grouping.
42 -- | Streams can be sliced into segments in space or in time. We use the
43 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
44 -- and the term @session@ to refer to a length in time (time window).
46 -- -- *** Using Element Separators
55 -- -- *** Splitting By Sequences
61 -- Keeping the delimiters
66 -- Splitting using multiple sequence separators
68 -- , splitOnAnySuffixSeq
69 -- , splitOnAnyPrefixSeq
71 -- -- *** Splitting By Streams
72 -- -- | Splitting a stream using another stream as separator.
74 -- ** Keyed Window Classification
76 -- | Split the stream into chunks or windows by position or time. Each
77 -- window can be associated with a key, all events associated with a
78 -- particular key in the window can be folded to a single result. The
79 -- window termination can be dynamically controlled by the fold.
81 -- The term "chunk" is used for a window defined by position of elements
82 -- and the term "session" is used for a time window.
84 -- *** Tumbling Windows
85 -- | A new window starts after the previous window is finished.
91 -- *** Keep Alive Windows
92 -- | The window size is extended if an event arrives within the specified
93 -- window size. This can represent sessions with idle or inactive timeout.
95 -- , classifyKeepAliveChunks
96 , classifyKeepAliveSessions
99 -- *** Sliding Windows
100 -- | A new window starts after the specified slide from the previous
101 -- window. Therefore windows can overlap.
102 , classifySlidingChunks
103 , classifySlidingSessions
104 -- *** Sliding Window Buffers
105 -- , slidingChunkBuffer
106 -- , slidingSessionBuffer
109 -- * Reduce By Parsers
110 -- ** Generic Parsing
111 -- | Apply parsers on a stream.
119 -- In imperative terms, grouped folding can be considered as a nested loop
120 -- where we loop over the stream to group elements and then loop over
121 -- individual groups to fold them to a single value that is yielded in the
124 , wordsBy -- stripAndCompactBy
129 -- -- *** Searching Sequences
130 -- , seqIndices -- search a sequence in the stream
132 -- -- *** Searching Multiple Sequences
133 -- , seqIndicesAny -- search any of the given sequence in the stream
135 -- -- -- ** Searching Streams
136 -- -- | Finding a stream within another stream.
138 -- * Nested splitting
149import Control.Concurrent (threadDelay)
150import Control.Exception (assert)
151import Control.Monad.Catch (MonadThrow)
152import Control.Monad.IO.Class (MonadIO(..))
153import Data.Heap (Entry(..))
154import Data.Kind (Type)
155import Data.Maybe (isNothing)
156import Foreign.Storable (Storable)
157import Streamly.Internal.Data.Fold.Type (Fold (..), Fold2 (..))
158import Streamly.Internal.Data.Parser (Parser (..))
159import Streamly.Internal.Data.Array.Foreign.Type (Array)
160import Streamly.Internal.Data.SVar (MonadAsync)
161import Streamly.Internal.Data.Stream.IsStream.Common
170import Streamly.Internal.Data.Stream.StreamK (IsStream)
171import Streamly.Internal.Data.Time.Units
172 ( AbsTime, MilliSecond64(..), addToAbsTime, toRelTime
174import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
176import qualified Data.Heap as H
177import qualified Data.Map.Strict as Map
178import qualified Streamly.Internal.Data.Array.Foreign.Type as A
179import qualified Streamly.Internal.Data.Fold as FL
180import qualified Streamly.Internal.Data.Parser.ParserK.Type as PRK
181import qualified Streamly.Internal.Data.Parser.ParserD as PRD
182import qualified Streamly.Internal.Data.Stream.Parallel as Par
183import qualified Streamly.Internal.Data.Stream.Serial as Serial
184import qualified Streamly.Internal.Data.Stream.StreamD as D
185import qualified Streamly.Internal.Data.Stream.StreamK as K
187import Prelude hiding (concatMap)
191-- >>> import Prelude hiding (zipWith, concatMap, concat)
192-- >>> import qualified Streamly.Prelude as Stream
193-- >>> import Streamly.Internal.Data.Stream.IsStream as Stream
194-- >>> import qualified Streamly.Data.Fold as Fold
195-- >>> import qualified Streamly.Internal.Data.Fold as Fold
196-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold
197-- >>> import qualified Streamly.Internal.Data.Parser as Parser
198-- >>> import qualified Streamly.Data.Array.Foreign as Array
200------------------------------------------------------------------------------
202------------------------------------------------------------------------------
204-- | Drop prefix from the input stream if present.
208-- /Unimplemented/ - Help wanted.
209{-# INLINE dropPrefix #-}
211 -- (Eq a, IsStream t, Monad m) =>
212 t m a -> t m a -> t m a
213dropPrefix = error "Not implemented yet!"
215-- | Drop all matching infix from the input stream if present. Infix stream
216-- may be consumed multiple times.
218-- Space: @O(n)@ where n is the length of the infix.
220-- /Unimplemented/ - Help wanted.
221{-# INLINE dropInfix #-}
223 -- (Eq a, IsStream t, Monad m) =>
224 t m a -> t m a -> t m a
225dropInfix = error "Not implemented yet!"
227-- | Drop suffix from the input stream if present. Suffix stream may be
228-- consumed multiple times.
230-- Space: @O(n)@ where n is the length of the suffix.
232-- /Unimplemented/ - Help wanted.
233{-# INLINE dropSuffix #-}
235 -- (Eq a, IsStream t, Monad m) =>
236 t m a -> t m a -> t m a
237dropSuffix = error "Not implemented yet!"
239------------------------------------------------------------------------------
241------------------------------------------------------------------------------
243-- Splitting operations that take a predicate and a Fold can be
244-- expressed using parseMany. Operations like chunksOf, intervalsOf, split*,
245-- can be expressed using parseMany when used with an appropriate Parser.
247-- XXX We need takeGE/takeBetween to implement "some" using "many".
249-- | Like 'foldMany' but appends empty fold output if the fold and stream
250-- termination aligns:
252-- >>> f = Fold.take 2 Fold.sum
253-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList []
255-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..9]
257-- >>> Stream.toList $ Stream.foldManyPost f $ Stream.fromList [1..10]
262{-# INLINE foldManyPost #-}
264 :: (IsStream t, Monad m)
268foldManyPost f m = D.fromStreamD $ D.foldManyPost f (D.toStreamD m)
270-- | Apply a 'Fold' repeatedly on a stream and emit the fold outputs in the
273-- To sum every two contiguous elements in a stream:
275-- >>> f = Fold.take 2 Fold.sum
276-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList [1..10]
279-- On an empty stream the output is empty:
281-- >>> Stream.toList $ Stream.foldMany f $ Stream.fromList []
284-- Note @Stream.foldMany (Fold.take 0)@ would result in an infinite loop in a
289{-# INLINE foldMany #-}
291 :: (IsStream t, Monad m)
295foldMany f m = D.fromStreamD $ D.foldMany f (D.toStreamD m)
297-- | Apply a stream of folds to an input stream and emit the results in the
302{-# INLINE foldSequence #-}
304 :: -- (IsStream t, Monad m) =>
308foldSequence _f _m = undefined
310-- | Iterate a fold generator on a stream. The initial value @b@ is used to
311-- generate the first fold, the fold is applied on the stream and the result of
312-- the fold is used to generate the next fold and so on.
315-- >>> import Data.Monoid (Sum(..))
316-- >>> f x = return (Fold.take 2 (Fold.sconcat x))
317-- >>> s = Stream.map Sum $ Stream.fromList [1..10]
318-- >>> Stream.toList $ Stream.map getSum $ Stream.foldIterateM f 0 s
323-- This is the streaming equivalent of monad like sequenced application of
324-- folds where next fold is dependent on the previous fold.
328{-# INLINE foldIterateM #-}
330 (IsStream t, Monad m) => (b -> m (Fold m a b)) -> b -> t m a -> t m b
331foldIterateM f i m = D.fromStreamD $ D.foldIterateM f i (D.toStreamD m)
333------------------------------------------------------------------------------
335------------------------------------------------------------------------------
337-- | Apply a 'Parser' repeatedly on a stream and emit the parsed values in the
340-- This is the streaming equivalent of the 'Streamly.Internal.Data.Parser.many'
343-- >>> Stream.toList $ Stream.parseMany (Parser.takeBetween 0 2 Fold.sum) $ Stream.fromList [1..10]
347-- > Stream.toList $ Stream.parseMany (Parser.line Fold.toList) $ Stream.fromList "hello\\nworld"
348-- ["hello\\n","world"]
353-- foldMany f = parseMany (fromFold f)
356-- Known Issues: When the parser fails there is no way to get the remaining
361{-# INLINE parseMany #-}
363 :: (IsStream t, MonadThrow m)
368 D.fromStreamD $ D.parseMany (PRK.fromParserK p) (D.toStreamD m)
370{-# INLINE parseManyD #-}
372 :: (IsStream t, MonadThrow m)
377 D.fromStreamD $ D.parseMany p (D.toStreamD m)
379-- | Apply a stream of parsers to an input stream and emit the results in the
384{-# INLINE parseSequence #-}
386 :: -- (IsStream t, Monad m) =>
390parseSequence _f _m = undefined
392-- XXX Change the parser arguments' order
394-- | @parseManyTill collect test stream@ tries the parser @test@ on the input,
395-- if @test@ fails it backtracks and tries @collect@, after @collect@ succeeds
396-- @test@ is tried again and so on. The parser stops when @test@ succeeds. The
397-- output of @test@ is discarded and the output of @collect@ is emitted in the
398-- output stream. The parser fails if @collect@ fails.
402{-# INLINE parseManyTill #-}
404 -- (IsStream t, MonadThrow m) =>
409parseManyTill = undefined
411-- | Iterate a parser generating function on a stream. The initial value @b@ is
412-- used to generate the first parser, the parser is applied on the stream and
413-- the result is used to generate the next parser and so on.
415-- >>> import Data.Monoid (Sum(..))
416-- >>> Stream.toList $ Stream.map getSum $ Stream.parseIterate (\b -> Parser.takeBetween 0 2 (Fold.sconcat b)) 0 $ Stream.map Sum $ Stream.fromList [1..10]
419-- This is the streaming equivalent of monad like sequenced application of
420-- parsers where next parser is dependent on the previous parser.
424{-# INLINE parseIterate #-}
426 :: (IsStream t, MonadThrow m)
427 => (b -> Parser m a b)
431parseIterate f i m = D.fromStreamD $
432 D.parseIterate (PRK.fromParserK . f) i (D.toStreamD m)
434------------------------------------------------------------------------------
435-- Generalized grouping
436------------------------------------------------------------------------------
438-- This combinator is the most general grouping combinator and can be used to
439-- implement all other grouping combinators.
441-- XXX check if this can implement the splitOn combinator i.e. we can slide in
442-- new elements, slide out old elements and incrementally compute the hash.
443-- Also, can we implement the windowed classification combinators using this?
445-- In fact this is a parse. Instead of using a special return value in the fold
446-- we are using a mapping function.
448-- Note that 'scanl'' (usually followed by a map to extract the desired value
449-- from the accumulator) can be used to realize many implementations e.g. a
450-- sliding window implementation. A scan followed by a mapMaybe is also a good
451-- pattern to express many problems where we want to emit a filtered output and
452-- not emit an output on every input.
454-- Passing on of the initial accumulator value to the next fold is equivalent
455-- to returning the leftover concept.
458-- | @groupScan splitter fold stream@ folds the input stream using @fold@.
459-- @splitter@ is applied on the accumulator of the fold every time an item is
460-- consumed by the fold. The fold continues until @splitter@ returns a 'Just'
461-- value. A 'Just' result from the @splitter@ specifies a result to be emitted
462-- in the output stream and the initial value of the accumulator for the next
463-- group's fold. This allows us to control whether to start fresh for the next
464-- fold or to continue from the previous fold's output.
466{-# INLINE groupScan #-}
468 :: (IsStream t, Monad m)
469 => (x -> m (Maybe (b, x))) -> Fold m a x -> t m a -> t m b
470groupScan split fold m = undefined
473------------------------------------------------------------------------------
475------------------------------------------------------------------------------
477-- | @groupsBy cmp f $ S.fromList [a,b,c,...]@ assigns the element @a@ to the
478-- first group, if @b \`cmp` a@ is 'True' then @b@ is also assigned to the same
479-- group. If @c \`cmp` a@ is 'True' then @c@ is also assigned to the same
480-- group and so on. When the comparison fails a new group is started. Each
481-- group is folded using the fold @f@ and the result of the fold is emitted in
484-- >>> Stream.toList $ Stream.groupsBy (>) Fold.toList $ Stream.fromList [1,3,7,0,2,5]
488{-# INLINE groupsBy #-}
490 :: (IsStream t, Monad m)
495groupsBy cmp f m = D.fromStreamD $ D.groupsBy cmp f (D.toStreamD m)
497-- | Unlike @groupsBy@ this function performs a rolling comparison of two
498-- successive elements in the input stream. @groupsByRolling cmp f $ S.fromList
499-- [a,b,c,...]@ assigns the element @a@ to the first group, if @a \`cmp` b@ is
500-- 'True' then @b@ is also assigned to the same group. If @b \`cmp` c@ is
501-- 'True' then @c@ is also assigned to the same group and so on. When the
502-- comparison fails a new group is started. Each group is folded using the fold
505-- >>> Stream.toList $ Stream.groupsByRolling (\a b -> a + 1 == b) Fold.toList $ Stream.fromList [1,2,3,7,8,9]
509{-# INLINE groupsByRolling #-}
511 :: (IsStream t, Monad m)
516groupsByRolling cmp f m = D.fromStreamD $ D.groupsRollingBy cmp f (D.toStreamD m)
519-- > groups = groupsBy (==)
520-- > groups = groupsByRolling (==)
522-- Groups contiguous spans of equal elements together in individual groups.
524-- >>> Stream.toList $ Stream.groups Fold.toList $ Stream.fromList [1,1,2,2]
529groups :: (IsStream t, Monad m, Eq a) => Fold m a b -> t m a -> t m b
530groups = groupsBy (==)
532------------------------------------------------------------------------------
533-- Splitting - by a predicate
534------------------------------------------------------------------------------
536-- In general we can use deintercalate for splitting. Then we can also use
537-- uniqBy to condense the separators. One way to generalize splitting is to
540-- data Segment a b = Empty | Segment b | Separator a
542-- XXX splitOn and splitOnSuffix have a different behavior on an empty stream,
545-- | Split on an infixed separator element, dropping the separator. The
546-- supplied 'Fold' is applied on the split segments. Splits the stream on
547-- separator elements determined by the supplied predicate, separator is
548-- considered as infixed between two segments:
550-- >>> splitOn' p xs = Stream.toList $ Stream.splitOn p Fold.toList (Stream.fromList xs)
551-- >>> splitOn' (== '.') "a.b"
554-- An empty stream is folded to the default value of the fold:
556-- >>> splitOn' (== '.') ""
559-- If one or both sides of the separator are missing then the empty segment on
560-- that side is folded to the default output of the fold:
562-- >>> splitOn' (== '.') "."
565-- >>> splitOn' (== '.') ".a"
568-- >>> splitOn' (== '.') "a."
571-- >>> splitOn' (== '.') "a..b"
574-- splitOn is an inverse of intercalating single element:
576-- > Stream.intercalate (Stream.fromPure '.') Unfold.fromList . Stream.splitOn (== '.') Fold.toList === id
578-- Assuming the input stream does not contain the separator:
580-- > Stream.splitOn (== '.') Fold.toList . Stream.intercalate (Stream.fromPure '.') Unfold.fromList === id
584{-# INLINE splitOn #-}
586 :: (IsStream t, Monad m)
587 => (a -> Bool) -> Fold m a b -> t m a -> t m b
589 -- We can express the infix splitting in terms of optional suffix split
590 -- fold. After applying a suffix split fold repeatedly if the last segment
591 -- ends with a suffix then we need to return the default output of the fold
592 -- after that to make it an infix split.
594 -- Alternately, we can also express it using an optional prefix split fold.
595 -- If the first segment starts with a prefix then we need to emit the
596 -- default output of the fold before that to make it an infix split, and
597 -- then apply prefix split fold repeatedly.
599 -- Since a suffix split fold can be easily expressed using a
600 -- non-backtracking fold, we use that.
601 foldManyPost (FL.takeEndBy_ predicate f)
603-- | Split on a suffixed separator element, dropping the separator. The
604-- supplied 'Fold' is applied on the split segments.
606-- >>> splitOnSuffix' p xs = Stream.toList $ Stream.splitOnSuffix p Fold.toList (Stream.fromList xs)
607-- >>> splitOnSuffix' (== '.') "a.b."
610-- >>> splitOnSuffix' (== '.') "a."
613-- An empty stream results in an empty output stream:
615-- >>> splitOnSuffix' (== '.') ""
618-- An empty segment consisting of only a suffix is folded to the default output
621-- >>> splitOnSuffix' (== '.') "."
624-- >>> splitOnSuffix' (== '.') "a..b.."
627-- A suffix is optional at the end of the stream:
629-- >>> splitOnSuffix' (== '.') "a"
632-- >>> splitOnSuffix' (== '.') ".a"
635-- >>> splitOnSuffix' (== '.') "a.b"
638-- > lines = splitOnSuffix (== '\n')
640-- 'splitOnSuffix' is an inverse of 'intercalateSuffix' with a single element:
642-- > Stream.intercalateSuffix (Stream.fromPure '.') Unfold.fromList . Stream.splitOnSuffix (== '.') Fold.toList === id
644-- Assuming the input stream does not contain the separator:
646-- > Stream.splitOnSuffix (== '.') Fold.toList . Stream.intercalateSuffix (Stream.fromPure '.') Unfold.fromList === id
650{-# INLINE splitOnSuffix #-}
652 :: (IsStream t, Monad m)
653 => (a -> Bool) -> Fold m a b -> t m a -> t m b
654splitOnSuffix predicate f = foldMany (FL.takeEndBy_ predicate f)
656-- | Split on a prefixed separator element, dropping the separator. The
657-- supplied 'Fold' is applied on the split segments.
660-- > splitOnPrefix' p xs = Stream.toList $ Stream.splitOnPrefix p (Fold.toList) (Stream.fromList xs)
661-- > splitOnPrefix' (== '.') ".a.b"
665-- An empty stream results in an empty output stream:
667-- > splitOnPrefix' (== '.') ""
671-- An empty segment consisting of only a prefix is folded to the default output
675-- > splitOnPrefix' (== '.') "."
678-- > splitOnPrefix' (== '.') ".a.b."
681-- > splitOnPrefix' (== '.') ".a..b"
686-- A prefix is optional at the beginning of the stream:
689-- > splitOnPrefix' (== '.') "a"
692-- > splitOnPrefix' (== '.') "a.b"
696-- 'splitOnPrefix' is an inverse of 'intercalatePrefix' with a single element:
698-- > Stream.intercalatePrefix (Stream.fromPure '.') Unfold.fromList . Stream.splitOnPrefix (== '.') Fold.toList === id
700-- Assuming the input stream does not contain the separator:
702-- > Stream.splitOnPrefix (== '.') Fold.toList . Stream.intercalatePrefix (Stream.fromPure '.') Unfold.fromList === id
706{-# INLINE splitOnPrefix #-}
707splitOnPrefix :: -- (IsStream t, MonadCatch m) =>
708 (a -> Bool) -> Fold m a b -> t m a -> t m b
709splitOnPrefix _predicate _f = undefined
710 -- parseMany (Parser.sliceBeginBy predicate f)
712-- | Like 'splitOn' after stripping leading, trailing, and repeated separators.
713-- Therefore, @".a..b."@ with '.' as the separator would be parsed as
714-- @["a","b"]@. In other words, its like parsing words from whitespace
717-- >>> wordsBy' p xs = Stream.toList $ Stream.wordsBy p Fold.toList (Stream.fromList xs)
719-- >>> wordsBy' (== ',') ""
722-- >>> wordsBy' (== ',') ","
725-- >>> wordsBy' (== ',') ",a,,b,"
728-- > words = wordsBy isSpace
732-- It is equivalent to splitting in any of the infix/prefix/suffix styles
733-- followed by removal of empty segments.
734{-# INLINE wordsBy #-}
736 :: (IsStream t, Monad m)
737 => (a -> Bool) -> Fold m a b -> t m a -> t m b
738wordsBy predicate f m =
739 D.fromStreamD $ D.wordsBy predicate f (D.toStreamD m)
741-- | Like 'splitOnSuffix' but keeps the suffix attached to the resulting
744-- >>> splitWithSuffix' p xs = Stream.toList $ splitWithSuffix p Fold.toList (Stream.fromList xs)
746-- >>> splitWithSuffix' (== '.') ""
749-- >>> splitWithSuffix' (== '.') "."
752-- >>> splitWithSuffix' (== '.') "a"
755-- >>> splitWithSuffix' (== '.') ".a"
758-- >>> splitWithSuffix' (== '.') "a."
761-- >>> splitWithSuffix' (== '.') "a.b"
764-- >>> splitWithSuffix' (== '.') "a.b."
767-- >>> splitWithSuffix' (== '.') "a..b.."
768-- ["a.",".","b.","."]
772{-# INLINE splitWithSuffix #-}
774 :: (IsStream t, Monad m)
775 => (a -> Bool) -> Fold m a b -> t m a -> t m b
776splitWithSuffix predicate f = foldMany (FL.takeEndBy predicate f)
778------------------------------------------------------------------------------
779-- Splitting - on a delimiter sequence
780------------------------------------------------------------------------------
782-- Int list examples for splitOn:
784-- >>> splitList [] [1,2,3,3,4]
785-- > [[1],[2],[3],[3],[4]]
787-- >>> splitList [5] [1,2,3,3,4]
790-- >>> splitList [1] [1,2,3,3,4]
793-- >>> splitList [4] [1,2,3,3,4]
796-- >>> splitList [2] [1,2,3,3,4]
799-- >>> splitList [3] [1,2,3,3,4]
802-- >>> splitList [3,3] [1,2,3,3,4]
805-- >>> splitList [1,2,3,3,4] [1,2,3,3,4]
809-- This can be implemented easily using Rabin Karp
810-- | Split on any one of the given patterns.
811{-# INLINE splitOnAny #-}
813 :: (IsStream t, Monad m, Storable a, Integral a)
814 => [Array a] -> Fold m a b -> t m a -> t m b
815splitOnAny subseq f m = undefined -- D.fromStreamD $ D.splitOnAny f subseq (D.toStreamD m)
818-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
819-- XXX Use two folds, one ring buffer fold for separator sequence and the other
820-- split consumer fold. The input is fed to the ring fold first and the
821-- rejected input is fed to the split fold. If the separator matches, the ring
822-- fold would consume all.
824-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
826-- >>> splitOn'_ pat xs = Stream.toList $ Stream.splitBySeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
828-- >>> splitOn'_ "" "hello"
829-- ["h","","e","","l","","l","","o"]
831-- >>> splitOn'_ "hello" ""
834-- >>> splitOn'_ "hello" "hello"
837-- >>> splitOn'_ "x" "hello"
840-- >>> splitOn'_ "h" "hello"
843-- >>> splitOn'_ "o" "hello"
846-- >>> splitOn'_ "e" "hello"
849-- >>> splitOn'_ "l" "hello"
850-- ["he","l","","l","o"]
852-- >>> splitOn'_ "ll" "hello"
856{-# INLINE splitBySeq #-}
858 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
859 => Array a -> Fold m a b -> t m a -> t m b
861 intersperseM (fold f (A.toStream patt)) $ splitOnSeq patt f m
863-- | Like 'splitSuffixBy' but the separator is a sequence of elements, instead
864-- of a predicate for a single element.
866-- >>> splitOnSuffixSeq_ pat xs = Stream.toList $ Stream.splitOnSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
868-- >>> splitOnSuffixSeq_ "." ""
871-- >>> splitOnSuffixSeq_ "." "."
874-- >>> splitOnSuffixSeq_ "." "a"
877-- >>> splitOnSuffixSeq_ "." ".a"
880-- >>> splitOnSuffixSeq_ "." "a."
883-- >>> splitOnSuffixSeq_ "." "a.b"
886-- >>> splitOnSuffixSeq_ "." "a.b."
889-- >>> splitOnSuffixSeq_ "." "a..b.."
892-- > lines = splitOnSuffixSeq "\n"
894-- 'splitOnSuffixSeq' is an inverse of 'intercalateSuffix'. The following law
897-- > intercalateSuffix . splitOnSuffixSeq == id
899-- The following law holds when the separator is non-empty and contains none of
900-- the elements present in the input lists:
902-- > splitSuffixOn . intercalateSuffix == id
905{-# INLINE splitOnSuffixSeq #-}
907 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
908 => Array a -> Fold m a b -> t m a -> t m b
909splitOnSuffixSeq patt f m =
910 D.fromStreamD $ D.splitOnSuffixSeq False patt f (D.toStreamD m)
913-- | Like 'splitOn' but drops any empty splits.
915{-# INLINE wordsOn #-}
917 :: (IsStream t, Monad m, Storable a, Eq a)
918 => Array a -> Fold m a b -> t m a -> t m b
919wordsOn subseq f m = undefined -- D.fromStreamD $ D.wordsOn f subseq (D.toStreamD m)
922-- | Like 'splitOnSuffixSeq' but keeps the suffix intact in the splits.
924-- >>> splitWithSuffixSeq' pat xs = Stream.toList $ Stream.splitWithSuffixSeq (Array.fromList pat) Fold.toList (Stream.fromList xs)
926-- >>> splitWithSuffixSeq' "." ""
929-- >>> splitWithSuffixSeq' "." "."
932-- >>> splitWithSuffixSeq' "." "a"
935-- >>> splitWithSuffixSeq' "." ".a"
938-- >>> splitWithSuffixSeq' "." "a."
941-- >>> splitWithSuffixSeq' "." "a.b"
944-- >>> splitWithSuffixSeq' "." "a.b."
947-- >>> splitWithSuffixSeq' "." "a..b.."
948-- ["a.",".","b.","."]
951{-# INLINE splitWithSuffixSeq #-}
953 :: (IsStream t, MonadIO m, Storable a, Enum a, Eq a)
954 => Array a -> Fold m a b -> t m a -> t m b
955splitWithSuffixSeq patt f m =
956 D.fromStreamD $ D.splitOnSuffixSeq True patt f (D.toStreamD m)
959-- This can be implemented easily using Rabin Karp
960-- | Split post any one of the given patterns.
961{-# INLINE splitOnSuffixSeqAny #-}
963 :: (IsStream t, Monad m, Storable a, Integral a)
964 => [Array a] -> Fold m a b -> t m a -> t m b
965splitOnSuffixSeqAny subseq f m = undefined
966 -- D.fromStreamD $ D.splitPostAny f subseq (D.toStreamD m)
969------------------------------------------------------------------------------
971------------------------------------------------------------------------------
973-- | Group the input stream into groups of @n@ elements each and then fold each
974-- group using the provided fold function.
976-- >>> Stream.toList $ Stream.chunksOf 2 Fold.sum (Stream.enumerateFromTo 1 10)
979-- This can be considered as an n-fold version of 'take' where we apply
980-- 'take' repeatedly on the leftover stream until the stream exhausts.
982-- @chunksOf n f = foldMany (FL.take n f)@
985{-# INLINE chunksOf #-}
987 :: (IsStream t, Monad m)
988 => Int -> Fold m a b -> t m a -> t m b
989chunksOf n f = D.fromStreamD . D.chunksOf n f . D.toStreamD
994{-# INLINE chunksOf2 #-}
996 :: (IsStream t, Monad m)
997 => Int -> m c -> Fold2 m c a b -> t m a -> t m b
998chunksOf2 n action f m = D.fromStreamD $ D.groupsOf2 n action f (D.toStreamD m)
1000-- | @arraysOf n stream@ groups the elements in the input stream into arrays of
1001-- @n@ elements each.
1003-- Same as the following but may be more efficient:
1005-- > arraysOf n = Stream.foldMany (A.writeN n)
1008{-# INLINE arraysOf #-}
1009arraysOf :: (IsStream t, MonadIO m, Storable a)
1010 => Int -> t m a -> t m (Array a)
1011arraysOf n = D.fromStreamD . A.arraysOf n . D.toStreamD
1013-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
1014-- XXX add this example after fixing the serial stream rate control
1016-- | Group the input stream into windows of @n@ second each and then fold each
1017-- group using the provided fold function.
1019-- >>> Stream.toList $ Stream.take 5 $ Stream.intervalsOf 1 Fold.sum $ Stream.constRate 2 $ Stream.enumerateFrom 1
1020-- [...,...,...,...,...]
1023{-# INLINE intervalsOf #-}
1025 :: (IsStream t, MonadAsync m)
1026 => Double -> Fold m a b -> t m a -> t m b
1028 splitWithSuffix isNothing (FL.catMaybes f)
1029 (interjectSuffix n (return Nothing) (Serial.map Just xs))
1031------------------------------------------------------------------------------
1032-- Windowed classification
1033------------------------------------------------------------------------------
1035-- We divide the stream into windows or chunks in space or time and each window
1036-- can be associated with a key, all events associated with a particular key in
1037-- the window can be folded to a single result. The stream can be split into
1038-- windows by size or by using a split predicate on the elements in the stream.
1039-- For example, when we receive a closing flag, we can close the window.
1041-- A "chunk" is a space window and a "session" is a time window. Are there any
1042-- other better short words to describe them. An alternative is to use
1043-- "swindow" and "twindow". Another word for "session" could be "spell".
1045-- TODO: To mark the position in space or time we can have Indexed or
1046-- TimeStamped types. That can make it easy to deal with the position indices
1049------------------------------------------------------------------------------
1050-- Keyed Sliding Windows
1051------------------------------------------------------------------------------
1054{-# INLINABLE classifySlidingChunks #-}
1055classifySlidingChunks
1056 :: (IsStream t, MonadAsync m, Ord k)
1057 => Int -- ^ window size
1058 -> Int -- ^ window slide
1059 -> Fold m a b -- ^ Fold to be applied to window events
1060 -> t m (k, a, Bool) -- ^ window key, data, close event
1062classifySlidingChunks wsize wslide (Fold step initial extract) str
1065-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1066-- slide the send window when an ack is received and we slide the receive
1067-- window when a sequence is complete. Sliding is stateful in case of TCP,
1068-- sliding releases the send buffer or makes data available to the user from
1069-- the receive buffer.
1070{-# INLINABLE classifySlidingSessions #-}
1071classifySlidingSessions
1072 :: (IsStream t, MonadAsync m, Ord k)
1073 => Double -- ^ timer tick in seconds
1074 -> Double -- ^ time window size
1075 -> Double -- ^ window slide
1076 -> Fold m a b -- ^ Fold to be applied to window events
1077 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1079classifySlidingSessions tick interval slide (Fold step initial extract) str
1083------------------------------------------------------------------------------
1084-- Sliding Window Buffers
1085------------------------------------------------------------------------------
1087-- These buffered versions could be faster than concurrent incremental folds of
1088-- all overlapping windows as in many cases we may not need all the values to
1089-- compute the fold, we can just compute the result using the old value and new
1090-- value. However, we may need the buffer once in a while, for example for
1091-- string search we usually compute the hash incrementally but when the hash
1092-- matches the hash of the pattern we need to compare the whole string.
1094-- XXX we should be able to implement sequence based splitting combinators
1095-- using this combinator.
1098-- | Buffer n elements of the input in a ring buffer. When t new elements are
1099-- collected, slide the window to remove the same number of oldest elements,
1100-- insert the new elements, and apply an incremental fold on the sliding
1101-- window, supplying the outgoing elements, the new ring buffer as arguments.
1103 :: (IsStream t, Monad m, Ord a, Storable a)
1104 => Int -- window size
1105 -> Int -- window slide
1106 -> Fold m (Ring a, Array a) b
1109slidingChunkBuffer = undefined
1111-- Buffer n seconds worth of stream elements of the input in a radix tree.
1112-- Every t seconds, remove the items that are older than n seconds, and apply
1113-- an incremental fold on the sliding window, supplying the outgoing elements,
1114-- and the new radix tree buffer as arguments.
1116 :: (IsStream t, Monad m, Ord a, Storable a)
1117 => Int -- window size
1119 -> Fold m (RTree a, Array a) b
1122slidingSessionBuffer = undefined
1125------------------------------------------------------------------------------
1126-- Keyed Session Windows
1127------------------------------------------------------------------------------
1130-- | Keyed variable size space windows. Close the window if we do not receive a
1131-- window event in the next "spaceout" elements.
1132{-# INLINABLE classifyChunksBy #-}
1134 :: (IsStream t, MonadAsync m, Ord k)
1135 => Int -- ^ window spaceout (spread)
1136 -> Bool -- ^ reset the spaceout when a chunk window element is received
1137 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1138 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1140classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1142-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1143-- received within the chunk size window. The chunk gets closed only if no
1144-- element is received within the chunk window.
1146{-# INLINABLE classifyKeepAliveChunks #-}
1147classifyKeepAliveChunks
1148 :: (IsStream t, MonadAsync m, Ord k)
1149 => Int -- ^ window spaceout (spread)
1150 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1151 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1153classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1156data SessionState t m k a b = SessionState
1157 { sessionCurTime :: !AbsTime -- ^ time since last event
1158 , sessionEventTime :: !AbsTime -- ^ time as per last event
1159 , sessionCount :: !Int -- ^ total number sessions in progress
1160 , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts
1161 , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys
1162 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
1165-- | @classifySessionsBy tick keepalive predicate timeout fold stream@
1166-- classifies an input event @stream@ consisting of @(timestamp, (key,
1167-- value))@ into sessions based on the @key@, folding all the values
1168-- corresponding to the same key into a session using the supplied @fold@.
1170-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the
1171-- session key and the folded value is emitted in the output stream. The
1172-- timeout is measured from the first event in the session. If the @keepalive@
1173-- option is set to 'True' the timeout is reset to 0 whenever an event is
1176-- The @timestamp@ in the input stream is an absolute time from some epoch,
1177-- characterizing the time when the input event was generated. The notion of
1178-- current time is maintained by a monotonic event time clock using the
1179-- timestamps seen in the input stream. The latest timestamp seen till now is
1180-- used as the base for the current time. When no new events are seen, a timer
1181-- is started with a clock resolution of @tick@ seconds. This timer is used to
1182-- detect session timeouts in the absence of new events.
1184-- To ensure an upper bound on the memory used the number of sessions can be
1185-- limited to an upper bound. If the ejection @predicate@ returns 'True', the
1186-- oldest session is ejected before inserting a new session.
1189-- Stream.mapM_ print
1190-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)
1191-- $ Stream.timestamped
1192-- $ Stream.delay 0.1
1193-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']
1201{-# INLINABLE classifySessionsBy #-}
1203 :: (IsStream t, MonadAsync m, Ord k)
1204 => Double -- ^ timer tick in seconds
1205 -> Bool -- ^ reset the timer when an event is received
1206 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1207 -> Double -- ^ session timeout in seconds
1208 -> Fold m a b -- ^ Fold to be applied to session data
1209 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1210 -> t m (k, b) -- ^ session key, fold result
1211classifySessionsBy tick reset ejectPred tmout
1212 (Fold step initial extract) str =
1213 concatMap sessionOutputStream $
1214 scanlMAfter' sstep (return szero) flush stream
1218 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
1219 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1220 szero = SessionState
1221 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
1222 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
1224 , sessionTimerHeap = H.empty
1225 , sessionKeyValueMap = Map.empty
1226 , sessionOutputStream = K.nil
1229 -- We can eject sessions based on the current session count to limit
1230 -- memory consumption. There are two possible strategies:
1232 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
1233 -- threshold even before timeout, effectively reduce the timeout.
1234 -- 2) Drop creation of new sessions but keep accepting new events for the
1237 -- We use the first strategy as of now.
1239 -- Got a new stream input element
1240 sstep session@SessionState{..} (Just (timestamp, (key, value))) = do
1241 -- XXX we should use a heap in pinned memory to scale it to a large
1244 -- To detect session inactivity we keep a timestamp of the latest event
1245 -- in the Map along with the fold result. When we purge the session
1246 -- from the heap we match the timestamp in the heap with the timestamp
1247 -- in the Map, if the latest timestamp is newer and has not expired we
1248 -- reinsert the key in the heap.
1250 -- XXX if the key is an Int, we can also use an IntMap for slightly
1251 -- better performance.
1253 let curTime = max sessionEventTime timestamp
1254 mOld = Map.lookup key sessionKeyValueMap
1256 -- deleting a key from the heap is expensive, so we never
1257 -- delete a key from heap, we just purge it from the Map and it
1258 -- gets purged from the heap on timeout. We just need an extra
1259 -- lookup in the Map when the key is purged from the heap, that
1260 -- should not be expensive.
1262 let (mp, cnt) = case mOld of
1263 Nothing -> (sessionKeyValueMap, sessionCount)
1264 Just _ -> (Map.delete key sessionKeyValueMap
1267 { sessionCurTime = curTime
1268 , sessionEventTime = curTime
1269 , sessionCount = cnt
1270 , sessionKeyValueMap = mp
1271 , sessionOutputStream = fromPure (key, fb)
1274 let acc = Tuple' timestamp fs1
1275 (hp1, mp1, out1, cnt1) <- do
1276 let vars = (sessionTimerHeap, sessionKeyValueMap,
1277 K.nil, sessionCount)
1279 -- inserting new entry
1281 -- Eject a session from heap and map is needed
1282 eject <- ejectPred sessionCount
1283 (hp, mp, out, cnt) <-
1288 -- Insert the new session in heap
1289 let expiry = addToAbsTime timestamp timeoutMs
1290 hp' = H.insert (Entry expiry key) hp
1291 in return (hp', mp, out, cnt + 1)
1292 -- updating old entry
1293 Just _ -> return vars
1295 let mp2 = Map.insert key acc mp1
1296 return $ SessionState
1297 { sessionCurTime = curTime
1298 , sessionEventTime = curTime
1299 , sessionCount = cnt1
1300 , sessionTimerHeap = hp1
1301 , sessionKeyValueMap = mp2
1302 , sessionOutputStream = out1
1307 Just (Tuple' _ acc) -> return $ FL.Partial acc
1309 FL.Done fb -> done fb
1311 res <- step fs value
1313 FL.Done fb -> done fb
1314 FL.Partial fs1 -> partial fs1
1316 -- Got a timer tick event
1317 sstep sessionState@SessionState{..} Nothing =
1318 let curTime = addToAbsTime sessionCurTime tickMs
1319 in ejectExpired sessionState curTime
1321 flush session@SessionState{..} = do
1322 (hp', mp', out, count) <-
1325 , sessionKeyValueMap
1330 { sessionCount = count
1331 , sessionTimerHeap = hp'
1332 , sessionKeyValueMap = mp'
1333 , sessionOutputStream = out
1336 -- delete from map and output the fold accumulator
1337 ejectEntry hp mp out cnt acc key = do
1339 let out1 = (key, sess) `K.cons` out
1340 let mp1 = Map.delete key mp
1341 return (hp, mp1, out1, cnt - 1)
1343 ejectAll (hp, mp, out, !cnt) = do
1344 let hres = H.uncons hp
1346 Just (Entry _ key, hp1) -> do
1347 r <- case Map.lookup key mp of
1348 Nothing -> return (hp1, mp, out, cnt)
1349 Just (Tuple' _ acc) -> ejectEntry hp1 mp out cnt acc key
1352 assert (Map.null mp) (return ())
1353 return (hp, mp, out, cnt)
1355 ejectOne (hp, mp, out, !cnt) = do
1356 let hres = H.uncons hp
1358 Just (Entry expiry key, hp1) ->
1359 case Map.lookup key mp of
1360 Nothing -> ejectOne (hp1, mp, out, cnt)
1361 Just (Tuple' latestTS acc) -> do
1362 let expiry1 = addToAbsTime latestTS timeoutMs
1363 if not reset || expiry1 <= expiry
1364 then ejectEntry hp1 mp out cnt acc key
1366 -- reset the session timeout and continue
1367 let hp2 = H.insert (Entry expiry1 key) hp1
1368 in ejectOne (hp2, mp, out, cnt)
1370 assert (Map.null mp) (return ())
1371 return (hp, mp, out, cnt)
1373 ejectExpired session@SessionState{..} curTime = do
1374 (hp', mp', out, count) <-
1375 ejectLoop sessionTimerHeap sessionKeyValueMap K.nil sessionCount
1377 { sessionCurTime = curTime
1378 , sessionCount = count
1379 , sessionTimerHeap = hp'
1380 , sessionKeyValueMap = mp'
1381 , sessionOutputStream = out
1386 ejectLoop hp mp out !cnt = do
1387 let hres = H.uncons hp
1389 Just (Entry expiry key, hp1) -> do
1391 if curTime >= expiry
1392 then return (True, False)
1398 case Map.lookup key mp of
1399 Nothing -> ejectLoop hp1 mp out cnt
1400 Just (Tuple' latestTS acc) -> do
1401 let expiry1 = addToAbsTime latestTS timeoutMs
1402 if expiry1 <= curTime || not reset || force
1404 (hp2,mp1,out1,cnt1) <-
1405 ejectEntry hp1 mp out cnt acc key
1406 ejectLoop hp2 mp1 out1 cnt1
1408 -- reset the session timeout and continue
1409 let hp2 = H.insert (Entry expiry1 key) hp1
1410 in ejectLoop hp2 mp out cnt
1411 else return (hp, mp, out, cnt)
1413 assert (Map.null mp) (return ())
1414 return (hp, mp, out, cnt)
1416 -- merge timer events in the stream
1417 stream = Serial.map Just str `Par.parallelFst` repeatM timer
1419 liftIO $ threadDelay (round $ tick * 1000000)
1422-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1423-- option set to 'True'.
1426-- classifyKeepAliveSessions = classifySessionsBy 1 True
1431{-# INLINABLE classifyKeepAliveSessions #-}
1432classifyKeepAliveSessions ::
1433 (IsStream t, MonadAsync m, Ord k)
1434 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
1435 -> Double -- ^ session inactive timeout
1436 -> Fold m a b -- ^ Fold to be applied to session payload data
1437 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1439classifyKeepAliveSessions = classifySessionsBy 1 True
1441------------------------------------------------------------------------------
1442-- Keyed tumbling windows
1443------------------------------------------------------------------------------
1445-- Tumbling windows is a special case of sliding windows where the window slide
1446-- is the same as the window size. Or it can be a special case of session
1447-- windows where the reset flag is set to False.
1449-- XXX instead of using the early termination flag in the stream, we can use an
1450-- early terminating fold instead.
1453-- | Split the stream into fixed size chunks of specified size. Within each
1454-- such chunk fold the elements in buckets identified by the keys. A particular
1455-- bucket fold can be terminated early if a closing flag is encountered in an
1456-- element for that key.
1459{-# INLINABLE classifyChunksOf #-}
1461 :: (IsStream t, MonadAsync m, Ord k)
1462 => Int -- ^ window size
1463 -> Fold m a b -- ^ Fold to be applied to window events
1464 -> t m (k, a, Bool) -- ^ window key, data, close event
1466classifyChunksOf wsize = classifyChunksBy wsize False
1469-- | Same as 'classifySessionsBy' with a timer tick of 1 second and keepalive
1470-- option set to 'False'.
1473-- classifySessionsOf = classifySessionsBy 1 False
1478{-# INLINABLE classifySessionsOf #-}
1479classifySessionsOf ::
1480 (IsStream t, MonadAsync m, Ord k)
1481 => (Int -> m Bool) -- ^ predicate to eject sessions on session count
1482 -> Double -- ^ time window size
1483 -> Fold m a b -- ^ Fold to be applied to session data
1484 -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)
1486classifySessionsOf = classifySessionsBy 1 False
1488------------------------------------------------------------------------------
1490------------------------------------------------------------------------------
1492-- | @splitInnerBy splitter joiner stream@ splits the inner containers @f a@ of
1493-- an input stream @t m (f a)@ using the @splitter@ function. Container
1494-- elements @f a@ are collected until a split occurs, then all the elements
1495-- before the split are joined using the @joiner@ function.
1497-- For example, if we have a stream of @Array Word8@, we may want to split the
1498-- stream into arrays representing lines separated by '\n' byte such that the
1499-- resulting stream after a split would be one array for each line.
1501-- CAUTION! This is not a true streaming function as the container size after
1502-- the split and merge may not be bounded.
1505{-# INLINE splitInnerBy #-}
1507 :: (IsStream t, Monad m)
1508 => (f a -> m (f a, Maybe (f a))) -- splitter
1509 -> (f a -> f a -> m (f a)) -- joiner
1512splitInnerBy splitter joiner xs =
1513 D.fromStreamD $ D.splitInnerBy splitter joiner $ D.toStreamD xs
1515-- | Like 'splitInnerBy' but splits assuming the separator joins the segment in
1519{-# INLINE splitInnerBySuffix #-}
1521 :: (IsStream t, Monad m, Eq (f a), Monoid (f a))
1522 => (f a -> m (f a, Maybe (f a))) -- splitter
1523 -> (f a -> f a -> m (f a)) -- joiner
1526splitInnerBySuffix splitter joiner xs =
1527 D.fromStreamD $ D.splitInnerBySuffix splitter joiner $ D.toStreamD xs
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list