git.iko.soy/streamly
on master help sign in
Add support for large streams in benchmarks

* Instead of using hard coded numbers scale them based on the stream size. * Add concatMapWith benchmarks for concurrent streams * Add a linear-async-long benchmark that works on streams of 10 million elements We need large streams to detect memory leak issues. Specifically, we could not figure out the concatMapWith memory leak issue without a stream of at least 10s of millions of elements. For long benchmarks we use gauge defaults so that we do not run many iterations/samples.

commit998f496af72f725d291035313e6d9c8b5a023fac authorHarendra Kumar <harendra.kumar@gmail.com> dateWed 25 Dec 2019, 21:03 (7 years ago) parent76c3f2b
4 files changed+106−51
compare with master
unifiedsplit
benchmark/Linear.hs+33−33 benchmark/LinearAsync.hs+32−5 benchmark/Streamly/Benchmark/Prelude.hs+21−13 streamly.cabal+20
benchmark/Linear.hs+33−33
@@ -368,46 +368,46 @@ main =
368368 , benchIOSink "insertBy" (Ops.insertBy 4)
369369 ]
370370 , bgroup "joining"
371− [ benchIOSrc1 "zip (2x50K)" (Ops.zip 50000)
372− , benchIOSrc1 "zipM (2x50K)" (Ops.zipM 50000)
373− , benchIOSrc1 "mergeBy (2x50K)" (Ops.mergeBy 50000)
374− , benchIOSrc1 "serial (2x50K)" (Ops.serial2 50000)
375− , benchIOSrc1 "append (2x50K)" (Ops.append2 50000)
376− , benchIOSrc1 "serial (2x2x25K)" (Ops.serial4 25000)
377− , benchIOSrc1 "append (2x2x25K)" (Ops.append4 25000)
378− , benchIOSrc1 "wSerial (2x50K)" Ops.wSerial2
379− , benchIOSrc1 "interleave (2x50K)" Ops.interleave2
380− , benchIOSrc1 "roundRobin (2x50K)" Ops.roundRobin2
371+ [ benchIOSrc1 "zip (2,x/2)" (Ops.zip (Ops.value `div` 2))
372+ , benchIOSrc1 "zipM (2,x/2)" (Ops.zipM (Ops.value `div` 2))
373+ , benchIOSrc1 "mergeBy (2,x/2)" (Ops.mergeBy (Ops.value `div` 2))
374+ , benchIOSrc1 "serial (2,x/2)" (Ops.serial2 (Ops.value `div` 2))
375+ , benchIOSrc1 "append (2,x/2)" (Ops.append2 (Ops.value `div` 2))
376+ , benchIOSrc1 "serial (2,2,x/4)" (Ops.serial4 (Ops.value `div` 4))
377+ , benchIOSrc1 "append (2,2,x/4)" (Ops.append4 (Ops.value `div` 4))
378+ , benchIOSrc1 "wSerial (2,x/2)" Ops.wSerial2
379+ , benchIOSrc1 "interleave (2,x/2)" Ops.interleave2
380+ , benchIOSrc1 "roundRobin (2,x/2)" Ops.roundRobin2
381381 ]
382382 , bgroup "concat-foldable"
383− [ benchIOSrc serially "foldMapWith (1x100K)" Ops.sourceFoldMapWith
384− , benchIOSrc serially "foldMapWithM (1x100K)" Ops.sourceFoldMapWithM
385− , benchIOSrc serially "foldMapM (1x100K)" Ops.sourceFoldMapM
386− , benchIOSrc serially "foldWithConcatMapId (1x100K)" Ops.sourceConcatMapId
383+ [ benchIOSrc serially "foldMapWith" Ops.sourceFoldMapWith
384+ , benchIOSrc serially "foldMapWithM" Ops.sourceFoldMapWithM
385+ , benchIOSrc serially "foldMapM" Ops.sourceFoldMapM
386+ , benchIOSrc serially "foldWithConcatMapId" Ops.sourceConcatMapId
387387 ]
388388 , bgroup "concat-serial"
389− [ benchIOSrc1 "concatMapPure (2x50K)" (Ops.concatMapPure 2 50000)
390− , benchIOSrc1 "concatMap (2x50K)" (Ops.concatMap 2 50000)
391− , benchIOSrc1 "concatMap (50Kx2)" (Ops.concatMap 50000 2)
392− , benchIOSrc1 "concatMapRepl (25Kx4)" Ops.concatMapRepl4xN
393− , benchIOSrc1 "concatUnfoldRepl (25Kx4)" Ops.concatUnfoldRepl4xN
394−
395− , benchIOSrc1 "concatMapWithSerial (2x50K)"
396− (Ops.concatMapWithSerial 2 50000)
397− , benchIOSrc1 "concatMapWithSerial (50Kx2)"
398− (Ops.concatMapWithSerial 50000 2)
399−
400− , benchIOSrc1 "concatMapWithAppend (2x50K)"
401− (Ops.concatMapWithAppend 2 50000)
389+ [ benchIOSrc1 "concatMapPure (2,x/2)" (Ops.concatMapPure 2 (Ops.value `div` 2))
390+ , benchIOSrc1 "concatMap (2,x/2)" (Ops.concatMap 2 (Ops.value `div` 2))
391+ , benchIOSrc1 "concatMap (x/2,2)" (Ops.concatMap (Ops.value `div` 2) 2)
392+ , benchIOSrc1 "concatMapRepl (x/4,4)" Ops.concatMapRepl4xN
393+ , benchIOSrc1 "concatUnfoldRepl (x/4,4)" Ops.concatUnfoldRepl4xN
394+
395+ , benchIOSrc1 "concatMapWithSerial (2,x/2)"
396+ (Ops.concatMapWithSerial 2 (Ops.value `div` 2))
397+ , benchIOSrc1 "concatMapWithSerial (x/2,2)"
398+ (Ops.concatMapWithSerial (Ops.value `div` 2) 2)
399+
400+ , benchIOSrc1 "concatMapWithAppend (2,x/2)"
401+ (Ops.concatMapWithAppend 2 (Ops.value `div` 2))
402402 ]
403403 , bgroup "concat-interleave"
404− [ benchIOSrc1 "concatMapWithWSerial (2x50K)"
405− (Ops.concatMapWithWSerial 2 50000)
406− , benchIOSrc1 "concatMapWithWSerial (50Kx2)"
407− (Ops.concatMapWithWSerial 50000 2)
408− , benchIOSrc1 "concatUnfoldInterleaveRepl (25Kx4)"
404+ [ benchIOSrc1 "concatMapWithWSerial (2,x/2)"
405+ (Ops.concatMapWithWSerial 2 (Ops.value `div` 2))
406+ , benchIOSrc1 "concatMapWithWSerial (x/2,2)"
407+ (Ops.concatMapWithWSerial (Ops.value `div` 2) 2)
408+ , benchIOSrc1 "concatUnfoldInterleaveRepl (x/4,4)"
409409 Ops.concatUnfoldInterleaveRepl4xN
410− , benchIOSrc1 "concatUnfoldRoundrobinRepl (25Kx4)"
410+ , benchIOSrc1 "concatUnfoldRoundrobinRepl (x/4,4)"
411411 Ops.concatUnfoldRoundrobinRepl4xN
412412 ]
413413 -- scanl-map and foldl-map are equivalent to the scan and fold in the foldl
benchmark/LinearAsync.hs+32−5
@@ -1,3 +1,4 @@
1+{-# LANGUAGE CPP #-}
12-- |
23-- Module : Main
34-- Copyright : (c) 2018 Harendra Kumar
@@ -31,6 +32,10 @@ benchSrcIO
3132benchSrcIO t name f
3233 = bench name $ nfIO $ randomRIO (1,1) >>= Ops.toNull t . f
3334
35+{-# INLINE benchMonadicSrcIO #-}
36+benchMonadicSrcIO :: String -> (Int -> IO ()) -> Benchmark
37+benchMonadicSrcIO name f = bench name $ nfIO $ randomRIO (1,1) >>= f
38+
3439{-
3540_benchId :: NFData b => String -> (Ops.Stream m Int -> Identity b) -> Benchmark
3641_benchId name f = bench name $ nf (runIdentity . f) (Ops.source 10)
@@ -38,7 +43,13 @@ _benchId name f = bench name $ nf (runIdentity . f) (Ops.source 10)
3843
3944main :: IO ()
4045main =
41− defaultMain
46+ defaultMainWith (defaultConfig
47+ { timeLimit = Just 1
48+ , minDuration = 0
49+#ifdef LONG_BENCHMARKS
50+ , includeFirstIter = True
51+#endif
52+ })
4253 [ bgroup "asyncly"
4354 [ benchSrcIO asyncly "unfoldr" Ops.sourceUnfoldr
4455 , benchSrcIO asyncly "unfoldrM" Ops.sourceUnfoldrM
@@ -52,8 +63,12 @@ main =
5263 , benchIO "mapM" $ Ops.mapM asyncly 1
5364 , benchSrcIO asyncly "unfoldrM maxThreads 1"
5465 (maxThreads 1 . Ops.sourceUnfoldrM)
55− , benchSrcIO asyncly "unfoldrM maxBuffer 1 (1000 ops)"
56− (maxBuffer 1 . Ops.sourceUnfoldrMN 1000)
66+ , benchSrcIO asyncly "unfoldrM maxBuffer 1 (x/10 ops)"
67+ (maxBuffer 1 . Ops.sourceUnfoldrMN (Ops.value `div` 10))
68+ , benchMonadicSrcIO "concatMapWith (2,x/2)"
69+ (Ops.concatStreamsWith async 2 (Ops.value `div` 2))
70+ , benchMonadicSrcIO "concatMapWith (x/2,2)"
71+ (Ops.concatStreamsWith async (Ops.value `div` 2) 2)
5772 ]
5873 , bgroup "wAsyncly"
5974 [ benchSrcIO wAsyncly "unfoldr" Ops.sourceUnfoldr
@@ -66,6 +81,10 @@ main =
6681 , benchIO "map" $ Ops.map' wAsyncly 1
6782 , benchIO "fmap" $ Ops.fmap' wAsyncly 1
6883 , benchIO "mapM" $ Ops.mapM wAsyncly 1
84+ , benchMonadicSrcIO "concatMapWith (2,x/2)"
85+ (Ops.concatStreamsWith wAsync 2 (Ops.value `div` 2))
86+ , benchMonadicSrcIO "concatMapWith (x/2,2)"
87+ (Ops.concatStreamsWith wAsync (Ops.value `div` 2) 2)
6988 ]
7089 -- unfoldr and fromFoldable are always serial and thereofore the same for
7190 -- all stream types.
@@ -81,9 +100,13 @@ main =
81100 , benchIO "mapM" $ Ops.mapM aheadly 1
82101 , benchSrcIO aheadly "unfoldrM maxThreads 1"
83102 (maxThreads 1 . Ops.sourceUnfoldrM)
84− , benchSrcIO aheadly "unfoldrM maxBuffer 1 (1000 ops)"
85− (maxBuffer 1 . Ops.sourceUnfoldrMN 1000)
103+ , benchSrcIO aheadly "unfoldrM maxBuffer 1 (x/10 ops)"
104+ (maxBuffer 1 . Ops.sourceUnfoldrMN (Ops.value `div` 10))
86105 -- , benchSrcIO aheadly "fromFoldable" Ops.sourceFromFoldable
106+ , benchMonadicSrcIO "concatMapWith (2,x/2)"
107+ (Ops.concatStreamsWith ahead 2 (Ops.value `div` 2))
108+ , benchMonadicSrcIO "concatMapWith (x/2,2)"
109+ (Ops.concatStreamsWith ahead (Ops.value `div` 2) 2)
87110 ]
88111 -- XXX need to use smaller streams to finish in reasonable time
89112 , bgroup "parallely"
@@ -104,5 +127,9 @@ main =
104127 -- Parallel stages in a pipeline
105128 , benchIO "parAppMap" Ops.parAppMap
106129 , benchIO "parAppSum" Ops.parAppSum
130+ , benchMonadicSrcIO "concatMapWith (2,x/2)"
131+ (Ops.concatStreamsWith parallel 2 (Ops.value `div` 2))
132+ , benchMonadicSrcIO "concatMapWith (x/10,10)"
133+ (Ops.concatStreamsWith parallel (Ops.value `div` 10) 10)
107134 ]
108135 ]
benchmark/Streamly/Benchmark/Prelude.hs+21−13
@@ -10,6 +10,7 @@
1010{-# LANGUAGE ScopedTypeVariables #-}
1111{-# LANGUAGE DeriveAnyClass #-}
1212{-# LANGUAGE DeriveGeneric #-}
13+{-# LANGUAGE RankNTypes #-}
1314
1415#ifdef __HADDOCK_VERSION__
1516#undef INSPECTION
@@ -53,7 +54,11 @@ import qualified Streamly.Internal.Data.Pipe as Pipe
5354import qualified Streamly.Internal.Data.Stream.Parallel as Par
5455
5556value, maxValue, value2 :: Int
56−#ifdef LINEAR_ASYNC
57+
58+-- To detect memory leak issues we need larger streams
59+#ifdef LONG_BENCHMARKS
60+value = 10000000
61+#elif defined(LINEAR_ASYNC)
5762value = 10000
5863#else
5964value = 100000
@@ -818,23 +823,29 @@ inspect $ hasNoTypeClasses 'concatMapRepl4xN
818823
819824-- concatMapWith
820825
821−{-# INLINE concatMapWithSerial #-}
822−concatMapWithSerial :: Int -> Int -> Int -> IO ()
823−concatMapWithSerial outer inner n =
824− S.drain $ S.concatMapWith S.serial
826+{-# INLINE concatStreamsWith #-}
827+concatStreamsWith
828+ :: (forall c. S.SerialT IO c -> S.SerialT IO c -> S.SerialT IO c)
829+ -> Int
830+ -> Int
831+ -> Int
832+ -> IO ()
833+concatStreamsWith op outer inner n =
834+ S.drain $ S.concatMapWith op
825835 (\_ -> sourceUnfoldrMN inner n)
826836 (sourceUnfoldrMN outer n)
827837
838+{-# INLINE concatMapWithSerial #-}
839+concatMapWithSerial :: Int -> Int -> Int -> IO ()
840+concatMapWithSerial = concatStreamsWith S.serial
841+
828842#ifdef INSPECTION
829843inspect $ hasNoTypeClasses 'concatMapWithSerial
830844#endif
831845
832846{-# INLINE concatMapWithAppend #-}
833847concatMapWithAppend :: Int -> Int -> Int -> IO ()
834−concatMapWithAppend outer inner n =
835− S.drain $ S.concatMapWith Internal.append
836− (\_ -> sourceUnfoldrMN inner n)
837− (sourceUnfoldrMN outer n)
848+concatMapWithAppend = concatStreamsWith Internal.append
838849
839850#ifdef INSPECTION
840851inspect $ hasNoTypeClasses 'concatMapWithAppend
@@ -842,10 +853,7 @@ inspect $ hasNoTypeClasses 'concatMapWithAppend
842853
843854{-# INLINE concatMapWithWSerial #-}
844855concatMapWithWSerial :: Int -> Int -> Int -> IO ()
845−concatMapWithWSerial outer inner n =
846− S.drain $ S.concatMapWith S.wSerial
847− (\_ -> sourceUnfoldrMN inner n)
848− (sourceUnfoldrMN outer n)
856+concatMapWithWSerial = concatStreamsWith S.wSerial
849857
850858#ifdef INSPECTION
851859inspect $ hasNoTypeClasses 'concatMapWithWSerial
streamly.cabal+20
@@ -598,6 +598,26 @@ benchmark linear-async
598598 build-depends: template-haskell >= 2.14 && < 2.16
599599 , inspection-testing >= 0.4 && < 0.5
600600
601+benchmark linear-async-long
602+ import: bench-options
603+ cpp-options: -DLONG_BENCHMARKS
604+ type: exitcode-stdio-1.0
605+ hs-source-dirs: benchmark
606+ main-is: LinearAsync.hs
607+ other-modules: Streamly.Benchmark.Prelude
608+ build-depends:
609+ streamly
610+ , base >= 4.8 && < 5
611+ , deepseq >= 1.4.1 && < 1.5
612+ , random >= 1.0 && < 2.0
613+ , gauge >= 0.2.4 && < 0.3
614+ if impl(ghc < 8.0)
615+ build-depends:
616+ transformers >= 0.4 && < 0.6
617+ if flag(inspection)
618+ build-depends: template-haskell >= 2.14 && < 2.16
619+ , inspection-testing >= 0.4 && < 0.5
620+
601621benchmark linear-rate
602622 import: bench-options
603623 type: exitcode-stdio-1.0