fca88a4 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Change the input stream type of classifySessionsBy

fca88a4Harendra Kumar6 years ago

benchmark/Streamly/Benchmark/Prelude/Serial/Transformation2.hs−1

compare with master

unifiedsplit

@@ -202,7 +202,6 @@ classifySessionsOf :: (S.MonadAsync m) => SerialT m Int -> m ()

202classifySessionsOf =

203 S.drain

204 . Internal.classifySessionsOf (const (return False)) 3 FL.drain

205− . S.map (\(ts,(k,a)) -> (k, a, ts))

206 . Internal.timestamped

205 . Internal.timestamped

207 . S.concatMap (\x -> S.map (x,) (S.enumerateFromTo 1 (10 :: Int)))

206 . S.concatMap (\x -> S.map (x,) (S.enumerateFromTo 1 (10 :: Int)))

208

207

src/Streamly/Internal/Data/Stream/IsStream/Nesting.hs+17−21

@@ -202,14 +202,13 @@ module Streamly.Internal.Data.Stream.IsStream.Nesting

202

203 -- ** Keyed Window Classification

204

205− -- | Split the stream into chunks or windows in space or time. Each window

205+ -- | Split the stream into chunks or windows by position or time. Each

206− -- can be associated with a key, all events associated with a particular

206+ -- window can be associated with a key, all events associated with a

207− -- key in the window can be folded to a single result. The stream is split

207+ -- particular key in the window can be folded to a single result. The

208− -- into windows of specified size, the window termination can be

209− -- dynamically controlled by the fold.

208+ -- window termination can be dynamically controlled by the fold.

210 --

209 --

211− -- The term "chunk" is used for a space window and the term "session" is

210+ -- The term "chunk" is used for a window defined by position of elements

212− -- used for a time window.

211+ -- and the term "session" is used for a time window.

213

212

214 -- *** Tumbling Windows

213 -- *** Tumbling Windows

215 -- | A new window starts after the previous window is finished.

214 -- | A new window starts after the previous window is finished.

@@ -1901,12 +1900,10 @@ data SessionState t m k a b = SessionState

1901 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions

1900 , sessionOutputStream :: t (m :: Type -> Type) (k, b) -- ^ Completed sessions

1902 }

1901 }

1903

1902

1904−-- XXX Perhaps we should use an "Event a" type to represent timestamped data.

1905−

1906-- | @classifySessionsBy tick keepalive predicate timeout fold stream@

1903-- | @classifySessionsBy tick keepalive predicate timeout fold stream@

1907−-- classifies an input event @stream@ consisting of @(key, value, timestamp)@

1904+-- classifies an input event @stream@ consisting of @(timestamp, (key,

1908−-- into sessions based on the @key@, folding all the values corresponding to

1905+-- value))@ into sessions based on the @key@, folding all the values

1909−-- the same key into a session using the supplied @fold@.

1906+-- corresponding to the same key into a session using the supplied @fold@.

1910--

1907--

1911-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the

1908-- When the fold terminates or a @timeout@ occurs, a tuple consisting of the

1912-- session key and the folded value is emitted in the output stream. The

1909-- session key and the folded value is emitted in the output stream. The

@@ -1928,11 +1925,10 @@ data SessionState t m k a b = SessionState

1928--

1925--

1929-- >>> :{

1926-- >>> :{

1930-- Stream.mapM_ print

1927-- Stream.mapM_ print

1931−-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.takeLE 3 Fold.toList)

1928+-- $ Stream.classifySessionsBy 1 False (const (return False)) 3 (Fold.take 3 Fold.toList)

1932−-- $ Stream.map (\\(ts,(k,a)) -> (k, a, ts))

1933-- $ Stream.timestamped

1929-- $ Stream.timestamped

1934-- $ Stream.delay 1

1930-- $ Stream.delay 1

1935−-- $ (,) \<$> Stream.fromList [1,2,3] \<*> Stream.fromList [\'a',\'b',\'c']

1931+-- $ (,) <$> Stream.fromList [1,2,3] <*> Stream.fromList ['a','b','c']

1936-- :}

1932-- :}

1937-- (1,"abc")

1933-- (1,"abc")

1938-- (2,"abc")

1934-- (2,"abc")

@@ -1947,8 +1943,8 @@ classifySessionsBy

1947 -> Bool -- ^ reset the timer when an event is received

1943 -> Bool -- ^ reset the timer when an event is received

1948 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count

1944 -> (Int -> m Bool) -- ^ predicate to eject sessions based on session count

1949 -> Double -- ^ session timeout in seconds

1945 -> Double -- ^ session timeout in seconds

1950− -> Fold m a b -- ^ Fold to be applied to session events

1946+ -> Fold m a b -- ^ Fold to be applied to session data

1951− -> t m (k, a, AbsTime) -- ^ session key, data, timestamp

1947+ -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)

1952 -> t m (k, b) -- ^ session key, fold result

1948 -> t m (k, b) -- ^ session key, fold result

1953classifySessionsBy tick reset ejectPred tmout

1949classifySessionsBy tick reset ejectPred tmout

1954 (Fold step initial extract) str =

1950 (Fold step initial extract) str =

@@ -1979,7 +1975,7 @@ classifySessionsBy tick reset ejectPred tmout

1979 -- We use the first strategy as of now.

1975 -- We use the first strategy as of now.

1980

1976

1981 -- Got a new stream input element

1977 -- Got a new stream input element

1982− sstep session@SessionState{..} (Just (key, value, timestamp)) = do

1978+ sstep session@SessionState{..} (Just (timestamp, (key, value))) = do

1983 -- XXX we should use a heap in pinned memory to scale it to a large

1979 -- XXX we should use a heap in pinned memory to scale it to a large

1984 -- size

1980 -- size

1985 --

1981 --

@@ -2176,7 +2172,7 @@ classifyKeepAliveSessions ::

2176 => (Int -> m Bool) -- ^ predicate to eject sessions on session count

2172 => (Int -> m Bool) -- ^ predicate to eject sessions on session count

2177 -> Double -- ^ session inactive timeout

2173 -> Double -- ^ session inactive timeout

2178 -> Fold m a b -- ^ Fold to be applied to session payload data

2174 -> Fold m a b -- ^ Fold to be applied to session payload data

2179− -> t m (k, a, AbsTime) -- ^ session key, data, timestamp

2175+ -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)

2180 -> t m (k, b)

2176 -> t m (k, b)

2181classifyKeepAliveSessions = classifySessionsBy 1 True

2177classifyKeepAliveSessions = classifySessionsBy 1 True

2182

2178

@@ -2222,8 +2218,8 @@ classifySessionsOf ::

2222 (IsStream t, MonadAsync m, Ord k)

2218 (IsStream t, MonadAsync m, Ord k)

2223 => (Int -> m Bool) -- ^ predicate to eject sessions on session count

2219 => (Int -> m Bool) -- ^ predicate to eject sessions on session count

2224 -> Double -- ^ time window size

2220 -> Double -- ^ time window size

2225− -> Fold m a b -- ^ Fold to be applied to session events

2221+ -> Fold m a b -- ^ Fold to be applied to session data

2226− -> t m (k, a, AbsTime) -- ^ session key, data, timestamp

2222+ -> t m (AbsTime, (k, a)) -- ^ timestamp, (session key, session data)

2227 -> t m (k, b)

2223 -> t m (k, b)

2228classifySessionsOf = classifySessionsBy 1 False

2224classifySessionsOf = classifySessionsBy 1 False

2229

2225