git.iko.soy/streamly
on 385f0fc help sign in
streamly/benchmark/LinearAsync.hs
147 lines 6.8 kB history raw
1{-# LANGUAGE CPP #-}
2-- |
3-- Module : Main
4-- Copyright : (c) 2018 Harendra Kumar
5--
6-- License : BSD3
7-- Maintainer : streamly@composewell.com
9import Control.DeepSeq (NFData)
10-- import Data.Functor.Identity (Identity, runIdentity)
11import System.Random (randomRIO)
13import Common (parseCLIOpts)
15import Streamly
16import Gauge
18import qualified Streamly.Benchmark.Prelude as Ops
20-- We need a monadic bind here to make sure that the function f does not get
21-- completely optimized out by the compiler in some cases.
22--
23-- | Takes a fold method, and uses it with a default source.
24{-# INLINE benchIO #-}
25benchIO :: (IsStream t, NFData b) => Int -> String -> (t IO Int -> IO b) -> Benchmark
26benchIO value name f = bench name $ nfIO $ randomRIO (1,1) >>= f . Ops.source value
28-- | Takes a source, and uses it with a default drain/fold method.
29{-# INLINE benchSrcIO #-}
30benchSrcIO
31 :: (t IO a -> SerialT IO a)
32 -> String
33 -> (Int -> t IO a)
34 -> Benchmark
35benchSrcIO t name f
36 = bench name $ nfIO $ randomRIO (1,1) >>= Ops.toNull t . f
38{-# INLINE benchMonadicSrcIO #-}
39benchMonadicSrcIO :: String -> (Int -> IO ()) -> Benchmark
40benchMonadicSrcIO name f = bench name $ nfIO $ randomRIO (1,1) >>= f
42{-
43_benchId :: NFData b => String -> (Ops.Stream m Int -> Identity b) -> Benchmark
44_benchId name f = bench name $ nf (runIdentity . f) (Ops.source 10)
45-}
47defaultStreamSize :: Int
48defaultStreamSize = 100000
50main :: IO ()
51main = do
52 -- XXX Fix indentation
53 (value, cfg, benches) <- parseCLIOpts defaultStreamSize
54 let value2 = round $ sqrt $ (fromIntegral value :: Double)
55 value2 `seq` value `seq` runMode (mode cfg) cfg benches
56 [ bgroup "asyncly"
57 [ benchSrcIO asyncly "unfoldr" (Ops.sourceUnfoldr value)
58 , benchSrcIO asyncly "unfoldrM" (Ops.sourceUnfoldrM value)
59 , benchSrcIO asyncly "fromFoldable" (Ops.sourceFromFoldable value)
60 , benchSrcIO asyncly "fromFoldableM" (Ops.sourceFromFoldableM value)
61 , benchSrcIO asyncly "foldMapWith" (Ops.sourceFoldMapWith value)
62 , benchSrcIO asyncly "foldMapWithM" (Ops.sourceFoldMapWithM value)
63 , benchSrcIO asyncly "foldMapM" (Ops.sourceFoldMapM value)
64 , benchIO value "map" $ Ops.map' asyncly 1
65 , benchIO value "fmap" $ Ops.fmap' asyncly 1
66 , benchIO value "mapM" $ Ops.mapM asyncly 1
67 , benchSrcIO asyncly "unfoldrM maxThreads 1"
68 (maxThreads 1 . Ops.sourceUnfoldrM value)
69 , benchSrcIO asyncly "unfoldrM maxBuffer 1 (x/10 ops)"
70 (maxBuffer 1 . Ops.sourceUnfoldrMN (value `div` 10))
71 , benchMonadicSrcIO "concatMapWith (2,x/2)"
72 (Ops.concatStreamsWith async 2 (value `div` 2))
73 , benchMonadicSrcIO "concatMapWith (sqrt x,sqrt x)"
74 (Ops.concatStreamsWith async value2 value2)
75 , benchMonadicSrcIO "concatMapWith (sqrt x * 2,sqrt x / 2)"
76 (Ops.concatStreamsWith async (value2 * 2) (value2 `div` 2))
77 ]
78 , bgroup "wAsyncly"
79 [ benchSrcIO wAsyncly "unfoldr" (Ops.sourceUnfoldr value)
80 , benchSrcIO wAsyncly "unfoldrM" (Ops.sourceUnfoldrM value)
81 , benchSrcIO wAsyncly "fromFoldable" (Ops.sourceFromFoldable value)
82 , benchSrcIO wAsyncly "fromFoldableM" (Ops.sourceFromFoldableM value)
83 , benchSrcIO wAsyncly "foldMapWith" (Ops.sourceFoldMapWith value)
84 , benchSrcIO wAsyncly "foldMapWithM" (Ops.sourceFoldMapWithM value)
85 , benchSrcIO wAsyncly "foldMapM" (Ops.sourceFoldMapM value)
86 , benchIO value "map" $ Ops.map' wAsyncly 1
87 , benchIO value "fmap" $ Ops.fmap' wAsyncly 1
88 , benchIO value "mapM" $ Ops.mapM wAsyncly 1
89 , benchSrcIO wAsyncly "unfoldrM maxThreads 1"
90 (maxThreads 1 . Ops.sourceUnfoldrM value)
91 , benchSrcIO wAsyncly "unfoldrM maxBuffer 1 (x/10 ops)"
92 (maxBuffer 1 . Ops.sourceUnfoldrMN (value `div` 10))
93 -- When we merge streams using wAsync the size of the queue increases
94 -- slowly because of the binary composition adding just one more item
95 -- to the work queue only after every scheduling pass through the
96 -- work queue.
97 --
98 -- We should see the memory consumption increasing slowly if these
99 -- benchmarks are left to run on infinite number of streams of infinite
100 -- sizes.
101 , benchMonadicSrcIO "concatMapWith (2,x/2)"
102 (Ops.concatStreamsWith wAsync 2 (value `div` 2))
103 , benchMonadicSrcIO "concatMapWith (sqrt x,sqrt x)"
104 (Ops.concatStreamsWith wAsync value2 value2)
105 , benchMonadicSrcIO "concatMapWith (sqrt x * 2,sqrt x / 2)"
106 (Ops.concatStreamsWith wAsync (value2 * 2) (value2 `div` 2))
107 ]
108 -- unfoldr and fromFoldable are always serial and therefore the same for
109 -- all stream types.
110 , bgroup "aheadly"
111 [ benchSrcIO aheadly "unfoldr" (Ops.sourceUnfoldr value)
112 , benchSrcIO aheadly "unfoldrM" (Ops.sourceUnfoldrM value)
113 , benchSrcIO aheadly "fromFoldableM" (Ops.sourceFromFoldableM value)
114 , benchSrcIO aheadly "foldMapWith" (Ops.sourceFoldMapWith value)
115 , benchSrcIO aheadly "foldMapWithM" (Ops.sourceFoldMapWithM value)
116 , benchSrcIO aheadly "foldMapM" (Ops.sourceFoldMapM value)
117 , benchIO value "map" $ Ops.map' aheadly 1
118 , benchIO value "fmap" $ Ops.fmap' aheadly 1
119 , benchIO value "mapM" $ Ops.mapM aheadly 1
120 , benchSrcIO aheadly "unfoldrM maxThreads 1"
121 (maxThreads 1 . Ops.sourceUnfoldrM value)
122 , benchSrcIO aheadly "unfoldrM maxBuffer 1 (x/10 ops)"
123 (maxBuffer 1 . Ops.sourceUnfoldrMN (value `div` 10))
124 , benchSrcIO aheadly "fromFoldable" (Ops.sourceFromFoldable value)
125 , benchMonadicSrcIO "concatMapWith (2,x/2)"
126 (Ops.concatStreamsWith ahead 2 (value `div` 2))
127 , benchMonadicSrcIO "concatMapWith (sqrt x,sqrt x)"
128 (Ops.concatStreamsWith ahead value2 value2)
129 , benchMonadicSrcIO "concatMapWith (sqrt x * 2,sqrt x / 2)"
130 (Ops.concatStreamsWith ahead (value2 * 2) (value2 `div` 2))
131 ]
132 , bgroup "zip"
133 [ benchSrcIO serially "zipAsync (2,x/2)" (Ops.zipAsync (value `div` 2))
134 , benchSrcIO serially "zipAsyncM (2,x/2)"
135 (Ops.zipAsyncM (value `div` 2))
136 , benchSrcIO serially "zipAsyncAp (2,x/2)"
137 (Ops.zipAsyncAp (value `div` 2))
138 , benchIO value "fmap zipAsyncly" $ Ops.fmap' zipAsyncly 1
139 , benchSrcIO serially "mergeAsyncBy (2,x/2)"
140 (Ops.mergeAsyncBy (value `div` 2))
141 , benchSrcIO serially "mergeAsyncByM (2,x/2)"
142 (Ops.mergeAsyncByM (value `div` 2))
143 -- Parallel stages in a pipeline
144 , benchIO value "parAppMap" Ops.parAppMap
145 , benchIO value "parAppSum" Ops.parAppSum
146 ]
147 ]