1{-# OPTIONS_GHC -Wno-orphans #-}
6-- Module : Streamly.Internal.Data.Stream.IsStream
7-- Copyright : (c) 2017 Composewell Technologies
10-- Maintainer : streamly@composewell.com
11-- Stability : experimental
14-- This is an Internal module consisting of released, unreleased and
15-- unimplemented APIs. For stable and released APIs please see
16-- "Streamly.Prelude" module. This module provides documentation only for the
17-- unreleased and unimplemented APIs. For documentation on released APIs please
18-- see "Streamly.Prelude" module.
20module Streamly.Internal.Data.Stream.IsStream
29 -- ** Speculative Streams
33 -- ** Asynchronous Streams
48 -- * Stream Type Adapters
87 -- ** Time Enumeration
108 -- ** Cyclic Elements
111 -- ** From Containers
138 -- ** Composable Left Folds
139 -- | See "Streamly.Internal.Data.Fold"
146 -- | See "Streamly.Internal.Data.Parser"
156 -- ** Concurrent Folds
163 -- -- ** To Summary (Full Folds)
171 -- -- ** To Summary (Maybe) (Full Folds)
179 -- -- ** To Containers (Full Folds)
185 -- ** Composable Left Folds
187 , toStream -- XXX rename to write?
188 , toStreamRev -- XXX rename to writeRev?
192 -- -- ** To Elements (Partial Folds)
196 -- -- | Folds that extract selected elements of a stream or their properties.
206 -- -- ** To Boolean (Partial Folds)
215 -- ** Multi-Stream folds
220 -- finding subsequences
226 -- trimming sequences
266 -- ** Scan Using Fold
270 -- XXX Once we have pipes the contravariant transformations can be
271 -- represented by attaching pipes before a transformation.
283 -- ** Concurrent Transformation
295 -- ** Deleting Elements
298 -- , uniqBy -- by predicate e.g. to remove duplicate "/" in a path
299 -- , uniqOn -- to remove duplicate sequences
300 -- , pruneBy -- dropAround + uniqBy - like words
302 -- ** Inserting Elements
311 , intersperseSuffixBySpan
312 -- , intersperseBySpan
313 -- , intersperseByIndices -- using an index function/stream
315 -- time domain intersperse
316 -- , intersperseByTime
317 -- , intersperseByEvent
328 -- , timestampedR -- timer
360 -- -- *** Searching Elements
364 -- -- *** Searching Sequences
365 -- , seqIndices -- search a sequence in the stream
367 -- -- *** Searching Multiple Sequences
368 -- , seqIndicesAny -- search any of the given sequence in the stream
370 -- -- -- ** Searching Streams
371 -- -- | Finding a stream within another stream.
374 -- | Streams can be sliced into segments in space or in time. We use the
375 -- term @chunk@ to refer to a spatial length of the stream (spatial window)
376 -- and the term @session@ to refer to a length in time (time window).
378 -- -- *** Using Element Separators
386 , wordsBy -- stripAndCompactBy
388 -- -- *** Splitting By Sequences
391 -- , splitOnPrefixSeq
393 -- Keeping the delimiters
396 -- , splitByPrefixSeq
399 -- Splitting using multiple sequence separators
401 -- , splitOnAnySuffixSeq
402 -- , splitOnAnyPrefixSeq
404 -- -- *** Splitting By Streams
405 -- -- | Splitting a stream using another stream as separator.
412 -- In imperative terms, grouped folding can be considered as a nested loop
413 -- where we loop over the stream to group elements and then loop over
414 -- individual groups to fold them to a single value that is yielded in the
427 -- * Windowed Classification
429 -- | Split the stream into windows or chunks in space or time. Each window
430 -- can be associated with a key, all events associated with a particular
431 -- key in the window can be folded to a single result. The stream is split
432 -- into windows of specified size, the window can be terminated early if
433 -- the closing flag is specified in the input stream.
435 -- The term "chunk" is used for a space window and the term "session" is
436 -- used for a time window.
438 -- ** Tumbling Windows
439 -- | A new window starts after the previous window is finished.
441 -- , classifyChunksOf
445 -- ** Keep Alive Windows
446 -- | The window size is extended if an event arrives within the specified
447 -- window size. This can represent sessions with idle or inactive timeout.
449 -- , classifyKeepAliveChunks
450 , classifyKeepAliveSessions
453 -- ** Sliding Windows
454 -- | A new window starts after the specified slide from the previous
455 -- window. Therefore windows can overlap.
456 , classifySlidingChunks
457 , classifySlidingSessions
459 -- ** Sliding Window Buffers
460 -- , slidingChunkBuffer
461 -- , slidingSessionBuffer
463 -- * Combining Streams
503 -- ** Flattening a Container of Streams
505 , concatMapFoldableWith
506 , concatForFoldableWith
508 -- ** Flattening a Stream of Streams
517 -- ** Flattening Using Unfolds
519 , concatUnfoldInterleave
520 , concatUnfoldRoundrobin
522 -- ** Flattening a Tree of Streams
525 -- ** Flattening a Graph of Streams
528 -- ** Inserting Streams in Streams
549 -- * Generalize Inner Monad
553 -- * Transform Inner Monad
569 , iterateMapLeftsWith
571 -- * Concurrency Control
601import Control.Concurrent (threadDelay)
602import Control.Exception (assert)
603import Control.Monad.IO.Class (MonadIO(..))
604import Data.Kind (Type)
605import Data.Heap (Entry(..))
606import Data.Maybe (isNothing)
607import Foreign.Storable (Storable)
608import Streamly.Internal.Data.Stream.Enumeration
609 (Enumerable(..), enumerate, enumerateTo)
610import Streamly.Internal.Data.Fold.Types (Fold (..))
611import Streamly.Internal.Data.Array.Storable.Foreign.Types (Array)
612import Streamly.Internal.Data.SVar (MonadAsync, Rate (..))
613import Streamly.Internal.Data.Stream.Ahead (AheadT, Ahead, aheadly)
614import Streamly.Internal.Data.Stream.Async
615 ( AsyncT, Async, WAsyncT, WAsync, mkAsync, asyncly
617import Streamly.Internal.Data.Stream.Combinators
618 ( inspectMode, maxBuffer, maxThreads, rate, avgRate, minRate
619 , maxRate, constRate)
620import Streamly.Internal.Data.Stream.Parallel
621 ( ParallelT, Parallel, parallely)
622import Streamly.Internal.Data.Stream.StreamK (IsStream((|:), consM), adapt)
623import Streamly.Internal.Data.Stream.Serial
624 ( SerialT, WSerialT, Serial, WSerial, serially
626import Streamly.Internal.Data.Stream.Zip
627 ( ZipSerialM, ZipSerial, ZipAsyncM, ZipAsync, zipSerially, zipAsyncly)
628import Streamly.Internal.Data.Time.Units
629 ( AbsTime, MilliSecond64(..), addToAbsTime, toRelTime
630 , toAbsTime, RelTime64)
631import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))
633import Streamly.Internal.Data.Stream.IsStream.Generate
634import Streamly.Internal.Data.Stream.IsStream.Eliminate
635import Streamly.Internal.Data.Stream.IsStream.Transform
636import Streamly.Internal.Data.Stream.IsStream.Exceptions
638import qualified Streamly.Internal.Data.Array.Storable.Foreign as A
639import qualified Streamly.Internal.Data.Fold as FL
640import qualified Streamly.Internal.Data.Stream.Prelude as P
641import qualified Streamly.Internal.Data.Stream.StreamK as K
642import qualified Streamly.Internal.Data.Stream.StreamD as D
643import qualified Streamly.Internal.Data.Stream.Serial as Serial
644import qualified Streamly.Internal.Data.Stream.Parallel as Par
645import qualified Streamly.Internal.Data.Stream.Zip as Z
646import qualified Data.Heap as H
647import qualified Data.Map.Strict as Map
650 ( filter, drop, dropWhile, take, takeWhile, zipWith, foldr
651 , foldl, map, mapM, mapM_, sequence, all, any, sum, product, elem
652 , notElem, maximum, minimum, head, last, tail, length, null
653 , reverse, iterate, init, and, or, lookup, foldr1, (!!)
654 , scanl, scanl1, replicate, concatMap, span, splitAt, break
655 , repeat, concat, mconcat)
657------------------------------------------------------------------------------
659------------------------------------------------------------------------------
661-- | Takes a callback setter function and provides it with a callback. The
662-- callback when invoked adds a value at the tail of the stream. Returns a
663-- stream of values generated by the callback.
667{-# INLINE fromCallback #-}
668fromCallback :: MonadAsync m => ((a -> m ()) -> m ()) -> SerialT m a
669fromCallback setCallback = concatM $ do
670 (callback, stream) <- D.newCallbackStream
674------------------------------------------------------------------------------
676------------------------------------------------------------------------------
679-- > drainN n = drain . take n
680-- > drainN n = fold (Fold.ltake n Fold.drain)
682-- Run maximum up to @n@ iterations of a stream.
686drainN :: Monad m => Int -> SerialT m a -> m ()
687drainN n = drain . take n
690-- > runN n = runStream . take n
692-- Run maximum up to @n@ iterations of a stream.
695{-# DEPRECATED runN "Please use \"drainN\" instead" #-}
697runN :: Monad m => Int -> SerialT m a -> m ()
701-- > drainWhile p = drain . takeWhile p
702-- > drainWhile p = fold (Fold.sliceSepBy (not . p) Fold.drain)
704-- Run a stream as long as the predicate holds true.
707{-# INLINE drainWhile #-}
708drainWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
709drainWhile p = drain . takeWhile p
712-- > runWhile p = runStream . takeWhile p
714-- Run a stream as long as the predicate holds true.
717{-# DEPRECATED runWhile "Please use \"drainWhile\" instead" #-}
718{-# INLINE runWhile #-}
719runWhile :: Monad m => (a -> Bool) -> SerialT m a -> m ()
722------------------------------------------------------------------------------
723-- Transformation by Inserting
724------------------------------------------------------------------------------
726-- | Intersperse a monadic action into the input stream after every @n@
730-- > S.drain $ S.interjectSuffix 1 (putChar ',') $ S.mapM (\\x -> threadDelay 1000000 >> putChar x) $ S.fromList "hello"
735{-# INLINE interjectSuffix #-}
737 :: (IsStream t, MonadAsync m)
738 => Double -> m a -> t m a -> t m a
739interjectSuffix n f xs = xs `Par.parallelFst` repeatM timed
740 where timed = liftIO (threadDelay (round $ n * 1000000)) >> f
742-------------------------------------------------------------------------------
744-------------------------------------------------------------------------------
746-- Note: The timestamp stream must be the second stream in the zip so that the
747-- timestamp is generated after generating the stream element and not before.
748-- If we do not do that then the following example will generate the same
749-- timestamp for first two elements:
751-- S.mapM_ print $ S.timestamped $ S.delay $ S.enumerateFromTo 1 3
753-- | Pair each element in a stream with an absolute timestamp, using a clock of
754-- specified granularity. The timestamp is generated just before the element
758-- >>> S.mapM_ print $ S.timestampWith 0.01 $ S.delay 1 $ S.enumerateFromTo 1 3
763{-# INLINE timestampWith #-}
764timestampWith :: (IsStream t, MonadAsync m, Functor (t m))
765 => Double -> t m a -> t m (AbsTime, a)
766timestampWith g stream = Z.zipWith (flip (,)) stream (absTimesWith g)
768-- TBD: check performance vs a custom implementation without using zipWith.
772{-# INLINE timestamped #-}
773timestamped :: (IsStream t, MonadAsync m, Functor (t m))
774 => t m a -> t m (AbsTime, a)
775timestamped = timestampWith 0.01
777-- | Pair each element in a stream with relative times starting from 0, using a
778-- clock with the specified granularity. The time is measured just before the
779-- element is consumed.
782-- >>> S.mapM_ print $ S.timeIndexWith 0.01 $ S.delay 1 $ S.enumerateFromTo 1 3
787{-# INLINE timeIndexWith #-}
788timeIndexWith :: (IsStream t, MonadAsync m, Functor (t m))
789 => Double -> t m a -> t m (RelTime64, a)
790timeIndexWith g stream = Z.zipWith (flip (,)) stream (relTimesWith g)
792-- | Pair each element in a stream with relative times starting from 0, using a
793-- 10 ms granularity clock. The time is measured just before the element is
797-- >>> S.mapM_ print $ S.timeIndexed $ S.delay 1 $ S.enumerateFromTo 1 3
798-- (RelTime64 (NanoSecond64 0),1)
799-- (RelTime64 (NanoSecond64 996239000),2)
800-- (RelTime64 (NanoSecond64 1996712000),3)
805{-# INLINE timeIndexed #-}
806timeIndexed :: (IsStream t, MonadAsync m, Functor (t m))
807 => t m a -> t m (RelTime64, a)
808timeIndexed = timeIndexWith 0.01
810------------------------------------------------------------------------------
812------------------------------------------------------------------------------
814-- | Returns the first index that satisfies the given predicate.
816-- > findIndex = fold Fold.findIndex
819{-# INLINE findIndex #-}
820findIndex :: Monad m => (a -> Bool) -> SerialT m a -> m (Maybe Int)
821findIndex p = head . findIndices p
823-- | Returns the first index where a given value is found in the stream.
825-- > elemIndex a = findIndex (== a)
828{-# INLINE elemIndex #-}
829elemIndex :: (Monad m, Eq a) => a -> SerialT m a -> m (Maybe Int)
830elemIndex a = findIndex (== a)
832------------------------------------------------------------------------------
834------------------------------------------------------------------------------
836-- | Returns 'True' if the first stream is a suffix of the second. A stream is
837-- considered a suffix of itself.
840-- > S.isSuffixOf (S.fromList "hello") (S.fromList "hello" :: SerialT IO Char)
844-- Space: @O(n)@, buffers entire input stream and the suffix.
848-- /Suboptimal/ - Help wanted.
850{-# INLINE isSuffixOf #-}
851isSuffixOf :: (Monad m, Eq a) => SerialT m a -> SerialT m a -> m Bool
852isSuffixOf suffix stream = reverse suffix `isPrefixOf` reverse stream
854-- | Returns 'True' if the first stream is an infix of the second. A stream is
855-- considered an infix of itself.
858-- > S.isInfixOf (S.fromList "hello") (S.fromList "hello" :: SerialT IO Char)
862-- Space: @O(n)@ worst case where @n@ is the length of the infix.
866-- /Requires 'Storable' constraint/ - Help wanted.
868{-# INLINE isInfixOf #-}
869isInfixOf :: (MonadIO m, Eq a, Enum a, Storable a)
870 => SerialT m a -> SerialT m a -> m Bool
871isInfixOf infx stream = do
872 arr <- fold A.write infx
873 -- XXX can use breakOnSeq instead (when available)
874 r <- null $ drop 1 $ splitOnSeq arr FL.drain stream
877-- Note: If we want to return a Maybe value to know whether the
878-- suffix/infix was present or not along with the stripped stream then
879-- we need to buffer the whole input stream.
881-- | Drops the given suffix from a stream. Returns 'Nothing' if the stream does
882-- not end with the given suffix. Returns @Just nil@ when the suffix is the
883-- same as the stream.
885-- It may be more efficient to convert the stream to an Array and use
886-- stripSuffix on that especially if the elements have a Storable or Prim
889-- Space: @O(n)@, buffers the entire input stream as well as the suffix
892{-# INLINE stripSuffix #-}
895 => SerialT m a -> SerialT m a -> m (Maybe (SerialT m a))
896stripSuffix m1 m2 = fmap reverse <$> stripPrefix (reverse m1) (reverse m2)
898------------------------------------------------------------------------------
899-- Split on a delimiter sequence
900------------------------------------------------------------------------------
902-- XXX use a non-monadic intersperse to remove the MonadAsync constraint.
904-- | Like 'splitOnSeq' but splits the separator as well, as an infix token.
906-- > splitOn'_ pat xs = S.toList $ S.splitOn' (A.fromList pat) (FL.toList) (S.fromList xs)
908-- >>> splitOn'_ "" "hello"
909-- > ["h","","e","","l","","l","","o"]
911-- >>> splitOn'_ "hello" ""
914-- >>> splitOn'_ "hello" "hello"
917-- >>> splitOn'_ "x" "hello"
920-- >>> splitOn'_ "h" "hello"
923-- >>> splitOn'_ "o" "hello"
926-- >>> splitOn'_ "e" "hello"
929-- >>> splitOn'_ "l" "hello"
930-- > ["he","l","","l","o"]
932-- >>> splitOn'_ "ll" "hello"
936{-# INLINE splitBySeq #-}
938 :: (IsStream t, MonadAsync m, Storable a, Enum a, Eq a)
939 => Array a -> Fold m a b -> t m a -> t m b
941 intersperseM (fold f (A.toStream patt)) $ splitOnSeq patt f m
943------------------------------------------------------------------------------
945------------------------------------------------------------------------------
947-- | Like 'iterateM' but iterates after mapping a stream generator on the
950-- Yield an input element in the output stream, map a stream generator on it
951-- and then do the same on the resulting stream. This can be used for a depth
952-- first traversal of a tree like structure.
954-- Note that 'iterateM' is a special case of 'iterateMapWith':
957-- iterateM f = iterateMapWith serial (yieldM . f) . yieldM
960-- It can be used to traverse a tree structure. For example, to list a
964-- Stream.iterateMapWith Stream.serial
965-- (either Dir.toEither (const nil))
966-- (yield (Left "tmp"))
971{-# INLINE iterateMapWith #-}
974 => (t m a -> t m a -> t m a)
978iterateMapWith combine f = concatMapWith combine go
980 go x = yield x `combine` concatMapWith combine go (f x)
983{-# INLINE iterateUnfold #-}
984iterateUnfold :: (IsStream t, MonadAsync m)
985 => Unfold m a a -> t m a -> t m a
986iterateUnfold unf xs = undefined
989------------------------------------------------------------------------------
991------------------------------------------------------------------------------
993-- To traverse graphs we need a state to be carried around in the traversal.
994-- For example, we can use a hashmap to store the visited status of nodes.
996-- | Like 'iterateMap' but carries a state in the stream generation function.
997-- This can be used to traverse graph like structures, we can remember the
998-- visited nodes in the state to avoid cycles.
1000-- Note that a combination of 'iterateMap' and 'usingState' can also be used to
1001-- traverse graphs. However, this function provides a more localized state
1002-- instead of using a global state.
1008{-# INLINE iterateSmapMWith #-}
1010 :: (IsStream t, Monad m)
1011 => (t m a -> t m a -> t m a)
1012 -> (b -> a -> m (b, t m a))
1016iterateSmapMWith combine f initial stream =
1017 concatMap (\b -> concatMapWith combine (go b) stream) (yieldM initial)
1021 go b a = yield a `combine` feedback b a
1025 (\(b1, s) -> concatMapWith combine (go b1) s)
1028------------------------------------------------------------------------------
1030------------------------------------------------------------------------------
1032-- | Given a stream value in the underlying monad, lift and join the underlying
1033-- monad with the stream monad.
1036-- concatM = concat . yieldM
1037-- concatM = concat . lift -- requires @(MonadTrans t)@
1038-- concatM = join . lift -- requires @(MonadTrans t@, @Monad (t m))@
1041-- See also: 'concat', 'sequence'
1045{-# INLINE concatM #-}
1046concatM :: (IsStream t, Monad m) => m (t m a) -> t m a
1047concatM generator = concatMapM (\() -> generator) (yield ())
1049------------------------------------------------------------------------------
1051------------------------------------------------------------------------------
1053-- | In an 'Either' stream iterate on 'Left's. This is a special case of
1057-- iterateMapLeftsWith combine f = iterateMapWith combine (either f (const nil))
1060-- To traverse a directory tree:
1063-- iterateMapLeftsWith serial Dir.toEither (yield (Left "tmp"))
1068{-# INLINE iterateMapLeftsWith #-}
1071 => (t m (Either a b) -> t m (Either a b) -> t m (Either a b))
1072 -> (a -> t m (Either a b))
1075iterateMapLeftsWith combine f = iterateMapWith combine (either f (const K.nil))
1077-------------------------------------------------------------------------------
1079-------------------------------------------------------------------------------
1081-- XXX we can implement this by repeatedly applying the 'lrunFor' fold.
1082-- XXX add this example after fixing the serial stream rate control
1083-- >>> S.toList $ S.take 5 $ intervalsOf 1 FL.sum $ constRate 2 $ S.enumerateFrom 1
1086-- | Group the input stream into windows of @n@ second each and then fold each
1087-- group using the provided fold function.
1090{-# INLINE intervalsOf #-}
1092 :: (IsStream t, MonadAsync m)
1093 => Double -> Fold m a b -> t m a -> t m b
1095 splitWithSuffix isNothing (FL.lcatMaybes f)
1096 (interjectSuffix n (return Nothing) (Serial.map Just xs))
1098------------------------------------------------------------------------------
1099-- Windowed classification
1100------------------------------------------------------------------------------
1102-- We divide the stream into windows or chunks in space or time and each window
1103-- can be associated with a key, all events associated with a particular key in
1104-- the window can be folded to a single result. The stream can be split into
1105-- windows by size or by using a split predicate on the elements in the stream.
1106-- For example, when we receive a closing flag, we can close the window.
1108-- A "chunk" is a space window and a "session" is a time window. Are there any
1109-- other better short words to describe them. An alternative is to use
1110-- "swindow" and "twindow". Another word for "session" could be "spell".
1112-- TODO: To mark the position in space or time we can have Indexed or
1113-- TimeStamped types. That can make it easy to deal with the position indices
1116------------------------------------------------------------------------------
1117-- Keyed Sliding Windows
1118------------------------------------------------------------------------------
1121{-# INLINABLE classifySlidingChunks #-}
1122classifySlidingChunks
1123 :: (IsStream t, MonadAsync m, Ord k)
1124 => Int -- ^ window size
1125 -> Int -- ^ window slide
1126 -> Fold m a b -- ^ Fold to be applied to window events
1127 -> t m (k, a, Bool) -- ^ window key, data, close event
1129classifySlidingChunks wsize wslide (Fold step initial extract) str
1132-- XXX Another variant could be to slide the window on an event, e.g. in TCP we
1133-- slide the send window when an ack is received and we slide the receive
1134-- window when a sequence is complete. Sliding is stateful in case of TCP,
1135-- sliding releases the send buffer or makes data available to the user from
1136-- the receive buffer.
1137{-# INLINABLE classifySlidingSessions #-}
1138classifySlidingSessions
1139 :: (IsStream t, MonadAsync m, Ord k)
1140 => Double -- ^ timer tick in seconds
1141 -> Double -- ^ time window size
1142 -> Double -- ^ window slide
1143 -> Fold m a b -- ^ Fold to be applied to window events
1144 -> t m (k, a, Bool, AbsTime) -- ^ window key, data, close flag, timestamp
1146classifySlidingSessions tick interval slide (Fold step initial extract) str
1150------------------------------------------------------------------------------
1151-- Sliding Window Buffers
1152------------------------------------------------------------------------------
1154-- These buffered versions could be faster than concurrent incremental folds of
1155-- all overlapping windows as in many cases we may not need all the values to
1156-- compute the fold, we can just compute the result using the old value and new
1157-- value. However, we may need the buffer once in a while, for example for
1158-- string search we usually compute the hash incrementally but when the hash
1159-- matches the hash of the pattern we need to compare the whole string.
1161-- XXX we should be able to implement sequence based splitting combinators
1162-- using this combinator.
1165-- | Buffer n elements of the input in a ring buffer. When t new elements are
1166-- collected, slide the window to remove the same number of oldest elements,
1167-- insert the new elements, and apply an incremental fold on the sliding
1168-- window, supplying the outgoing elements, the new ring buffer as arguments.
1170 :: (IsStream t, Monad m, Ord a, Storable a)
1171 => Int -- window size
1172 -> Int -- window slide
1173 -> Fold m (Ring a, Array a) b
1176slidingChunkBuffer = undefined
1178-- Buffer n seconds worth of stream elements of the input in a radix tree.
1179-- Every t seconds, remove the items that are older than n seconds, and apply
1180-- an incremental fold on the sliding window, supplying the outgoing elements,
1181-- and the new radix tree buffer as arguments.
1183 :: (IsStream t, Monad m, Ord a, Storable a)
1184 => Int -- window size
1186 -> Fold m (RTree a, Array a) b
1189slidingSessionBuffer = undefined
1192------------------------------------------------------------------------------
1193-- Keyed Session Windows
1194------------------------------------------------------------------------------
1197-- | Keyed variable size space windows. Close the window if we do not receive a
1198-- window event in the next "spaceout" elements.
1199{-# INLINABLE classifyChunksBy #-}
1201 :: (IsStream t, MonadAsync m, Ord k)
1202 => Int -- ^ window spaceout (spread)
1203 -> Bool -- ^ reset the spaceout when a chunk window element is received
1204 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1205 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1207classifyChunksBy spanout reset (Fold step initial extract) str = undefined
1209-- | Like 'classifyChunksOf' but the chunk size is reset if an element is
1210-- received within the chunk size window. The chunk gets closed only if no
1211-- element is received within the chunk window.
1213{-# INLINABLE classifyKeepAliveChunks #-}
1214classifyKeepAliveChunks
1215 :: (IsStream t, MonadAsync m, Ord k)
1216 => Int -- ^ window spaceout (spread)
1217 -> Fold m a b -- ^ Fold to be applied to chunk window elements
1218 -> t m (k, a, Bool) -- ^ chunk key, data, last element
1220classifyKeepAliveChunks spanout = classifyChunksBy spanout True
1223data SessionState t m k a b = SessionState
1224 { sessionCurTime :: !AbsTime -- ^ time since last event
1225 , sessionEventTime :: !AbsTime -- ^ time as per last event
1226 , sessionCount :: !Int -- ^ total number sessions in progress
1227 , sessionTimerHeap :: H.Heap (H.Entry AbsTime k) -- ^ heap for timeouts
1228 , sessionKeyValueMap :: Map.Map k a -- ^ Stored sessions for keys
1229 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions
1234-- XXX Perhaps we should use an "Event a" type to represent timestamped data.
1235-- | @classifySessionsBy tick timeout idle pred f stream@ groups timestamped
1236-- events in an input event stream into sessions based on a session key. Each
1237-- element in the input stream is an event consisting of a triple @(session key,
1238-- sesssion data, timestamp)@. @session key@ is a key that uniquely identifies
1239-- the session. All the events belonging to a session are folded using the fold
1240-- @f@ until the fold terminates or a timeout has occurred. The session key and
1241-- the result of the fold are emitted in the output stream when the session is
1244-- When @idle@ is 'False', @timeout@ is the maximum lifetime of a session in
1245-- seconds, measured from the @timestamp@ of the first event in that session.
1246-- When @idle@ is 'True' then the timeout is an idle timeout, it is reset after
1247-- every event received in the session.
1249-- @timestamp@ in an event characterizes the time when the input event was
1250-- generated, this is an absolute time measured from some @Epoch@. The notion
1251-- of current time is maintained by a monotonic event time clock using the
1252-- timestamps seen in the input stream. The latest timestamp seen till now is
1253-- used as the base for the current time. When no new events are seen, a timer
1254-- is started with a tick duration specified by @tick@. This timer is used to
1255-- detect session timeouts in the absence of new events.
1257-- The predicate @pred@ is invoked with the current session count, if it
1258-- returns 'True' a session is ejected from the session cache before inserting
1259-- a new session. This could be useful to alert or eject sessions when the
1260-- number of sessions becomes too high.
1264{-# INLINABLE classifySessionsBy #-}
1266 :: (IsStream t, MonadAsync m, Ord k)
1267 => Double -- ^ timer tick in seconds
1268 -> Double -- ^ session timeout in seconds
1269 -> Bool -- ^ reset the timeout when an event is received
1270 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count
1271 -> Fold m a b -- ^ Fold to be applied to session events
1272 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
1273 -> t m (k, b) -- ^ session key, fold result
1274classifySessionsBy tick tmout reset ejectPred
1275 (Fold step initial extract) str =
1276 concatMap sessionOutputStream $
1277 scanlMAfter' sstep (return szero) flush stream
1281 timeoutMs = toRelTime (round (tmout * 1000) :: MilliSecond64)
1282 tickMs = toRelTime (round (tick * 1000) :: MilliSecond64)
1283 szero = SessionState
1284 { sessionCurTime = toAbsTime (0 :: MilliSecond64)
1285 , sessionEventTime = toAbsTime (0 :: MilliSecond64)
1287 , sessionTimerHeap = H.empty
1288 , sessionKeyValueMap = Map.empty
1289 , sessionOutputStream = K.nil
1292 -- We can eject sessions based on the current session count to limit
1293 -- memory consumption. There are two possible strategies:
1295 -- 1) Eject old sessions or sessions beyond a certain/lower timeout
1296 -- threshold even before timeout, effectively reduce the timeout.
1297 -- 2) Drop creation of new sessions but keep accepting new events for the
1300 -- We use the first strategy as of now.
1302 -- Got a new stream input element
1303 sstep session@SessionState{..} (Just (key, value, timestamp)) = do
1304 -- XXX we should use a heap in pinned memory to scale it to a large
1307 -- To detect session inactivity we keep a timestamp of the latest event
1308 -- in the Map along with the fold result. When we purge the session
1309 -- from the heap we match the timestamp in the heap with the timestamp
1310 -- in the Map, if the latest timestamp is newer and has not expired we
1311 -- reinsert the key in the heap.
1313 -- XXX if the key is an Int, we can also use an IntMap for slightly
1314 -- better performance.
1316 let curTime = max sessionEventTime timestamp
1317 mOld = Map.lookup key sessionKeyValueMap
1322 Just (Tuple' _ acc) -> return acc
1323 res <- step fs value
1326 -- deleting a key from the heap is expensive, so we never
1327 -- delete a key from heap, we just purge it from the Map and it
1328 -- gets purged from the heap on timeout. We just need an extra
1329 -- lookup in the Map when the key is purged from the heap, that
1330 -- should not be expensive.
1332 let (mp, cnt) = case mOld of
1333 Nothing -> (sessionKeyValueMap, sessionCount)
1334 Just _ -> (Map.delete key sessionKeyValueMap
1337 { sessionCurTime = curTime
1338 , sessionEventTime = curTime
1339 , sessionCount = cnt
1340 , sessionKeyValueMap = mp
1341 , sessionOutputStream = yield (key, fb)
1343 FL.Partial fs1 -> do
1344 let acc = Tuple' timestamp fs1
1345 (hp1, mp1, out1, cnt1) <- do
1346 let vars = (sessionTimerHeap, sessionKeyValueMap,
1347 K.nil, sessionCount)
1349 -- inserting new entry
1351 -- Eject a session from heap and map is needed
1352 eject <- ejectPred sessionCount
1353 (hp, mp, out, cnt) <-
1358 -- Insert the new session in heap
1359 let expiry = addToAbsTime timestamp timeoutMs
1360 hp' = H.insert (Entry expiry key) hp
1361 in return (hp', mp, out, cnt + 1)
1362 -- updating old entry
1363 Just _ -> return vars
1365 let mp2 = Map.insert key acc mp1
1366 return $ SessionState
1367 { sessionCurTime = curTime
1368 , sessionEventTime = curTime
1369 , sessionCount = cnt1
1370 , sessionTimerHeap = hp1
1371 , sessionKeyValueMap = mp2
1372 , sessionOutputStream = out1
1375 -- Got a timer tick event
1376 sstep sessionState@SessionState{..} Nothing =
1377 let curTime = addToAbsTime sessionCurTime tickMs
1378 in ejectExpired sessionState curTime
1380 flush session@SessionState{..} = do
1381 (hp', mp', out, count) <-
1384 , sessionKeyValueMap
1389 { sessionCount = count
1390 , sessionTimerHeap = hp'
1391 , sessionKeyValueMap = mp'
1392 , sessionOutputStream = out
1395 -- delete from map and output the fold accumulator
1396 ejectEntry hp mp out cnt acc key = do
1398 let out1 = (key, sess) `K.cons` out
1399 let mp1 = Map.delete key mp
1400 return (hp, mp1, out1, cnt - 1)
1402 ejectAll (hp, mp, out, !cnt) = do
1403 let hres = H.uncons hp
1405 Just (Entry _ key, hp1) -> do
1406 r <- case Map.lookup key mp of
1407 Nothing -> return (hp1, mp, out, cnt)
1408 Just (Tuple' _ acc) -> ejectEntry hp1 mp out cnt acc key
1411 assert (Map.null mp) (return ())
1412 return (hp, mp, out, cnt)
1414 ejectOne (hp, mp, out, !cnt) = do
1415 let hres = H.uncons hp
1417 Just (Entry expiry key, hp1) ->
1418 case Map.lookup key mp of
1419 Nothing -> ejectOne (hp1, mp, out, cnt)
1420 Just (Tuple' latestTS acc) -> do
1421 let expiry1 = addToAbsTime latestTS timeoutMs
1422 if not reset || expiry1 <= expiry
1423 then ejectEntry hp1 mp out cnt acc key
1425 -- reset the session timeout and continue
1426 let hp2 = H.insert (Entry expiry1 key) hp1
1427 in ejectOne (hp2, mp, out, cnt)
1429 assert (Map.null mp) (return ())
1430 return (hp, mp, out, cnt)
1432 ejectExpired session@SessionState{..} curTime = do
1433 (hp', mp', out, count) <-
1434 ejectLoop sessionTimerHeap sessionKeyValueMap K.nil sessionCount
1436 { sessionCurTime = curTime
1437 , sessionCount = count
1438 , sessionTimerHeap = hp'
1439 , sessionKeyValueMap = mp'
1440 , sessionOutputStream = out
1445 ejectLoop hp mp out !cnt = do
1446 let hres = H.uncons hp
1448 Just (Entry expiry key, hp1) -> do
1450 if curTime >= expiry
1451 then return (True, False)
1457 case Map.lookup key mp of
1458 Nothing -> ejectLoop hp1 mp out cnt
1459 Just (Tuple' latestTS acc) -> do
1460 let expiry1 = addToAbsTime latestTS timeoutMs
1461 if expiry1 <= curTime || not reset || force
1463 (hp2,mp1,out1,cnt1) <-
1464 ejectEntry hp1 mp out cnt acc key
1465 ejectLoop hp2 mp1 out1 cnt1
1467 -- reset the session timeout and continue
1468 let hp2 = H.insert (Entry expiry1 key) hp1
1469 in ejectLoop hp2 mp out cnt
1470 else return (hp, mp, out, cnt)
1472 assert (Map.null mp) (return ())
1473 return (hp, mp, out, cnt)
1475 -- merge timer events in the stream
1476 stream = Serial.map Just str `Par.parallelFst` repeatM timer
1478 liftIO $ threadDelay (round $ tick * 1000000)
1481-- | Like 'classifySessionsOf' but the session is kept alive if an event is
1482-- received within the session window. The session times out and gets closed
1483-- only if no event is received within the specified session window size.
1485-- If the ejection predicate returns 'True', the session that was idle for
1486-- the longest time is ejected before inserting a new session.
1489-- classifyKeepAliveSessions timeout pred = classifySessionsBy 1 timeout True pred
1494{-# INLINABLE classifyKeepAliveSessions #-}
1495classifyKeepAliveSessions ::
1496 (IsStream t, MonadAsync m, Ord k)
1497 => Double -- ^ session inactive timeout
1498 -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
1499 -> Fold m a b -- ^ Fold to be applied to session payload data
1500 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
1502classifyKeepAliveSessions tmout =
1503 classifySessionsBy 1 tmout True
1505------------------------------------------------------------------------------
1506-- Keyed tumbling windows
1507------------------------------------------------------------------------------
1509-- Tumbling windows is a special case of sliding windows where the window slide
1510-- is the same as the window size. Or it can be a special case of session
1511-- windows where the reset flag is set to False.
1513-- XXX instead of using the early termination flag in the stream, we can use an
1514-- early terminating fold instead.
1517-- | Split the stream into fixed size chunks of specified size. Within each
1518-- such chunk fold the elements in buckets identified by the keys. A particular
1519-- bucket fold can be terminated early if a closing flag is encountered in an
1520-- element for that key.
1523{-# INLINABLE classifyChunksOf #-}
1525 :: (IsStream t, MonadAsync m, Ord k)
1526 => Int -- ^ window size
1527 -> Fold m a b -- ^ Fold to be applied to window events
1528 -> t m (k, a, Bool) -- ^ window key, data, close event
1530classifyChunksOf wsize = classifyChunksBy wsize False
1533-- | Split the stream into fixed size time windows of specified interval in
1534-- seconds. Within each such window, fold the elements in sessions identified
1535-- by the session keys. The fold result is emitted in the output stream if the
1536-- fold returns a 'Left' result or if the time window ends.
1538-- Session @timestamp@ in the input stream is an absolute time from some epoch,
1539-- characterizing the time when the input element was generated. To detect
1540-- session window end, a monotonic event time clock is maintained synced with
1541-- the timestamps with a clock resolution of 1 second.
1543-- If the ejection predicate returns 'True', the session with the longest
1544-- lifetime is ejected before inserting a new session.
1547-- classifySessionsOf interval pred = classifySessionsBy 1 interval False pred
1552-- $ S.classifySessionsOf 3 (const (return False)) (fmap Right FL.toList)
1553-- $ S.map (\(ts,(k,a)) -> (k, a, ts))
1556-- $ (,) <$> S.fromList [1,2,3] <*> S.fromList [1,2,3]
1561{-# INLINABLE classifySessionsOf #-}
1562classifySessionsOf ::
1563 (IsStream t, MonadAsync m, Ord k)
1564 => Double -- ^ time window size
1565 -> (Int -> m Bool) -- ^ predicate to eject sessions on session count
1566 -> Fold m a b -- ^ Fold to be applied to session events
1567 -> t m (k, a, AbsTime) -- ^ session key, data, timestamp
1569classifySessionsOf interval =
1570 classifySessionsBy 1 interval False
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