benchmark/Streamly/Benchmark/Data/Stream/Reduce.hs · streamly

git.iko.soy/streamly

on efecb26 help sign in

streamly/benchmark/Streamly/Benchmark/Data/Stream/Reduce.hs

588 lines 19 kB history raw

1-- |

2-- Module : Stream.Reduce

4-- License : BSD-3-Clause

5-- Maintainer : streamly@composewell.com

7{-# LANGUAGE CPP #-}

8{-# LANGUAGE FlexibleContexts #-}

9{-# LANGUAGE ScopedTypeVariables #-}

10{-# LANGUAGE DeriveAnyClass #-}

11{-# LANGUAGE DeriveGeneric #-}

12{-# LANGUAGE RankNTypes #-}

14#ifdef USE_PRELUDE

15{-# OPTIONS_GHC -Wno-deprecations #-}

16#endif

18module Stream.Reduce (benchmarks) where

20import Control.DeepSeq (NFData(..))

21import Control.Monad.IO.Class (MonadIO(..))

22import Data.Monoid (Sum(..))

23import GHC.Generics (Generic)

25import qualified Streamly.Internal.Data.Refold.Type as Refold

26import qualified Streamly.Internal.Data.Fold as FL

27import qualified Stream.Common as Common

29#ifndef USE_STREAMLY_CORE

30import Data.HashMap.Strict (HashMap)

31import Data.Proxy (Proxy(..))

32import Streamly.Internal.Data.IsMap.HashMap ()

33#endif

35#ifdef USE_PRELUDE

36import Control.Monad (when)

37import qualified Streamly.Internal.Data.Stream.IsStream as S

38import qualified Streamly.Prelude as S

39import Streamly.Prelude (fromSerial)

40import Streamly.Benchmark.Prelude hiding

41 ( benchIO, benchIOSrc, sourceUnfoldrM, apDiscardFst, apDiscardSnd, apLiftA2

42 , toNullAp, monadThen, toNullM, toNullM3, filterAllInM, filterAllOutM

43 , filterSome, breakAfterSome, toListM, toListSome, transformMapM

44 , transformComposeMapM, transformTeeMapM, transformZipMapM)

45#else

47import Streamly.Internal.Data.Stream (Stream)

48import qualified Streamly.Internal.Data.Stream as S

49#ifndef USE_STREAMLY_CORE

50import qualified Streamly.Data.Stream.Prelude as S

51import qualified Streamly.Internal.Data.Stream.Time as S

52#endif

54#ifdef USE_STREAMK

55import Streamly.Internal.Data.StreamK (StreamK)

56import qualified Streamly.Internal.Data.Parser as PR

57import qualified Streamly.Internal.Data.StreamK as K

58#else

59import qualified Streamly.Internal.Data.Stream as K

60#endif

62#endif

64import Gauge

65import Streamly.Benchmark.Common

66import Stream.Common

67import Prelude hiding (reverse, tail)

69#ifdef USE_PRELUDE

70type Stream = S.SerialT

71#endif

73-- Apply transformation g count times on a stream of length len

74#ifdef USE_STREAMK

75{-# INLINE iterateSource #-}

76iterateSource ::

77 MonadAsync m

78 => (StreamK m Int -> StreamK m Int)

79 -> Int

80 -> Int

81 -> Int

82 -> StreamK m Int

83iterateSource g count len n = f count (fromStream $ sourceUnfoldrM len n)

85 where

87 f (0 :: Int) stream = stream

88 f i stream = f (i - 1) (g stream)

89#else

90{-# INLINE iterateSource #-}

91iterateSource ::

92 MonadAsync m

93 => (Stream m Int -> Stream m Int)

94 -> Int

95 -> Int

96 -> Int

97 -> Stream m Int

98iterateSource g count len n = f count (sourceUnfoldrM len n)

100 where

101

102 f (0 :: Int) stream = stream

103 f i stream = f (i - 1) (g stream)

104#endif

105

106-------------------------------------------------------------------------------

107-- Grouping transformations

108-------------------------------------------------------------------------------

109

110{-# INLINE groups #-}

111groups :: MonadIO m => Stream m Int -> m ()

112groups =

113#ifdef USE_PRELUDE

114 Common.drain . S.groups FL.drain

115#else

116 Common.drain . S.groupsWhile (==) FL.drain

117#endif

118

119{-# INLINE groupsWhileLT #-}

120groupsWhileLT :: MonadIO m => Stream m Int -> m ()

121groupsWhileLT =

122#ifdef USE_PRELUDE

123 Common.drain . S.groupsBy (>) FL.drain

124#else

125 Common.drain . S.groupsWhile (<) FL.drain

126#endif

127

128{-# INLINE groupsWhileEq #-}

129groupsWhileEq :: MonadIO m => Stream m Int -> m ()

130groupsWhileEq =

131#ifdef USE_PRELUDE

132 Common.drain . S.groupsBy (==) FL.drain

133#else

134 Common.drain . S.groupsWhile (==) FL.drain

135#endif

136

137

138{-# INLINE groupsByRollingLT #-}

139groupsByRollingLT :: MonadIO m => Stream m Int -> m ()

140groupsByRollingLT =

141#ifdef USE_PRELUDE

142 Common.drain . S.groupsByRolling (<) FL.drain

143#else

144 Common.drain . S.groupsRollingBy (<) FL.drain

145#endif

146

147{-# INLINE groupsByRollingEq #-}

148groupsByRollingEq :: MonadIO m => Stream m Int -> m ()

149groupsByRollingEq =

150#ifdef USE_PRELUDE

151 Common.drain . S.groupsByRolling (==) FL.drain

152#else

153 Common.drain . S.groupsRollingBy (==) FL.drain

154#endif

155

156{-# INLINE foldMany #-}

157foldMany :: Monad m => Stream m Int -> m ()

158foldMany =

159 Common.drain

160 . fmap getSum

161 . S.foldMany (FL.take 2 FL.mconcat)

162 . fmap Sum

163

164{-# INLINE foldManyPost #-}

165foldManyPost :: Monad m => Stream m Int -> m ()

166foldManyPost =

167 Common.drain

168 . fmap getSum

169 . S.foldManyPost (FL.take 2 FL.mconcat)

170 . fmap Sum

171

172{-# INLINE refoldMany #-}

173refoldMany :: Monad m => Stream m Int -> m ()

174refoldMany =

175 Common.drain

176 . fmap getSum

177 . S.refoldMany (Refold.take 2 Refold.sconcat) (return mempty)

178 . fmap Sum

179

180{-# INLINE foldIterateM #-}

181foldIterateM :: Monad m => Stream m Int -> m ()

182foldIterateM =

183 Common.drain

184 . fmap getSum

185 . S.foldIterateM

186 (return . FL.take 2 . FL.sconcat) (return (Sum 0))

187 . fmap Sum

188

189{-# INLINE refoldIterateM #-}

190refoldIterateM :: Monad m => Stream m Int -> m ()

191refoldIterateM =

192 Common.drain

193 . fmap getSum

194 . S.refoldIterateM

195 (Refold.take 2 Refold.sconcat) (return (Sum 0))

196 . fmap Sum

197

198#ifdef USE_STREAMK

199{-# INLINE parseBreak #-}

200parseBreak :: Monad m => StreamK m Int -> m ()

201parseBreak s = do

202 r <- K.parseDBreak PR.one s

203 case r of

204 (Left _, _) -> return ()

205 (Right _, s1) -> parseBreak s1

206#endif

207

208o_1_space_grouping :: Int -> [Benchmark]

209o_1_space_grouping value =

210 -- Buffering operations using heap proportional to group/window sizes.

211 [ bgroup "grouping"

212 [

213 benchIOSink value "groups" groups

214 , benchIOSink value "groupsWhileLT" groupsWhileLT

215 , benchIOSink value "groupsWhileEq" groupsWhileEq

216 , benchIOSink value "groupsByRollingLT" groupsByRollingLT

217 , benchIOSink value "groupsByRollingEq" groupsByRollingEq

218 ,

219

220 -- XXX parseMany/parseIterate benchmarks are in the Parser/ParserD

221 -- modules we can bring those here. chunksOf benchmarks are in

222 -- Parser/ParserD/Array.Stream/FileSystem.Handle.

223 benchIOSink value "foldMany" foldMany

224 , benchIOSink value "foldManyPost" foldManyPost

225 , benchIOSink value "refoldMany" refoldMany

226 , benchIOSink value "foldIterateM" foldIterateM

227 , benchIOSink value "refoldIterateM" refoldIterateM

228#ifdef USE_STREAMK

229 , benchIOSink value "parseBreak (recursive)" (parseBreak . fromStream)

230#endif

231

232#ifndef USE_STREAMLY_CORE

233 , benchIOSink value "classifySessionsOf (10000 buckets)"

234 (classifySessionsOf (getKey 10000))

235 , benchIOSink value "classifySessionsOf (64 buckets)"

236 (classifySessionsOf (getKey 64))

237 , benchIOSink value "classifySessionsOfHash (10000 buckets)"

238 (classifySessionsOfHash (getKey 10000))

239 , benchIOSink value "classifySessionsOfHash (64 buckets)"

240 (classifySessionsOfHash (getKey 64))

241#endif

242 ]

243 ]

244

245#ifndef USE_STREAMLY_CORE

246 where

247

248 getKey :: Int -> Int -> Int

249 getKey n = (`mod` n)

250#endif

251

252-------------------------------------------------------------------------------

253-- Size conserving transformations (reordering, buffering, etc.)

254-------------------------------------------------------------------------------

255

256#ifndef USE_PRELUDE

257{-# INLINE reverse #-}

258reverse :: MonadIO m => Int -> Stream m Int -> m ()

259reverse n = composeN n (toStream . K.reverse . fromStream)

260

261{-# INLINE reverse' #-}

262reverse' :: MonadIO m => Int -> Stream m Int -> m ()

263reverse' n = composeN n S.reverseUnbox

264#endif

265

266o_n_heap_buffering :: Int -> [Benchmark]

267o_n_heap_buffering value =

268 [ bgroup "buffered"

269 [

270#ifndef USE_PRELUDE

271 -- Reversing a stream

272 benchIOSink value "reverse" (reverse 1)

273 , benchIOSink value "reverse'" (reverse' 1)

274#else

275 benchIOSink value "mkAsync" (mkAsync fromSerial)

276#endif

277 ]

278 ]

279

280-------------------------------------------------------------------------------

281-- Grouping/Splitting

282-------------------------------------------------------------------------------

283

284#ifndef USE_STREAMLY_CORE

285{-# INLINE classifySessionsOf #-}

286classifySessionsOf :: S.MonadAsync m => (Int -> Int) -> Stream m Int -> m ()

287classifySessionsOf getKey =

288 Common.drain

289 . S.classifySessionsOf

290 (const (return False)) 3 (FL.take 10 FL.sum)

291 . S.timestamped

292 . fmap (\x -> (getKey x, x))

293

294{-# INLINE classifySessionsOfHash #-}

295classifySessionsOfHash :: S.MonadAsync m =>

296 (Int -> Int) -> Stream m Int -> m ()

297classifySessionsOfHash getKey =

298 Common.drain

299 . S.classifySessionsByGeneric

300 (Proxy :: Proxy (HashMap k))

301 1 False (const (return False)) 3 (FL.take 10 FL.sum)

302 . S.timestamped

303 . fmap (\x -> (getKey x, x))

304#endif

305

306-------------------------------------------------------------------------------

307-- Mixed Transformation

308-------------------------------------------------------------------------------

309

310{-# INLINE scanMap #-}

311scanMap :: MonadIO m => Int -> Stream m Int -> m ()

312scanMap n = composeN n $ fmap (subtract 1) . Common.scanl' (+) 0

313

314{-# INLINE dropMap #-}

315dropMap :: MonadIO m => Int -> Stream m Int -> m ()

316dropMap n = composeN n $ fmap (subtract 1) . S.drop 1

317

318{-# INLINE dropScan #-}

319dropScan :: MonadIO m => Int -> Stream m Int -> m ()

320dropScan n = composeN n $ Common.scanl' (+) 0 . S.drop 1

321

322{-# INLINE takeDrop #-}

323takeDrop :: MonadIO m => Int -> Int -> Stream m Int -> m ()

324takeDrop value n = composeN n $ S.drop 1 . S.take (value + 1)

325

326{-# INLINE takeScan #-}

327takeScan :: MonadIO m => Int -> Int -> Stream m Int -> m ()

328takeScan value n = composeN n $ Common.scanl' (+) 0 . S.take (value + 1)

329

330{-# INLINE takeMap #-}

331takeMap :: MonadIO m => Int -> Int -> Stream m Int -> m ()

332takeMap value n = composeN n $ fmap (subtract 1) . S.take (value + 1)

333

334{-# INLINE filterDrop #-}

335filterDrop :: MonadIO m => Int -> Int -> Stream m Int -> m ()

336filterDrop value n = composeN n $ S.drop 1 . S.filter (<= (value + 1))

337

338{-# INLINE filterTake #-}

339filterTake :: MonadIO m => Int -> Int -> Stream m Int -> m ()

340filterTake value n = composeN n $ S.take (value + 1) . S.filter (<= (value + 1))

341

342{-# INLINE filterScan #-}

343filterScan :: MonadIO m => Int -> Stream m Int -> m ()

344filterScan n = composeN n $ Common.scanl' (+) 0 . S.filter (<= maxBound)

345

346#ifdef USE_PRELUDE

347{-# INLINE filterScanl1 #-}

348filterScanl1 :: MonadIO m => Int -> Stream m Int -> m ()

349filterScanl1 n = composeN n $ S.scanl1' (+) . S.filter (<= maxBound)

350#endif

351

352{-# INLINE filterMap #-}

353filterMap :: MonadIO m => Int -> Int -> Stream m Int -> m ()

354filterMap value n = composeN n $ fmap (subtract 1) . S.filter (<= (value + 1))

355

356-------------------------------------------------------------------------------

357-- Scan and fold

358-------------------------------------------------------------------------------

359

360data Pair a b =

361 Pair !a !b

362 deriving (Generic, NFData)

363

364{-# INLINE sumProductFold #-}

365sumProductFold :: Monad m => Stream m Int -> m (Pair Int Int)

366sumProductFold =

367 Common.foldl' (\(Pair s p) x -> Pair (s + x) (p * x)) (Pair 0 1)

368

369{-# INLINE sumProductScan #-}

370sumProductScan :: Monad m => Stream m Int -> m (Pair Int Int)

371sumProductScan =

372 Common.foldl' (\(Pair _ p) (s0, x) -> Pair s0 (p * x)) (Pair 0 1) .

373 Common.scanl' (\(s, _) x -> (s + x, x)) (0, 0)

374

375{-# INLINE foldl'ReduceMap #-}

376foldl'ReduceMap :: Monad m => Stream m Int -> m Int

377foldl'ReduceMap = fmap (+ 1) . Common.foldl' (+) 0

378

379o_1_space_transformations_mixed :: Int -> [Benchmark]

380o_1_space_transformations_mixed value =

381 -- scanl-map and foldl-map are equivalent to the scan and fold in the foldl

382 -- library. If scan/fold followed by a map is efficient enough we may not

383 -- need monolithic implementations of these.

384 [ bgroup "mixed"

385 [ benchIOSink value "scanl-map" (scanMap 1)

386 , benchIOSink value "foldl-map" foldl'ReduceMap

387 , benchIOSink value "sum-product-fold" sumProductFold

388 , benchIOSink value "sum-product-scan" sumProductScan

389 ]

390 ]

391

392o_1_space_transformations_mixedX4 :: Int -> [Benchmark]

393o_1_space_transformations_mixedX4 value =

394 [ bgroup "mixedX4"

395 [ benchIOSink value "scan-map" (scanMap 4)

396 , benchIOSink value "drop-map" (dropMap 4)

397 , benchIOSink value "drop-scan" (dropScan 4)

398 , benchIOSink value "take-drop" (takeDrop value 4)

399 , benchIOSink value "take-scan" (takeScan value 4)

400 , benchIOSink value "take-map" (takeMap value 4)

401 , benchIOSink value "filter-drop" (filterDrop value 4)

402 , benchIOSink value "filter-take" (filterTake value 4)

403 , benchIOSink value "filter-scan" (filterScan 4)

404#ifdef USE_PRELUDE

405 , benchIOSink value "filter-scanl1" (filterScanl1 4)

406#endif

407 , benchIOSink value "filter-map" (filterMap value 4)

408 ]

409 ]

410

411-------------------------------------------------------------------------------

412-- Iterating a transformation over and over again

413-------------------------------------------------------------------------------

414

415#ifdef USE_STREAMK

416{-

417-- this is quadratic

418{-# INLINE iterateScan #-}

419iterateScan :: MonadAsync m => Int -> Int -> Int -> Stream m Int

420iterateScan count len = toStream . iterateSource (K.scanl' (+) 0) count len

421-}

422

423{-# INLINE iterateMapM #-}

424iterateMapM :: MonadAsync m => Int -> Int -> Int -> Stream m Int

425iterateMapM count len = toStream . iterateSource (K.mapM return) count len

426

427{-# INLINE iterateFilterEven #-}

428iterateFilterEven :: MonadAsync m => Int -> Int -> Int -> Stream m Int

429iterateFilterEven count len =

430 toStream . iterateSource (K.filter even) count len

431

432{-# INLINE iterateTakeAll #-}

433iterateTakeAll :: MonadAsync m => Int -> Int -> Int -> Int -> Stream m Int

434iterateTakeAll value count len =

435 toStream . iterateSource (K.take (value + 1)) count len

436

437{-# INLINE iterateDropOne #-}

438iterateDropOne :: MonadAsync m => Int -> Int -> Int -> Stream m Int

439iterateDropOne count len = toStream . iterateSource (K.drop 1) count len

440

441{-# INLINE iterateDropWhileTrue #-}

442iterateDropWhileTrue :: MonadAsync m

443 => Int -> Int -> Int -> Int -> Stream m Int

444iterateDropWhileTrue value count len =

445 toStream . iterateSource (K.dropWhile (<= (value + 1))) count len

446

447{-# INLINE iterateDropWhileFalse #-}

448iterateDropWhileFalse :: MonadAsync m

449 => Int -> Int -> Int -> Int -> Stream m Int

450iterateDropWhileFalse value count len =

451 toStream . iterateSource (K.dropWhile (> (value + 1))) count len

452

453#else

454

455-- this is quadratic

456{-# INLINE iterateScan #-}

457iterateScan :: MonadAsync m => Int -> Int -> Int -> Stream m Int

458iterateScan = iterateSource (Common.scanl' (+) 0)

459

460#ifdef USE_PRELUDE

461-- this is quadratic

462{-# INLINE iterateScanl1 #-}

463iterateScanl1 :: MonadAsync m => Int -> Int -> Int -> Stream m Int

464iterateScanl1 = iterateSource (S.scanl1' (+))

465#endif

466

467{-# INLINE iterateMapM #-}

468iterateMapM :: MonadAsync m => Int -> Int -> Int -> Stream m Int

469iterateMapM = iterateSource (S.mapM return)

470

471{-# INLINE iterateFilterEven #-}

472iterateFilterEven :: MonadAsync m => Int -> Int -> Int -> Stream m Int

473iterateFilterEven = iterateSource (S.filter even)

474

475{-# INLINE iterateTakeAll #-}

476iterateTakeAll :: MonadAsync m => Int -> Int -> Int -> Int -> Stream m Int

477iterateTakeAll value = iterateSource (S.take (value + 1))

478

479{-# INLINE iterateDropOne #-}

480iterateDropOne :: MonadAsync m => Int -> Int -> Int -> Stream m Int

481iterateDropOne = iterateSource (S.drop 1)

482

483{-# INLINE iterateDropWhileTrue #-}

484iterateDropWhileTrue :: MonadAsync m

485 => Int -> Int -> Int -> Int -> Stream m Int

486iterateDropWhileTrue value = iterateSource (S.dropWhile (<= (value + 1)))

487#endif

488

489#ifdef USE_PRELUDE

490{-# INLINE tail #-}

491tail :: Monad m => Stream m a -> m ()

492tail s = S.tail s >>= mapM_ tail

493

494{-# INLINE nullHeadTail #-}

495nullHeadTail :: Monad m => Stream m Int -> m ()

496nullHeadTail s = do

497 r <- S.null s

498 when (not r) $ do

499 _ <- S.head s

500 S.tail s >>= mapM_ nullHeadTail

501#endif

502

503-- Head recursive operations.

504o_n_stack_iterated :: Int -> [Benchmark]

505o_n_stack_iterated value = by10 `seq` by100 `seq`

506 [ bgroup "iterated"

507 [ benchIOSrc "mapM (n/10 x 10)" $ iterateMapM by10 10

508#ifndef USE_STREAMK

509 , benchIOSrc "scanl' (quadratic) (n/100 x 100)" $

510 iterateScan by100 100

511#endif

512#ifdef USE_PRELUDE

513 , benchIOSrc "scanl1' (n/10 x 10)" $ iterateScanl1 by10 10

514#endif

515 , benchIOSrc "filterEven (n/10 x 10)" $

516 iterateFilterEven by10 10

517 , benchIOSrc "takeAll (n/10 x 10)" $

518 iterateTakeAll value by10 10

519 , benchIOSrc "dropOne (n/10 x 10)" $ iterateDropOne by10 10

520#ifdef USE_STREAMK

521 , benchIOSrc "dropWhileFalse (n/10 x 10)" $

522 iterateDropWhileFalse value by10 10

523#endif

524 , benchIOSrc "dropWhileTrue (n/10 x 10)" $

525 iterateDropWhileTrue value by10 10

526#ifdef USE_PRELUDE

527 , benchIOSink value "tail" tail

528 , benchIOSink value "nullHeadTail" nullHeadTail

529#endif

530 ]

531 ]

532

533 where

534

535 by10 = value `div` 10

536 by100 = value `div` 100

537

538-------------------------------------------------------------------------------

539-- Pipes

540-------------------------------------------------------------------------------

541

542o_1_space_pipes :: Int -> [Benchmark]

543o_1_space_pipes value =

544 [ bgroup "pipes"

545 [ benchIOSink value "mapM" (transformMapM 1)

546 , benchIOSink value "compose" (transformComposeMapM 1)

547 , benchIOSink value "tee" (transformTeeMapM 1)

548#ifdef DEVBUILD

549 -- XXX this take 1 GB memory to compile

550 , benchIOSink value "zip" (transformZipMapM 1)

551#endif

552 ]

553 ]

554

555o_1_space_pipesX4 :: Int -> [Benchmark]

556o_1_space_pipesX4 value =

557 [ bgroup "pipesX4"

558 [ benchIOSink value "mapM" (transformMapM 4)

559 , benchIOSink value "compose" (transformComposeMapM 4)

560 , benchIOSink value "tee" (transformTeeMapM 4)

561#ifdef DEVBUILD

562 -- XXX this take 1 GB memory to compile

563 , benchIOSink value "zip" (transformZipMapM 4)

564#endif

565 ]

566 ]

567

568-------------------------------------------------------------------------------

569-- Main

570-------------------------------------------------------------------------------

571

572-- In addition to gauge options, the number of elements in the stream can be

573-- passed using the --stream-size option.

574--

575benchmarks :: String -> Int -> [Benchmark]

576benchmarks moduleName size =

577 [ bgroup (o_1_space_prefix moduleName) $ Prelude.concat

578 [ o_1_space_grouping size

579 , o_1_space_transformations_mixed size

580 , o_1_space_transformations_mixedX4 size

581

582 -- pipes

583 , o_1_space_pipes size

584 , o_1_space_pipesX4 size

585 ]

586 , bgroup (o_n_stack_prefix moduleName) (o_n_stack_iterated size)

587 , bgroup (o_n_heap_prefix moduleName) (o_n_heap_buffering size)

588 ]