fa8a2cf · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Merge LinearAsync.hs into Prelude.hs

fa8a2cfadithyaov7 years ago

benchmark/LinearAsync.hs+37−135

compare with master

unifiedsplit

@@ -1,4 +1,3 @@

1−{-# LANGUAGE CPP #-}

2-- |

1-- |

3-- Module : Main

2-- Module : Main

@@ -6,142 +5,45 @@

6-- License : BSD3

5-- License : BSD3

7-- Maintainer : streamly@composewell.com

6-- Maintainer : streamly@composewell.com

9−import Control.DeepSeq (NFData)

8+import Streamly.Benchmark.Common

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

9+import Streamly.Benchmark.Prelude

11−import System.Random (randomRIO)

13−import Streamly.Benchmark.Common (parseCLIOpts)

14−

15−import Streamly

16import Gauge

11import Gauge

18−import qualified Streamly.Benchmark.Prelude as Ops

19−

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 #-}

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

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

27−

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

29−{-# INLINE benchSrcIO #-}

30−benchSrcIO

31− :: (t IO a -> SerialT IO a)

32− -> String

33− -> (Int -> t IO a)

34− -> Benchmark

35−benchSrcIO t name f

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

37−

38−{-# INLINE benchMonadicSrcIO #-}

39−benchMonadicSrcIO :: String -> (Int -> IO ()) -> Benchmark

40−benchMonadicSrcIO name f = bench name $ nfIO $ randomRIO (1,1) >>= f

41−

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−-}

46−

47−defaultStreamSize :: Int

48−defaultStreamSize = 100000

49−

50main :: IO ()

13main :: IO ()

51main = do

14main = do

52− -- XXX Fix indentation

53− (value, cfg, benches) <- parseCLIOpts defaultStreamSize

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

54− let value2 = round $ sqrt $ (fromIntegral value :: Double)

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

55− value2 `seq` value `seq` runMode (mode cfg) cfg benches

17+ value `seq`

56− [ bgroup "asyncly"

18+ runMode

57− [ benchSrcIO asyncly "unfoldr" (Ops.sourceUnfoldr value)

19+ (mode cfg)

58− , benchSrcIO asyncly "unfoldrM" (Ops.sourceUnfoldrM value)

20+ cfg

59− , benchSrcIO asyncly "fromFoldable" (Ops.sourceFromFoldable value)

21+ benches

60− , benchSrcIO asyncly "fromFoldableM" (Ops.sourceFromFoldableM value)

22+ (concat

61− , benchSrcIO asyncly "foldMapWith" (Ops.sourceFoldMapWith value)

23+ [ async value

62− , benchSrcIO asyncly "foldMapWithM" (Ops.sourceFoldMapWithM value)

24+ , wAsync value

63− , benchSrcIO asyncly "foldMapM" (Ops.sourceFoldMapM value)

25+ , ahead value

64− , benchIO value "map" $ Ops.map' asyncly 1

26+ , o_1_space_async_zip value

65− , benchIO value "fmap" $ Ops.fmap' asyncly 1

27+ ])

66− , benchIO value "mapM" $ Ops.mapM asyncly 1

28+ where

67− , benchSrcIO asyncly "unfoldrM maxThreads 1"

29+ async value =

68− (maxThreads 1 . Ops.sourceUnfoldrM value)

30+ concat

69− , benchSrcIO asyncly "unfoldrM maxBuffer 1 (x/10 ops)"

31+ [ o_1_space_async_generation value

70− (maxBuffer 1 . Ops.sourceUnfoldrMN (value `div` 10))

32+ , o_1_space_async_concatFoldable value

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

33+ , o_1_space_async_concatMap value

72− (Ops.concatStreamsWith async 2 (value `div` 2))

34+ , o_1_space_async_transformation value

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

35+ ]

74− (Ops.concatStreamsWith async value2 value2)

36+ wAsync value =

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

37+ concat

76− (Ops.concatStreamsWith async (value2 * 2) (value2 `div` 2))

38+ [ o_1_space_wAsync_generation value

77− ]

39+ , o_1_space_wAsync_concatFoldable value

78− , bgroup "wAsyncly"

40+ , o_1_space_wAsync_concatMap value

79− [ benchSrcIO wAsyncly "unfoldr" (Ops.sourceUnfoldr value)

41+ , o_1_space_wAsync_transformation value

80− , benchSrcIO wAsyncly "unfoldrM" (Ops.sourceUnfoldrM value)

42+ ]

81− , benchSrcIO wAsyncly "fromFoldable" (Ops.sourceFromFoldable value)

43+ ahead value =

82− , benchSrcIO wAsyncly "fromFoldableM" (Ops.sourceFromFoldableM value)

44+ concat

83− , benchSrcIO wAsyncly "foldMapWith" (Ops.sourceFoldMapWith value)

45+ [ o_1_space_ahead_generation value

84− , benchSrcIO wAsyncly "foldMapWithM" (Ops.sourceFoldMapWithM value)

46+ , o_1_space_ahead_concatFoldable value

85− , benchSrcIO wAsyncly "foldMapM" (Ops.sourceFoldMapM value)

47+ , o_1_space_ahead_concatMap value

86− , benchIO value "map" $ Ops.map' wAsyncly 1

48+ , o_1_space_ahead_transformation value

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− ]

49+ ]

benchmark/lib/Streamly/Benchmark/Prelude.hs+294−2

@@ -36,7 +36,7 @@ import Prelude

36 (Monad, String, Int, (+), ($), (.), return, even, (>), (<=), (==), (>=),

37 subtract, undefined, Maybe(..), Bool, not, (>>=), curry,

38 maxBound, div, IO, compare, Double, fromIntegral, Integer, (<$>),

39− (<*>), flip)

39+ (<*>), flip, sqrt, round, (*), seq)

40import qualified Prelude as P

41import qualified Data.Foldable as F

42import qualified GHC.Exts as GHC

@@ -48,7 +48,22 @@ import Test.Inspection

48import qualified Streamly.Internal.Data.Stream.StreamD as D

49#endif

51−import Streamly (SerialT, IsStream, wSerially, zipSerially, serially)

51+import Streamly

52+ ( IsStream

53+ , SerialT

54+ , aheadly

55+ , asyncly

56+ , serially

57+ , wAsyncly

58+ , wSerially

59+ , zipSerially

60+ , ahead

61+ , async

62+ , wAsync

63+ , maxBuffer

64+ , maxThreads

65+ )

66+

52import qualified Streamly as S hiding (runStream)

67import qualified Streamly as S hiding (runStream)

53import qualified Streamly.Prelude as S

68import qualified Streamly.Prelude as S

54import qualified Streamly.Internal.Prelude as Internal

69import qualified Streamly.Internal.Prelude as Internal

@@ -1289,9 +1304,25 @@ benchPureSinkIO

1289benchPureSinkIO value name f =

1304benchPureSinkIO value name f =

1290 bench name $ nfIO $ randomRIO (1, 1) >>= f . sourceUnfoldr value

1305 bench name $ nfIO $ randomRIO (1, 1) >>= f . sourceUnfoldr value

1291

1306

1307+{-# INLINE benchIO #-}

1292benchIO :: (NFData b) => String -> (Int -> IO b) -> Benchmark

1308benchIO :: (NFData b) => String -> (Int -> IO b) -> Benchmark

1293benchIO name f = bench name $ nfIO $ randomRIO (1,1) >>= f

1309benchIO name f = bench name $ nfIO $ randomRIO (1,1) >>= f

1294

1310

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

1312+{-# INLINE benchSrcIO #-}

1313+benchSrcIO

1314+ :: (t IO a -> SerialT IO a)

1315+ -> String

1316+ -> (Int -> t IO a)

1317+ -> Benchmark

1318+benchSrcIO t name f

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

1320+

1321+{-# INLINE benchMonadicSrcIO #-}

1322+benchMonadicSrcIO :: String -> (Int -> IO ()) -> Benchmark

1323+benchMonadicSrcIO name f = bench name $ nfIO $ randomRIO (1,1) >>= f

1324+

1325+

1295-------------------------------------------------------------------------------

1326-------------------------------------------------------------------------------

1296-- Serial : O(1) Space

1327-- Serial : O(1) Space

1297-------------------------------------------------------------------------------

1328-------------------------------------------------------------------------------

@@ -2192,3 +2223,264 @@ o_n_stack_serial_iterated value =

2192 ]

2223 ]

2193 ]

2224 ]

2194 ]

