git.iko.soy/streamly
on 3feeec2 help sign in
streamly/benchmark/Streamly/Benchmark/Prelude/Parallel.hs
239 lines 8.4 kB history raw
1-- |
2-- Module : Main
3-- Copyright : (c) 2018 Composewell Technologies
4--
5-- License : BSD3
6-- Maintainer : streamly@composewell.com
8{-# LANGUAGE FlexibleContexts #-}
10import Prelude hiding (mapM)
12import Data.Function ((&))
13import Streamly.Prelude
14 ( SerialT, fromParallel, parallel, fromSerial, maxBuffer, maxThreads)
16import qualified Streamly.Prelude as S
17import qualified Streamly.Internal.Data.Fold as FL
18import qualified Streamly.Internal.Data.Stream.Parallel as Par
19import qualified Streamly.Internal.Data.Stream.IsStream as Internal
21import Streamly.Benchmark.Common
22import Streamly.Benchmark.Prelude
24import Gauge
26moduleName :: String
27moduleName = "Prelude.Parallel"
29-------------------------------------------------------------------------------
30-- Merging
31-------------------------------------------------------------------------------
33{-# INLINE mergeAsyncByM #-}
34mergeAsyncByM :: (S.IsStream t, S.MonadAsync m) => Int -> Int -> t m Int
35mergeAsyncByM count n =
36 S.mergeAsyncByM
37 (\a b -> return (a `compare` b))
38 (sourceUnfoldrM count n)
39 (sourceUnfoldrM count (n + 1))
41{-# INLINE mergeAsyncBy #-}
42mergeAsyncBy :: (S.IsStream t, S.MonadAsync m) => Int -> Int -> t m Int
43mergeAsyncBy count n =
44 S.mergeAsyncBy
45 compare
46 (sourceUnfoldrM count n)
47 (sourceUnfoldrM count (n + 1))
49-------------------------------------------------------------------------------
50-- Application/fold
51-------------------------------------------------------------------------------
53{-# INLINE parAppMap #-}
54parAppMap :: S.MonadAsync m => SerialT m Int -> m ()
55parAppMap src = S.drain $ S.map (+1) S.|$ src
57{-# INLINE parAppSum #-}
58parAppSum :: S.MonadAsync m => SerialT m Int -> m ()
59parAppSum src = (S.sum S.|$. src) >>= \x -> seq x (return ())
61{-# INLINE (|&) #-}
62(|&) :: S.MonadAsync m => SerialT m Int -> m ()
63(|&) src = src S.|& S.map (+ 1) & S.drain
65{-# INLINE (|&.) #-}
66(|&.) :: S.MonadAsync m => SerialT m Int -> m ()
67(|&.) src = (src S.|&. S.sum) >>= \x -> seq x (return ())
69-------------------------------------------------------------------------------
70-- Tapping
71-------------------------------------------------------------------------------
73{-# INLINE tapAsyncS #-}
74tapAsyncS :: S.MonadAsync m => Int -> SerialT m Int -> m ()
75tapAsyncS n = composeN n $ Par.tapAsync S.sum
77{-# INLINE tapAsync #-}
78tapAsync :: S.MonadAsync m => Int -> SerialT m Int -> m ()
79tapAsync n = composeN n $ Internal.tapAsync FL.sum
81o_1_space_merge_app_tap :: Int -> [Benchmark]
82o_1_space_merge_app_tap value =
83 [ bgroup "merge-app-tap"
84 [ benchIOSrc fromSerial "mergeAsyncBy (2,x/2)"
85 (mergeAsyncBy (value `div` 2))
86 , benchIOSrc fromSerial "mergeAsyncByM (2,x/2)"
87 (mergeAsyncByM (value `div` 2))
88 -- Parallel stages in a pipeline
89 , benchIOSink value "parAppMap" parAppMap
90 , benchIOSink value "parAppSum" parAppSum
91 , benchIOSink value "(|&)" (|&)
92 , benchIOSink value "(|&.)" (|&.)
93 , benchIOSink value "tapAsync" (tapAsync 1)
94 , benchIOSink value "tapAsyncS" (tapAsyncS 1)
95 ]
96 ]
98-------------------------------------------------------------------------------
99-- Generation
100-------------------------------------------------------------------------------
102o_n_heap_generation :: Int -> [Benchmark]
103o_n_heap_generation value =
104 [ bgroup
105 "generation"
106 [ benchIOSrc fromParallel "unfoldr" (sourceUnfoldr value)
107 , benchIOSrc fromParallel "unfoldrM" (sourceUnfoldrM value)
108 , benchIOSrc fromParallel "fromFoldable" (sourceFromFoldable value)
109 , benchIOSrc fromParallel "fromFoldableM" (sourceFromFoldableM value)
110 , benchIOSrc fromParallel "unfoldrM maxThreads 1"
111 (maxThreads 1 . sourceUnfoldrM value)
112 , benchIOSrc fromParallel "unfoldrM maxBuffer 1 (x/10 ops)"
113 (maxBuffer 1 . sourceUnfoldrM (value `div` 10))
114 ]
115 ]
117-------------------------------------------------------------------------------
118-- Mapping
119-------------------------------------------------------------------------------
121o_n_heap_mapping :: Int -> [Benchmark]
122o_n_heap_mapping value =
123 [ bgroup "mapping"
124 [ benchIOSink value "map" $ mapN fromParallel 1
125 , benchIOSink value "fmap" $ fmapN fromParallel 1
126 , benchIOSink value "mapM" $ mapM fromParallel 1 . fromSerial
127 ]
128 ]
131-------------------------------------------------------------------------------
132-- Joining
133-------------------------------------------------------------------------------
135{-# INLINE parallel2 #-}
136parallel2 :: Int -> Int -> IO ()
137parallel2 count n =
138 S.drain $
139 (sourceUnfoldrM count n) `parallel` (sourceUnfoldrM count (n + 1))
141o_1_space_joining :: Int -> [Benchmark]
142o_1_space_joining value =
143 [ bgroup "joining"
144 [ benchIOSrc1 "parallel (2 of n/2)" (parallel2 (value `div` 2))
145 ]
146 ]
148-------------------------------------------------------------------------------
149-- Concat
150-------------------------------------------------------------------------------
152o_n_heap_concatFoldable :: Int -> [Benchmark]
153o_n_heap_concatFoldable value =
154 [ bgroup
155 "concat-foldable"
156 [ benchIOSrc fromParallel "foldMapWith (<>) (List)"
157 (sourceFoldMapWith value)
158 , benchIOSrc fromParallel "foldMapWith (<>) (Stream)"
159 (sourceFoldMapWithStream value)
160 , benchIOSrc fromParallel "foldMapWithM (<>) (List)"
161 (sourceFoldMapWithM value)
162 , benchIOSrc fromSerial "S.concatFoldableWith (<>) (List)"
163 (concatFoldableWith value)
164 , benchIOSrc fromSerial "S.concatForFoldableWith (<>) (List)"
165 (concatForFoldableWith value)
166 , benchIOSrc fromParallel "foldMapM (List)" (sourceFoldMapM value)
167 ]
168 ]
170o_n_heap_concat :: Int -> [Benchmark]
171o_n_heap_concat value =
172 value2 `seq`
173 [ bgroup "concat"
174 -- This is for comparison with foldMapWith
175 [ benchIOSrc fromSerial "concatMapWithId (n of 1) (fromFoldable)"
176 (S.concatMapWith parallel id . sourceConcatMapId value)
178 , benchIO "concatMapWith (n of 1)"
179 (concatStreamsWith parallel value 1)
180 , benchIO "concatMapWith (sqrt x of sqrt x)"
181 (concatStreamsWith parallel value2 value2)
182 , benchIO "concatMapWith (1 of n)"
183 (concatStreamsWith parallel 1 value)
184 ]
185 ]
187 where
189 value2 = round $ sqrt (fromIntegral value :: Double)
191-------------------------------------------------------------------------------
192-- Monadic outer product
193-------------------------------------------------------------------------------
195o_n_heap_outerProduct :: Int -> [Benchmark]
196o_n_heap_outerProduct value =
197 [ bgroup "monad-outer-product"
198 [ benchIO "toNullAp" $ toNullAp value fromParallel
199 , benchIO "toNull" $ toNullM value fromParallel
200 , benchIO "toNull3" $ toNullM3 value fromParallel
201 , benchIO "filterAllOut" $ filterAllOutM value fromParallel
202 , benchIO "filterAllIn" $ filterAllInM value fromParallel
203 , benchIO "filterSome" $ filterSome value fromParallel
204 , benchIO "breakAfterSome" $ breakAfterSome value fromParallel
205 ]
206 ]
208o_n_space_outerProduct :: Int -> [Benchmark]
209o_n_space_outerProduct value =
210 [ bgroup "monad-outer-product"
211 [ benchIO "toList" $ toListM value fromParallel
212 -- XXX disabled due to a bug for now
213 -- , benchIO "toListSome" $ toListSome value fromParallel
214 ]
215 ]
217-------------------------------------------------------------------------------
218-- Main
219-------------------------------------------------------------------------------
221main :: IO ()
222main = runWithCLIOpts defaultStreamSize allBenchmarks
224 where
226 allBenchmarks value =
227 [ bgroup (o_1_space_prefix moduleName) $ concat
228 [ o_1_space_merge_app_tap value
229 , o_1_space_joining value
230 ]
231 , bgroup (o_n_heap_prefix moduleName) $ concat
232 [ o_n_heap_generation value
233 , o_n_heap_mapping value
234 , o_n_heap_concatFoldable value
235 , o_n_heap_concat value
236 , o_n_heap_outerProduct value
237 ]
238 , bgroup (o_n_space_prefix moduleName) (o_n_space_outerProduct value)
239 ]