git.iko.soy/streamly
on d3b4e36 help sign in
streamly/benchmark/Streamly/Benchmark/Prelude/Parallel.hs
238 lines 8.3 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.IsStream as Internal
20import Streamly.Benchmark.Common
21import Streamly.Benchmark.Prelude
23import Gauge
25moduleName :: String
26moduleName = "Prelude.Parallel"
28-------------------------------------------------------------------------------
29-- Merging
30-------------------------------------------------------------------------------
32{-# INLINE mergeAsyncByM #-}
33mergeAsyncByM :: (S.IsStream t, S.MonadAsync m) => Int -> Int -> t m Int
34mergeAsyncByM count n =
35 S.mergeAsyncByM
36 (\a b -> return (a `compare` b))
37 (sourceUnfoldrM count n)
38 (sourceUnfoldrM count (n + 1))
40{-# INLINE mergeAsyncBy #-}
41mergeAsyncBy :: (S.IsStream t, S.MonadAsync m) => Int -> Int -> t m Int
42mergeAsyncBy count n =
43 S.mergeAsyncBy
44 compare
45 (sourceUnfoldrM count n)
46 (sourceUnfoldrM count (n + 1))
48-------------------------------------------------------------------------------
49-- Application/fold
50-------------------------------------------------------------------------------
52{-# INLINE parAppMap #-}
53parAppMap :: S.MonadAsync m => SerialT m Int -> m ()
54parAppMap src = S.drain $ S.map (+1) S.|$ src
56{-# INLINE parAppSum #-}
57parAppSum :: S.MonadAsync m => SerialT m Int -> m ()
58parAppSum src = (S.sum S.|$. src) >>= \x -> seq x (return ())
60{-# INLINE (|&) #-}
61(|&) :: S.MonadAsync m => SerialT m Int -> m ()
62(|&) src = src S.|& S.map (+ 1) & S.drain
64{-# INLINE (|&.) #-}
65(|&.) :: S.MonadAsync m => SerialT m Int -> m ()
66(|&.) src = (src S.|&. S.sum) >>= \x -> seq x (return ())
68-------------------------------------------------------------------------------
69-- Tapping
70-------------------------------------------------------------------------------
72{-# INLINE tapAsyncS #-}
73tapAsyncS :: S.MonadAsync m => Int -> SerialT m Int -> m ()
74tapAsyncS n = composeN n $ Internal.tapAsyncK S.sum
76{-# INLINE tapAsync #-}
77tapAsync :: S.MonadAsync m => Int -> SerialT m Int -> m ()
78tapAsync n = composeN n $ Internal.tapAsync FL.sum
80o_1_space_merge_app_tap :: Int -> [Benchmark]
81o_1_space_merge_app_tap value =
82 [ bgroup "merge-app-tap"
83 [ benchIOSrc fromSerial "mergeAsyncBy (2,x/2)"
84 (mergeAsyncBy (value `div` 2))
85 , benchIOSrc fromSerial "mergeAsyncByM (2,x/2)"
86 (mergeAsyncByM (value `div` 2))
87 -- Parallel stages in a pipeline
88 , benchIOSink value "parAppMap" parAppMap
89 , benchIOSink value "parAppSum" parAppSum
90 , benchIOSink value "(|&)" (|&)
91 , benchIOSink value "(|&.)" (|&.)
92 , benchIOSink value "tapAsync" (tapAsync 1)
93 , benchIOSink value "tapAsyncS" (tapAsyncS 1)
94 ]
95 ]
97-------------------------------------------------------------------------------
98-- Generation
99-------------------------------------------------------------------------------
101o_n_heap_generation :: Int -> [Benchmark]
102o_n_heap_generation value =
103 [ bgroup
104 "generation"
105 [ benchIOSrc fromParallel "unfoldr" (sourceUnfoldr value)
106 , benchIOSrc fromParallel "unfoldrM" (sourceUnfoldrM value)
107 , benchIOSrc fromParallel "fromFoldable" (sourceFromFoldable value)
108 , benchIOSrc fromParallel "fromFoldableM" (sourceFromFoldableM value)
109 , benchIOSrc fromParallel "unfoldrM maxThreads 1"
110 (maxThreads 1 . sourceUnfoldrM value)
111 , benchIOSrc fromParallel "unfoldrM maxBuffer 1 (x/10 ops)"
112 (maxBuffer 1 . sourceUnfoldrM (value `div` 10))
113 ]
114 ]
116-------------------------------------------------------------------------------
117-- Mapping
118-------------------------------------------------------------------------------
120o_n_heap_mapping :: Int -> [Benchmark]
121o_n_heap_mapping value =
122 [ bgroup "mapping"
123 [ benchIOSink value "map" $ mapN fromParallel 1
124 , benchIOSink value "fmap" $ fmapN fromParallel 1
125 , benchIOSink value "mapM" $ mapM fromParallel 1 . fromSerial
126 ]
127 ]
130-------------------------------------------------------------------------------
131-- Joining
132-------------------------------------------------------------------------------
134{-# INLINE parallel2 #-}
135parallel2 :: Int -> Int -> IO ()
136parallel2 count n =
137 S.drain $
138 (sourceUnfoldrM count n) `parallel` (sourceUnfoldrM count (n + 1))
140o_1_space_joining :: Int -> [Benchmark]
141o_1_space_joining value =
142 [ bgroup "joining"
143 [ benchIOSrc1 "parallel (2 of n/2)" (parallel2 (value `div` 2))
144 ]
145 ]
147-------------------------------------------------------------------------------
148-- Concat
149-------------------------------------------------------------------------------
151o_n_heap_concatFoldable :: Int -> [Benchmark]
152o_n_heap_concatFoldable value =
153 [ bgroup
154 "concat-foldable"
155 [ benchIOSrc fromParallel "foldMapWith (<>) (List)"
156 (sourceFoldMapWith value)
157 , benchIOSrc fromParallel "foldMapWith (<>) (Stream)"
158 (sourceFoldMapWithStream value)
159 , benchIOSrc fromParallel "foldMapWithM (<>) (List)"
160 (sourceFoldMapWithM value)
161 , benchIOSrc fromSerial "S.concatFoldableWith (<>) (List)"
162 (concatFoldableWith value)
163 , benchIOSrc fromSerial "S.concatForFoldableWith (<>) (List)"
164 (concatForFoldableWith value)
165 , benchIOSrc fromParallel "foldMapM (List)" (sourceFoldMapM value)
166 ]
167 ]
169o_n_heap_concat :: Int -> [Benchmark]
170o_n_heap_concat value =
171 value2 `seq`
172 [ bgroup "concat"
173 -- This is for comparison with foldMapWith
174 [ benchIOSrc fromSerial "concatMapWithId (n of 1) (fromFoldable)"
175 (S.concatMapWith parallel id . sourceConcatMapId value)
177 , benchIO "concatMapWith (n of 1)"
178 (concatStreamsWith parallel value 1)
179 , benchIO "concatMapWith (sqrt x of sqrt x)"
180 (concatStreamsWith parallel value2 value2)
181 , benchIO "concatMapWith (1 of n)"
182 (concatStreamsWith parallel 1 value)
183 ]
184 ]
186 where
188 value2 = round $ sqrt (fromIntegral value :: Double)
190-------------------------------------------------------------------------------
191-- Monadic outer product
192-------------------------------------------------------------------------------
194o_n_heap_outerProduct :: Int -> [Benchmark]
195o_n_heap_outerProduct value =
196 [ bgroup "monad-outer-product"
197 [ benchIO "toNullAp" $ toNullAp value fromParallel
198 , benchIO "toNull" $ toNullM value fromParallel
199 , benchIO "toNull3" $ toNullM3 value fromParallel
200 , benchIO "filterAllOut" $ filterAllOutM value fromParallel
201 , benchIO "filterAllIn" $ filterAllInM value fromParallel
202 , benchIO "filterSome" $ filterSome value fromParallel
203 , benchIO "breakAfterSome" $ breakAfterSome value fromParallel
204 ]
205 ]
207o_n_space_outerProduct :: Int -> [Benchmark]
208o_n_space_outerProduct value =
209 [ bgroup "monad-outer-product"
210 [ benchIO "toList" $ toListM value fromParallel
211 -- XXX disabled due to a bug for now
212 -- , benchIO "toListSome" $ toListSome value fromParallel
213 ]
214 ]
216-------------------------------------------------------------------------------
217-- Main
218-------------------------------------------------------------------------------
220main :: IO ()
221main = runWithCLIOpts defaultStreamSize allBenchmarks
223 where
225 allBenchmarks value =
226 [ bgroup (o_1_space_prefix moduleName) $ concat
227 [ o_1_space_merge_app_tap value
228 , o_1_space_joining value
229 ]
230 , bgroup (o_n_heap_prefix moduleName) $ concat
231 [ o_n_heap_generation value
232 , o_n_heap_mapping value
233 , o_n_heap_concatFoldable value
234 , o_n_heap_concat value
235 , o_n_heap_outerProduct value
236 ]
237 , bgroup (o_n_space_prefix moduleName) (o_n_space_outerProduct value)
238 ]