2225 ]

2226+

2227+o_1_space_async_generation :: Int -> [Benchmark]

2228+o_1_space_async_generation value =

2229+ [ bgroup

2230+ "asyncly"

2231+ [ bgroup

2232+ "generation"

2233+ [ benchSrcIO asyncly "unfoldr" (sourceUnfoldr value)

2234+ , benchSrcIO asyncly "unfoldrM" (sourceUnfoldrM value)

2235+ , benchSrcIO asyncly "fromFoldable" (sourceFromFoldable value)

2236+ , benchSrcIO asyncly "fromFoldableM" (sourceFromFoldableM value)

2237+ , benchSrcIO

2238+ asyncly

2239+ "unfoldrM maxThreads 1"

2240+ (maxThreads 1 . sourceUnfoldrM value)

2241+ , benchSrcIO

2242+ asyncly

2243+ "unfoldrM maxBuffer 1 (x/10 ops)"

2244+ (maxBuffer 1 . sourceUnfoldrMN (value `div` 10))

2245+ ]

2246+ ]

2247+ ]

2248+

2249+o_1_space_async_concatFoldable :: Int -> [Benchmark]

2250+o_1_space_async_concatFoldable value =

2251+ [ bgroup

2252+ "asyncly"

2253+ [ bgroup

2254+ "concat-foldable"

2255+ [ benchSrcIO asyncly "foldMapWith" (sourceFoldMapWith value)

2256+ , benchSrcIO

2257+ asyncly

2258+ "foldMapWithM"

2259+ (sourceFoldMapWithM value)

2260+ , benchSrcIO asyncly "foldMapM" (sourceFoldMapM value)

2261+ ]

2262+ ]

2263+ ]

