on master help sign in
Add benchmarks and tests for new operations

Also fix the signature of foldl1' Add notes about state passing

commitc39a442ddf871c3c6102b39f626f8cc18599a57d authorHarendra Kumar <harendra.kumar@gmail.com> dateThu 26 Jul 2018, 00:42 (8 years ago) parentcd50805
7 files changed+185−80
CONTRIBUTING.md+19 benchmark/BaseStreams.hs+2 benchmark/Linear.hs+12−1 benchmark/LinearOps.hs+119−72 benchmark/StreamKOps.hs+18 src/Streamly/Prelude.hs+3−3 test/Prop.hs+12−4
CONTRIBUTING.md+19
@@ -137,3 +137,22 @@ do not encourage `let` over `where`, you can freely use `where` clauses.
137137Also note that this is not possible in mutually recursive code. The mutually
138138recursive code has to be considered a single block.
139139
140+### Tricky Parts
141+
142+The state-passing through each API is currently fragile. Every time we run a
143+stream we need to be careful about the state we are passing to it. In case of
144+folds where there is no incoming state, we start with the initial state
145+`defState`. When we have an incoming state passed to us there are two cases:
146+
147+1. When we are building a concurrent stream that needs to share the same `SVar`
148+ we pass the incoming state as is.
149+2. In all other cases we must not share the SVar and everytime we pass on the
150+ state to run a stream we must use `rstState` to reset the `SVar` in the
151+ state.
152+
153+When in doubt just use `rstState` on the state before passing it on, we will at
154+most lose concurrency but the behavior will be correct.
155+
156+There is no type level enforcement about this as of now, and therefore we need
157+to be careful when coding. There are specific tests to detect and report any
158+problems due to this, all transform operations must be added to those tests.
benchmark/BaseStreams.hs+2
@@ -94,6 +94,8 @@ main = do
9494 , bgroup "elimination"
9595 [ benchIO "toNull" K.toNull K.sourceUnfoldrM
9696 , benchIO "uncons" K.uncons K.sourceUnfoldrM
97+ , benchFold "init" K.init K.sourceUnfoldrM
98+ , benchFold "tail" K.tail K.sourceUnfoldrM
9799 , benchIO "nullHeadTail" K.nullHeadTail K.sourceUnfoldrM
98100 , benchFold "toList" K.toList K.sourceUnfoldrM
99101 , benchFold "fold" K.foldl K.sourceUnfoldrM
benchmark/Linear.hs+12−1
@@ -56,12 +56,16 @@ main = do
5656 , bgroup "elimination"
5757 [ benchIO "toNull" $ Ops.toNull serially
5858 , benchIO "uncons" Ops.uncons
59+ , benchIO "init" Ops.init
60+ , benchIO "tail" Ops.tail
5961 , benchIO "nullHeadTail" Ops.nullHeadTail
6062 , benchIO "mapM_" Ops.mapM_
6163 , benchIO "toList" Ops.toList
6264 , benchIO "foldr" Ops.foldr
65+ , benchIO "foldr1" Ops.foldr1
6366 , benchIO "foldrM" Ops.foldrM
64− , benchIO "foldl'" Ops.foldl
67+ , benchIO "foldl'" Ops.foldl'
68+ , benchIO "foldl1'" Ops.foldl1'
6569
6670 , benchIO "last" Ops.last
6771 , benchIO "length" Ops.length
@@ -69,6 +73,11 @@ main = do
6973 , benchIO "notElem" Ops.notElem
7074 , benchIO "all" Ops.all
7175 , benchIO "any" Ops.any
76+ , benchIO "and" Ops.and
77+ , benchIO "or" Ops.or
78+ , benchIO "find" Ops.find
79+ , benchIO "findIndex" Ops.findIndex
80+ , benchIO "elemIndex" Ops.elemIndex
7281 , benchIO "maximum" Ops.maximum
7382 , benchIO "minimum" Ops.minimum
7483 , benchIO "sum" Ops.sum
@@ -83,6 +92,8 @@ main = do
8392 , benchIO "mapMaybeM" Ops.mapMaybeM
8493 , bench "sequence" $ nfIO $ randomRIO (1,1000) >>= \n ->
8594 (Ops.sequence serially) (Ops.sourceUnfoldrMAction n)
95+ , benchIO "findIndices" Ops.findIndices
96+ , benchIO "elemIndices" Ops.elemIndices
8697 , benchIO "concat" Ops.concat
8798 ]
8899 , bgroup "filtering"
benchmark/LinearOps.hs+119−72
@@ -11,7 +11,7 @@ module LinearOps where
1111
1212import Data.Maybe (fromJust)
1313import Prelude
14− (Monad, Int, (+), ($), (.), return, fmap, even, (>), (<=),
14+ (Monad, Int, (+), ($), (.), return, fmap, even, (>), (<=), (==), (<=),
1515 subtract, undefined, Maybe(..), odd, Bool, not)
1616
1717import qualified Streamly as S
@@ -25,76 +25,6 @@ maxValue = value + 1000
2525-- Benchmark ops
2626-------------------------------------------------------------------------------
2727
28−{-# INLINE uncons #-}
29−{-# INLINE nullHeadTail #-}
30−{-# INLINE scan #-}
31−{-# INLINE mapM_ #-}
32−{-# INLINE map #-}
33−{-# INLINE fmap #-}
34−{-# INLINE mapMaybe #-}
35−{-# INLINE filterEven #-}
36−{-# INLINE filterAllOut #-}
37−{-# INLINE filterAllIn #-}
38−{-# INLINE takeOne #-}
39−{-# INLINE takeAll #-}
40−{-# INLINE takeWhileTrue #-}
41−{-# INLINE takeWhileMTrue #-}
42−{-# INLINE dropAll #-}
43−{-# INLINE dropWhileTrue #-}
44−{-# INLINE dropWhileMTrue #-}
45−{-# INLINE zip #-}
46−{-# INLINE zipM #-}
47−{-# INLINE concat #-}
48−{-# INLINE composeAllInFilters #-}
49−{-# INLINE composeAllOutFilters #-}
50−{-# INLINE composeMapAllInFilter #-}
51−uncons, nullHeadTail, scan, mapM_, map, fmap, mapMaybe, filterEven, filterAllOut,
52− filterAllIn, takeOne, takeAll, takeWhileTrue, takeWhileMTrue, dropAll,
53− dropWhileTrue, dropWhileMTrue, zip, zipM,
54− concat, composeAllInFilters, composeAllOutFilters,
55− composeMapAllInFilter
56− :: Monad m
57− => Stream m Int -> m ()
58−
59−{-# INLINE composeMapM #-}
60−{-# INLINE zipAsync #-}
61−{-# INLINE zipAsyncM #-}
62−{-# INLINE mapMaybeM #-}
63−composeMapM, zipAsync, zipAsyncM, mapMaybeM :: S.MonadAsync m => Stream m Int -> m ()
64−
65−{-# INLINE toList #-}
66−{-# INLINE foldr #-}
67−{-# INLINE foldrM #-}
68−toList, foldr, foldrM :: Monad m => Stream m Int -> m [Int]
69−
70−{-# INLINE last #-}
71−{-# INLINE maximum #-}
72−{-# INLINE minimum #-}
73−last, minimum, maximum :: Monad m => Stream m Int -> m (Maybe Int)
74−
75−{-# INLINE foldl #-}
76−{-# INLINE length #-}
77−{-# INLINE sum #-}
78−{-# INLINE product #-}
79−foldl, length, sum, product :: Monad m => Stream m Int -> m Int
80−
81−{-# INLINE all #-}
82−{-# INLINE any #-}
83−{-# INLINE elem #-}
84−{-# INLINE notElem #-}
85−elem, notElem, all, any :: Monad m => Stream m Int -> m Bool
86−
87−{-# INLINE toNull #-}
88−toNull :: Monad m => (t m Int -> S.SerialT m Int) -> t m Int -> m ()
89−
90−{-# INLINE mapM #-}
91−mapM :: (S.IsStream t, S.MonadAsync m)
92− => (t m Int -> S.SerialT m Int) -> t m Int -> m ()
93−
94−{-# INLINE sequence #-}
95−sequence :: (S.IsStream t, S.MonadAsync m)
96− => (t m Int -> S.SerialT m Int) -> t m (m Int) -> m ()
97−
9828-------------------------------------------------------------------------------
9929-- Stream generation and elimination
10030-------------------------------------------------------------------------------
@@ -167,12 +97,65 @@ sourceUnfoldrMAction n = S.serially $ S.unfoldrM step n
16797runStream :: Monad m => Stream m a -> m ()
16898runStream = S.runStream
16999
100+{-# INLINE toList #-}
101+{-# INLINE foldr #-}
102+{-# INLINE foldrM #-}
103+toList, foldr, foldrM :: Monad m => Stream m Int -> m [Int]
104+
105+{-# INLINE last #-}
106+{-# INLINE maximum #-}
107+{-# INLINE minimum #-}
108+{-# INLINE find #-}
109+{-# INLINE findIndex #-}
110+{-# INLINE elemIndex #-}
111+{-# INLINE foldl1' #-}
112+{-# INLINE foldr1 #-}
113+last, minimum, maximum, find, findIndex, elemIndex, foldl1', foldr1 :: Monad m => Stream m Int -> m (Maybe Int)
114+
115+{-# INLINE foldl' #-}
116+{-# INLINE length #-}
117+{-# INLINE sum #-}
118+{-# INLINE product #-}
119+foldl', length, sum, product :: Monad m => Stream m Int -> m Int
120+
121+{-# INLINE all #-}
122+{-# INLINE any #-}
123+{-# INLINE and #-}
124+{-# INLINE or #-}
125+{-# INLINE elem #-}
126+{-# INLINE notElem #-}
127+elem, notElem, all, any, and, or :: Monad m => Stream m Int -> m Bool
128+
129+{-# INLINE toNull #-}
130+toNull :: Monad m => (t m Int -> S.SerialT m Int) -> t m Int -> m ()
170131toNull t = runStream . t
132+
133+{-# INLINE uncons #-}
134+uncons :: Monad m => Stream m Int -> m ()
171135uncons s = do
172136 r <- S.uncons s
173137 case r of
174138 Nothing -> return ()
175139 Just (_, t) -> uncons t
140+
141+{-# INLINE init #-}
142+init :: Monad m => Stream m a -> m ()
143+init s = do
144+ r <- S.init s
145+ case r of
146+ Nothing -> return ()
147+ Just x -> S.runStream x
148+
149+{-# INLINE tail #-}
150+tail :: Monad m => Stream m a -> m ()
151+tail s = do
152+ r <- S.tail s
153+ case r of
154+ Nothing -> return ()
155+ Just x -> tail x
156+
157+{-# INLINE nullHeadTail #-}
158+nullHeadTail :: Monad m => Stream m Int -> m ()
176159nullHeadTail s = do
177160 r <- S.null s
178161 if not r
@@ -183,17 +166,25 @@ nullHeadTail s = do
183166 Nothing -> return ()
184167 Just x -> nullHeadTail x
185168 else return ()
169+
186170mapM_ = S.mapM_ (\_ -> return ())
187171toList = S.toList
188172foldr = S.foldr (:) []
173+foldr1 = S.foldr1 (+)
189174foldrM = S.foldrM (\a xs -> return (a : xs)) []
190−foldl = S.foldl' (+) 0
175+foldl' = S.foldl' (+) 0
176+foldl1' = S.foldl1' (+)
191177last = S.last
192178elem = S.elem maxValue
193179notElem = S.notElem maxValue
194180length = S.length
195181all = S.all (<= maxValue)
196182any = S.any (> maxValue)
183+and = S.and . S.map (<= maxValue)
184+or = S.or . S.map (> maxValue)
185+find = S.find (== maxValue)
186+findIndex = S.findIndex (== maxValue)
187+elemIndex = S.elemIndex maxValue
197188maximum = S.maximum
198189minimum = S.minimum
199190sum = S.sum
@@ -207,6 +198,41 @@ product = S.product
207198transform :: Monad m => Stream m a -> m ()
208199transform = runStream
209200
201+{-# INLINE scan #-}
202+{-# INLINE mapM_ #-}
203+{-# INLINE map #-}
204+{-# INLINE fmap #-}
205+{-# INLINE mapMaybe #-}
206+{-# INLINE filterEven #-}
207+{-# INLINE filterAllOut #-}
208+{-# INLINE filterAllIn #-}
209+{-# INLINE takeOne #-}
210+{-# INLINE takeAll #-}
211+{-# INLINE takeWhileTrue #-}
212+{-# INLINE takeWhileMTrue #-}
213+{-# INLINE dropAll #-}
214+{-# INLINE dropWhileTrue #-}
215+{-# INLINE dropWhileMTrue #-}
216+{-# INLINE findIndices #-}
217+{-# INLINE elemIndices #-}
218+scan, mapM_, map, fmap, mapMaybe, filterEven, filterAllOut,
219+ filterAllIn, takeOne, takeAll, takeWhileTrue, takeWhileMTrue, dropAll,
220+ dropWhileTrue, dropWhileMTrue,
221+ findIndices, elemIndices
222+ :: Monad m
223+ => Stream m Int -> m ()
224+
225+{-# INLINE mapMaybeM #-}
226+mapMaybeM :: S.MonadAsync m => Stream m Int -> m ()
227+
228+{-# INLINE mapM #-}
229+mapM :: (S.IsStream t, S.MonadAsync m)
230+ => (t m Int -> S.SerialT m Int) -> t m Int -> m ()
231+
232+{-# INLINE sequence #-}
233+sequence :: (S.IsStream t, S.MonadAsync m)
234+ => (t m Int -> S.SerialT m Int) -> t m (m Int) -> m ()
235+
210236scan = transform . S.scanl' (+) 0
211237fmap = transform . Prelude.fmap (+1)
212238map = transform . S.map (+1)
@@ -226,11 +252,22 @@ takeWhileMTrue = transform . S.takeWhileM (return . (<= maxValue))
226252dropAll = transform . S.drop maxValue
227253dropWhileTrue = transform . S.dropWhile (<= maxValue)
228254dropWhileMTrue = transform . S.dropWhileM (return . (<= maxValue))
255+findIndices = transform . S.findIndices (== maxValue)
256+elemIndices = transform . S.elemIndices maxValue
229257
230258-------------------------------------------------------------------------------
231259-- Zipping and concat
232260-------------------------------------------------------------------------------
233261
262+{-# INLINE zip #-}
263+{-# INLINE zipM #-}
264+{-# INLINE concat #-}
265+zip, zipM, concat :: Monad m => Stream m Int -> m ()
266+
267+{-# INLINE zipAsync #-}
268+{-# INLINE zipAsyncM #-}
269+zipAsync, zipAsyncM :: S.MonadAsync m => Stream m Int -> m ()
270+
234271zip src = do
235272 r <- S.tail src
236273 let src1 = fromJust r
@@ -257,6 +294,16 @@ concat _n = return ()
257294compose :: Monad m => (Stream m Int -> Stream m Int) -> Stream m Int -> m ()
258295compose f = transform . f . f . f . f
259296
297+{-# INLINE composeMapM #-}
298+{-# INLINE composeAllInFilters #-}
299+{-# INLINE composeAllOutFilters #-}
300+{-# INLINE composeMapAllInFilter #-}
301+composeAllInFilters, composeAllOutFilters,
302+ composeMapAllInFilter
303+ :: Monad m
304+ => Stream m Int -> m ()
305+composeMapM :: S.MonadAsync m => Stream m Int -> m ()
306+
260307composeMapM = compose (S.mapM return)
261308composeAllInFilters = compose (S.filter (<= maxValue))
262309composeAllOutFilters = compose (S.filter (> maxValue))
benchmark/StreamKOps.hs+18
@@ -130,6 +130,24 @@ uncons s = do
130130 Nothing -> return ()
131131 Just (_, t) -> uncons t
132132
133+{-# INLINE init #-}
134+init :: (Monad m, S.IsStream t) => t m a -> m ()
135+init s = do
136+ r <- S.init s
137+ case r of
138+ Nothing -> return ()
139+ Just x -> S.runStream x
140+
141+{-# INLINE tail #-}
142+tail :: (Monad m, S.IsStream t) => t m a -> m ()
143+tail s = do
144+ r <- S.tail s
145+ case r of
146+ Nothing -> return ()
147+ Just x -> tail x
148+
149+-- | If the stream is not null get its head and tail and then do the same to
150+-- the tail.
133151nullHeadTail s = do
134152 r <- S.null s
135153 if not r
src/Streamly/Prelude.hs+3−3
@@ -498,14 +498,14 @@ foldl' step begin m = S.foldl' step begin $ toStreamS m
498498
499499-- | Strict left fold, for non-empty streams, using first element as the
500500-- starting value. Returns 'Nothing' if the stream is empty.
501−foldl1' :: Monad m => (a -> a -> a) -> (a -> b) -> SerialT m a -> m (Maybe b)
502−foldl1' step done m = do
501+foldl1' :: Monad m => (a -> a -> a) -> SerialT m a -> m (Maybe a)
502+foldl1' step m = do
503503 r <- uncons m
504504 case r of
505505 Nothing -> return Nothing
506506 Just (h, t) -> do
507507 res <- foldl' step h t
508− return $ Just $ done res
508+ return $ Just res
509509
510510-- XXX replace the recursive "go" with explicit continuations.
511511-- | Like 'foldx', but with a monadic step function.
test/Prop.hs+12−4
@@ -471,6 +471,10 @@ transformCombineOpsOrdered constr desc t eq = do
471471 prop (desc ++ " concurrent application") $
472472 transform (& (map (+1))) t (|& (S.map (+1)))
473473
474+ prop (desc ++ " findIndices") $
475+ transform (findIndices odd) t (S.findIndices odd)
476+ prop (desc ++ " elemIndices") $
477+ transform (elemIndices 0) t (S.elemIndices 0)
474478
475479wrapMaybe :: Eq a1 => ([a1] -> a2) -> [a1] -> Maybe a2
476480wrapMaybe f =
@@ -487,14 +491,18 @@ eliminationOps
487491eliminationOps constr desc t = do
488492 -- Elimination
489493 prop (desc ++ " null") $ eliminateOp constr null $ S.null . t
490− prop (desc ++ " foldl") $
494+ prop (desc ++ " foldl'") $
491495 eliminateOp constr (foldl' (+) 0) $ (S.foldl' (+) 0) . t
492− prop (desc ++ " foldl1") $
493− eliminateOp constr (wrapMaybe $ foldl1' (+)) $ (S.foldl1' (+) id) . t
496+ prop (desc ++ " foldl1'") $
497+ eliminateOp constr (wrapMaybe $ foldl1' (+)) $ (S.foldl1' (+)) . t
494498 prop (desc ++ " foldr1") $
495499 eliminateOp constr (wrapMaybe $ foldr1 (+)) $ (S.foldr1 (+)) . t
496500 prop (desc ++ " all") $ eliminateOp constr (all even) $ (S.all even) . t
497501 prop (desc ++ " any") $ eliminateOp constr (any even) $ (S.any even) . t
502+ prop (desc ++ " and") $ eliminateOp constr (and . map (> 0)) $
503+ (S.and . S.map (> 0)) . t
504+ prop (desc ++ " or") $ eliminateOp constr (or . map (> 0)) $
505+ (S.or . S.map (> 0)) . t
498506 prop (desc ++ " length") $ eliminateOp constr length $ S.length . t
499507 prop (desc ++ " sum") $ eliminateOp constr sum $ S.sum . t
500508 prop (desc ++ " product") $ eliminateOp constr product $ S.product . t
@@ -813,7 +821,7 @@ main = hspec $ do
813821 concurrentFoldlApplication
814822
815823 -- These tests are specifically targeted towards detecting illegal sharing
816− -- of SVar across conurrent streams.
824+ -- of SVar across conurrent streams. All transform ops must be added here.
817825 describe "Stream transform and combine operations" $ do
818826 transformCombineOpsCommon S.fromFoldable "serially" serially (==)
819827 transformCombineOpsCommon S.fromFoldable "aheadly" aheadly (==)