f5c55df · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

separate "parallel" style benchmarks

f5c55dfHarendra Kumar7 years ago

benchmark/LinearAsync.hs−22

compare with master

unifiedsplit

@@ -129,28 +129,6 @@ main = do

129 , benchMonadicSrcIO "concatMapWith (sqrt x * 2,sqrt x / 2)"

130 (Ops.concatStreamsWith ahead (value2 * 2) (value2 `div` 2))

131 ]

132− -- XXX need to use smaller streams to finish in reasonable time

133− , bgroup "parallely"

134− [ -- unfoldr is pure and works serially irrespective of the stream type

135− benchSrcIO parallely "unfoldr" (Ops.sourceUnfoldr value)

136− , benchSrcIO parallely "unfoldrM" (Ops.sourceUnfoldrM value)

137− , benchSrcIO parallely "fromFoldable" (Ops.sourceFromFoldable value)

138− , benchSrcIO parallely "fromFoldableM" (Ops.sourceFromFoldableM value)

139− , benchSrcIO parallely "foldMapWith" (Ops.sourceFoldMapWith value)

140− , benchSrcIO parallely "foldMapWithM" (Ops.sourceFoldMapWithM value)

141− , benchSrcIO parallely "foldMapM" (Ops.sourceFoldMapM value)

142− -- map/fmap are pure and therefore no concurrency would be added on top

143− -- of what the source stream (i.e. unfoldrM) provides.

144− , benchIO value "map" $ Ops.map' parallely 1

145− , benchIO value "fmap" $ Ops.fmap' parallely 1

146− , benchIO value "mapM" $ Ops.mapM parallely 1

147− , benchMonadicSrcIO "concatMapWith (2,x/2)"

148− (Ops.concatStreamsWith parallel 2 (value `div` 2))

149− , benchMonadicSrcIO "concatMapWith (sqrt x,sqrt x)"

150− (Ops.concatStreamsWith parallel value2 value2)

151− , benchMonadicSrcIO "concatMapWith (sqrt x * 2,sqrt x / 2)"

152− (Ops.concatStreamsWith parallel (value2 * 2) (value2 `div` 2))

153− ]

154 , bgroup "zip"

132 , bgroup "zip"