2264+

2265+

2266+o_1_space_async_concatMap :: Int -> [Benchmark]

2267+o_1_space_async_concatMap value =

2268+ value2 `seq`

2269+ [ bgroup

2270+ "asyncly"

2271+ [ bgroup

2272+ "concatMap"

2273+ [ benchMonadicSrcIO

2274+ "concatMapWith (2,x/2)"

2275+ (concatStreamsWith async 2 (value `div` 2))

2276+ , benchMonadicSrcIO

2277+ "concatMapWith (sqrt x,sqrt x)"

2278+ (concatStreamsWith async value2 value2)

2279+ , benchMonadicSrcIO

2280+ "concatMapWith (sqrt x * 2,sqrt x / 2)"

2281+ (concatStreamsWith async (value2 * 2) (value2 `div` 2))

2282+ ]

2283+ ]

2284+ ]

2285+ where

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

2287+

2288+

2289+o_1_space_async_transformation :: Int -> [Benchmark]

2290+o_1_space_async_transformation value =

2291+ [ bgroup

2292+ "asyncly"

2293+ [ bgroup

2294+ "transformation"

2295+ [ benchIOSink value "map" $ map' asyncly 1

2296+ , benchIOSink value "fmap" $ fmap' asyncly 1

2297+ , benchIOSink value "mapM" $ mapM asyncly 1

2298+ ]

2299+ ]

2300+ ]

2301+

2302+o_1_space_wAsync_generation :: Int -> [Benchmark]

2303+o_1_space_wAsync_generation value =

