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