155 [ benchIO value "zip" Ops.zipAsync

133 [ benchIO value "zip" Ops.zipAsync

156 , benchIO value "zipM" Ops.zipAsyncM

134 , benchIO value "zipM" Ops.zipAsyncM

benchmark/NestedConcurrent.hs−14

@@ -67,20 +67,6 @@ main = do

67 , benchIO "breakAfterSome" $ Ops.breakAfterSome linearCount wAsyncly

68 ]

70− -- XXX move this in a separate benchmark

71− , bgroup "parallely"

72− [ benchIO "toNullAp" $ Ops.toNullAp linearCount parallely

73− , benchIO "toNull" $ Ops.toNull linearCount parallely

74− , benchIO "toNull3" $ Ops.toNull3 linearCount parallely

75− -- , benchIO "toList" $ Ops.toList linearCount parallely

76− -- XXX fix thread blocked indefinitely in MVar

77− -- , benchIO "toListSome" $ Ops.toListSome linearCount parallely

78− , benchIO "filterAllOut" $ Ops.filterAllOut linearCount parallely

79− , benchIO "filterAllIn" $ Ops.filterAllIn linearCount parallely

80− , benchIO "filterSome" $ Ops.filterSome linearCount parallely

81− , benchIO "breakAfterSome" $ Ops.breakAfterSome linearCount parallely

82− ]

83−

84 , bgroup "zipAsyncly"

70 , bgroup "zipAsyncly"

85 [ benchIO "toNullAp" $ Ops.toNullAp linearCount zipAsyncly

71 [ benchIO "toNullAp" $ Ops.toNullAp linearCount zipAsyncly

86 ]

72 ]

benchmark/Parallel.hsnew file+93

@@ -0,0 +1,93 @@

1+{-# LANGUAGE CPP #-}

2+-- |

3+-- Module : Main

5+--

6+-- License : BSD3

7+-- Maintainer : streamly@composewell.com

9+import Control.DeepSeq (NFData)

10+-- import Data.Functor.Identity (Identity, runIdentity)

11+import System.Random (randomRIO)

12+

13+import Common (parseCLIOpts)

14+

15+import Streamly

16+import Gauge

17+

18+import qualified Streamly.Benchmark.Prelude as Ops

19+import qualified NestedOps as Nested

20+

21+{-# INLINE benchIONested #-}

22+benchIONested :: (NFData b) => String -> (Int -> IO b) -> Benchmark

23+benchIONested name f = bench name $ nfIO $ randomRIO (1,1) >>= f

24+

25+-- We need a monadic bind here to make sure that the function f does not get

26+-- completely optimized out by the compiler in some cases.

27+--

28+-- | Takes a fold method, and uses it with a default source.

29+{-# INLINE benchIO #-}

30+benchIO :: (IsStream t, NFData b) => Int -> String -> (t IO Int -> IO b) -> Benchmark

31+benchIO value name f = bench name $ nfIO $ randomRIO (1,1) >>= f . Ops.source value

32+

33+-- | Takes a source, and uses it with a default drain/fold method.

34+{-# INLINE benchSrcIO #-}

35+benchSrcIO

36+ :: (t IO Int -> SerialT IO Int)

37+ -> String

38+ -> (Int -> t IO Int)

39+ -> Benchmark

40+benchSrcIO t name f

41+ = bench name $ nfIO $ randomRIO (1,1) >>= Ops.toNull t . f

42+

43+defaultStreamSize :: Int

44+defaultStreamSize = 100000

45+

46+linear :: Int -> Int -> [Benchmark]

47+linear value value2 =

48+ [ -- unfoldr is pure and works serially irrespective of the stream type

49+ benchSrcIO parallely "unfoldr" (Ops.sourceUnfoldr value)

50+ , benchSrcIO parallely "unfoldrM" (Ops.sourceUnfoldrM value)

51+ , benchSrcIO parallely "fromFoldable" (Ops.sourceFromFoldable value)

52+ , benchSrcIO parallely "fromFoldableM" (Ops.sourceFromFoldableM value)

53+ , benchSrcIO parallely "foldMapWith" (Ops.sourceFoldMapWith value)

54+ , benchSrcIO parallely "foldMapWithM" (Ops.sourceFoldMapWithM value)

55+ , benchSrcIO parallely "foldMapM" (Ops.sourceFoldMapM value)

56+ -- map/fmap are pure and therefore no concurrency would be added on top

57+ -- of what the source stream (i.e. unfoldrM) provides.

58+ , benchIO value "map" $ Ops.map' parallely 1

59+ , benchIO value "fmap" $ Ops.fmap' parallely 1

60+ , benchIO value "mapM" $ Ops.mapM parallely 1

61+ , benchIONested "concatMapWith (2,x/2)"

62+ (Ops.concatStreamsWith parallel 2 (value `div` 2))

63+ , benchIONested "concatMapWith (sqrt x,sqrt x)"

64+ (Ops.concatStreamsWith parallel value2 value2)

65+ , benchIONested "concatMapWith (sqrt x * 2,sqrt x / 2)"

66+ (Ops.concatStreamsWith parallel (value2 * 2) (value2 `div` 2))

67+ ]

68+

69+nested :: Int -> [Benchmark]

70+nested value =

71+ [ benchIONested "toNullAp" $ Nested.toNullAp value parallely

72+ , benchIONested "toNull" $ Nested.toNull value parallely

73+ , benchIONested "toNull3" $ Nested.toNull3 value parallely

74+ -- , benchIO "toList" $ Ops.toList value parallely

75+ -- XXX fix thread blocked indefinitely in MVar

76+ -- , benchIO "toListSome" $ Ops.toListSome value parallely

77+ , benchIONested "filterAllOut" $ Nested.filterAllOut value parallely

78+ , benchIONested "filterAllIn" $ Nested.filterAllIn value parallely

79+ , benchIONested "filterSome" $ Nested.filterSome value parallely

80+ , benchIONested "breakAfterSome" $ Nested.breakAfterSome value parallely

81+ ]

82+

83+main :: IO ()

84+main = do

85+ (value, cfg, benches) <- parseCLIOpts defaultStreamSize

86+ let value2 = round $ sqrt $ (fromIntegral value :: Double)

87+ value2 `seq` value `seq`

88+ runMode (mode cfg) cfg benches $

89+ [ bgroup "parallelly"

90+ [ bgroup "linear" $ linear value value2

91+ , bgroup "nested" $ nested value

92+ ]

93+ ]

streamly.cabal+19

@@ -760,6 +760,25 @@ benchmark nested-concurrent

760 build-depends:

761 transformers >= 0.4 && < 0.6

762

763+benchmark parallel

764+ import: bench-options-threaded

765+ type: exitcode-stdio-1.0

766+ hs-source-dirs: benchmark

767+ main-is: Parallel.hs

768+ other-modules: Streamly.Benchmark.Prelude, NestedOps, Common

769+ build-depends:

770+ streamly

771+ , base >= 4.8 && < 5

772+ , deepseq >= 1.4.1 && < 1.5

773+ , random >= 1.0 && < 2.0

774+ , gauge >= 0.2.4 && < 0.3

775+ if impl(ghc < 8.0)

776+ build-depends:

777+ transformers >= 0.4 && < 0.6

778+ if flag(inspection)

779+ build-depends: template-haskell >= 2.14 && < 2.16

780+ , inspection-testing >= 0.4 && < 0.5

781+

763benchmark linear-rate

782benchmark linear-rate

764 import: bench-options-threaded

783 import: bench-options-threaded

765 type: exitcode-stdio-1.0

784 type: exitcode-stdio-1.0