2304+ [ bgroup

2305+ "wAsyncly"

2306+ [ bgroup

2307+ "generation"

2308+ [ benchSrcIO wAsyncly "unfoldr" (sourceUnfoldr value)

2309+ , benchSrcIO wAsyncly "unfoldrM" (sourceUnfoldrM value)

2310+ , benchSrcIO wAsyncly "fromFoldable" (sourceFromFoldable value)

2311+ , benchSrcIO

2312+ wAsyncly

2313+ "fromFoldableM"

2314+ (sourceFromFoldableM value)

2315+ , benchSrcIO

2316+ wAsyncly

2317+ "unfoldrM maxThreads 1"

2318+ (maxThreads 1 . sourceUnfoldrM value)

2319+ , benchSrcIO

2320+ wAsyncly

2321+ "unfoldrM maxBuffer 1 (x/10 ops)"

2322+ (maxBuffer 1 . sourceUnfoldrMN (value `div` 10))

2323+ ]

2324+ ]

2325+ ]

2326+

2327+o_1_space_wAsync_concatFoldable :: Int -> [Benchmark]

2328+o_1_space_wAsync_concatFoldable value =

2329+ [ bgroup

2330+ "wAsyncly"

2331+ [ bgroup

2332+ "concat-foldable"

2333+ [ benchSrcIO wAsyncly "foldMapWith" (sourceFoldMapWith value)

2334+ , benchSrcIO wAsyncly "foldMapWithM" (sourceFoldMapWithM value)

2335+ , benchSrcIO wAsyncly "foldMapM" (sourceFoldMapM value)

2336+ ]

2337+ ]

2338+ ]

2339+

2340+-- When we merge streams using wAsync the size of the queue increases

2341+-- slowly because of the binary composition adding just one more item

2342+-- to the work queue only after every scheduling pass through the

2343+-- work queue.

2344+--

2345+-- We should see the memory consumption increasing slowly if these

2346+-- benchmarks are left to run on infinite number of streams of infinite

2347+-- sizes.

2348+o_1_space_wAsync_concatMap :: Int -> [Benchmark]

2349+o_1_space_wAsync_concatMap value =

2350+ value2 `seq`

2351+ [ bgroup

2352+ "wAsyncly"

2353+ [ benchMonadicSrcIO

2354+ "concatMapWith (2,x/2)"

2355+ (concatStreamsWith wAsync 2 (value `div` 2))

2356+ , benchMonadicSrcIO

2357+ "concatMapWith (sqrt x,sqrt x)"

2358+ (concatStreamsWith wAsync value2 value2)

2359+ , benchMonadicSrcIO

2360+ "concatMapWith (sqrt x * 2,sqrt x / 2)"

2361+ (concatStreamsWith wAsync (value2 * 2) (value2 `div` 2))

2362+ ]

2363+ ]

2364+ where

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

2366+

2367+

2368+o_1_space_wAsync_transformation :: Int -> [Benchmark]

2369+o_1_space_wAsync_transformation value =

2370+ [ bgroup

2371+ "wAsyncly"

2372+ [ bgroup

2373+ "transformation"

2374+ [ benchIOSink value "map" $ map' wAsyncly 1

2375+ , benchIOSink value "fmap" $ fmap' wAsyncly 1

2376+ , benchIOSink value "mapM" $ mapM wAsyncly 1

2377+ ]

2378+ ]

2379+ ]

2380+

2381+-- unfoldr and fromFoldable are always serial and therefore the same for

2382+-- all stream types. They can be removed to reduce the number of benchmarks.

2383+o_1_space_ahead_generation :: Int -> [Benchmark]

2384+o_1_space_ahead_generation value =

2385+ [ bgroup

2386+ "aheadly"

2387+ [ bgroup

2388+ "generation"

2389+ [ benchSrcIO aheadly "unfoldr" (sourceUnfoldr value)

2390+ , benchSrcIO aheadly "unfoldrM" (sourceUnfoldrM value)

2391+-- , benchSrcIO aheadly "fromFoldable" (sourceFromFoldable value)

2392+ , benchSrcIO

2393+ aheadly

2394+ "fromFoldableM"

2395+ (sourceFromFoldableM value)

2396+ , benchSrcIO

2397+ aheadly

2398+ "unfoldrM maxThreads 1"

2399+ (maxThreads 1 . sourceUnfoldrM value)

2400+ , benchSrcIO

2401+ aheadly

2402+ "unfoldrM maxBuffer 1 (x/10 ops)"

2403+ (maxBuffer 1 . sourceUnfoldrMN (value `div` 10))

2404+ ]

2405+ ]

2406+ ]

2407+

2408+o_1_space_ahead_concatFoldable :: Int -> [Benchmark]

2409+o_1_space_ahead_concatFoldable value =

2410+ [ bgroup

2411+ "aheadly"

2412+ [ bgroup

2413+ "concat-foldable"

2414+ [ benchSrcIO aheadly "foldMapWith" (sourceFoldMapWith value)

2415+ , benchSrcIO aheadly "foldMapWithM" (sourceFoldMapWithM value)

2416+ , benchSrcIO aheadly "foldMapM" (sourceFoldMapM value)

2417+ ]

2418+ ]

2419+ ]

2420+

2421+o_1_space_ahead_concatMap :: Int -> [Benchmark]

2422+o_1_space_ahead_concatMap value =

2423+ value2 `seq`

2424+ [ bgroup

2425+ "aheadly"

2426+ [ benchMonadicSrcIO

2427+ "concatMapWith (2,x/2)"

2428+ (concatStreamsWith ahead 2 (value `div` 2))

2429+ , benchMonadicSrcIO

2430+ "concatMapWith (sqrt x,sqrt x)"

2431+ (concatStreamsWith ahead value2 value2)

2432+ , benchMonadicSrcIO

2433+ "concatMapWith (sqrt x * 2,sqrt x / 2)"

2434+ (concatStreamsWith ahead (value2 * 2) (value2 `div` 2))

2435+ ]

2436+ ]

2437+ where

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

2439+

2440+

2441+o_1_space_ahead_transformation :: Int -> [Benchmark]

2442+o_1_space_ahead_transformation value =

2443+ [ bgroup

2444+ "aheadly"

2445+ [ bgroup

2446+ "transformation"

2447+ [ benchIOSink value "map" $ map' aheadly 1

2448+ , benchIOSink value "fmap" $ fmap' aheadly 1

2449+ , benchIOSink value "mapM" $ mapM aheadly 1

2450+ ]

2451+ ]

2452+ ]

2453+

2454+o_1_space_async_zip :: Int -> [Benchmark]

2455+o_1_space_async_zip value =

2456+ [ bgroup

2457+ "asyncly"

2458+ [ bgroup

2459+ "zip"

2460+ [ benchSrcIO

2461+ serially

2462+ "zipAsync (2,x/2)"

2463+ (zipAsync (value `div` 2))

2464+ , benchSrcIO

2465+ serially

2466+ "zipAsyncM (2,x/2)"

2467+ (zipAsyncM (value `div` 2))

2468+ , benchSrcIO

2469+ serially

2470+ "zipAsyncAp (2,x/2)"

2471+ (zipAsyncAp (value `div` 2))

2472+ , benchIOSink value "fmap zipAsyncly" $ fmap' S.zipAsyncly 1

2473+ , benchSrcIO

2474+ serially

2475+ "mergeAsyncBy (2,x/2)"

2476+ (mergeAsyncBy (value `div` 2))

2477+ , benchSrcIO

2478+ serially

2479+ "mergeAsyncByM (2,x/2)"

2480+ (mergeAsyncByM (value `div` 2))

2481+ -- Parallel stages in a pipeline

2482+ , benchIOSink value "parAppMap" parAppMap

2483+ , benchIOSink value "parAppSum" parAppSum

2484+ ]

2485+ ]

2486+ ]