on ec94487 help sign in
576 lines 20 kB history raw
1-- |
2-- Module : Stream.Transform
3-- Copyright : (c) 2018 Composewell Technologies
4-- License : BSD-3-Clause
5-- Maintainer : streamly@composewell.com
7{-# LANGUAGE CPP #-}
8{-# LANGUAGE FlexibleContexts #-}
9{-# LANGUAGE FlexibleInstances #-}
10{-# LANGUAGE ScopedTypeVariables #-}
11{-# LANGUAGE RankNTypes #-}
13{-# OPTIONS_GHC -Wno-orphans #-}
14#ifdef USE_PRELUDE
15{-# OPTIONS_GHC -Wno-deprecations #-}
16#endif
18#ifdef __HADDOCK_VERSION__
19#undef INSPECTION
20#endif
22#ifdef INSPECTION
23{-# LANGUAGE TemplateHaskell #-}
24{-# OPTIONS_GHC -fplugin Test.Inspection.Plugin #-}
25#endif
27module Stream.Transform (benchmarks) where
29import Control.DeepSeq (NFData(..))
30import Control.Monad.IO.Class (MonadIO(..))
31import Data.Functor.Identity (Identity(..))
33import System.Random (randomRIO)
35import qualified Streamly.Internal.Data.Fold as FL
37import qualified Prelude
38import qualified Stream.Common as Common
39import qualified Streamly.Internal.Data.Fold as Fold
40import qualified Streamly.Internal.Data.Unfold as Unfold
41#ifdef USE_PRELUDE
42import qualified Streamly.Internal.Data.Stream.IsStream as Stream
43import Streamly.Internal.Data.Time.Units
44#else
45import qualified Streamly.Internal.Data.Stream.Time as Stream
46import qualified Streamly.Internal.Data.Stream as Stream
47#endif
49import Gauge
50import Streamly.Internal.Data.Stream (Stream)
51import Stream.Common hiding (scanl')
52import Streamly.Benchmark.Common
53import Prelude hiding (sequence, mapM)
55-------------------------------------------------------------------------------
56-- Pipelines (stream-to-stream transformations)
57-------------------------------------------------------------------------------
59-------------------------------------------------------------------------------
60-- one-to-one transformations
61-------------------------------------------------------------------------------
63-------------------------------------------------------------------------------
64-- Traversable Instance
65-------------------------------------------------------------------------------
67{-# INLINE traversableTraverse #-}
68traversableTraverse :: Stream Identity Int -> IO (Stream Identity Int)
69traversableTraverse = traverse return
71{-# INLINE traversableSequenceA #-}
72traversableSequenceA :: Stream Identity Int -> IO (Stream Identity Int)
73traversableSequenceA = sequenceA . Prelude.fmap return
75{-# INLINE traversableMapM #-}
76traversableMapM :: Stream Identity Int -> IO (Stream Identity Int)
77traversableMapM = Prelude.mapM return
79{-# INLINE traversableSequence #-}
80traversableSequence :: Stream Identity Int -> IO (Stream Identity Int)
81traversableSequence = Prelude.sequence . Prelude.fmap return
83{-# INLINE benchPureSinkIO #-}
84benchPureSinkIO
85 :: NFData b
86 => Int -> String -> (Stream Identity Int -> IO b) -> Benchmark
87benchPureSinkIO value name f =
88 bench name $ nfIO $ randomRIO (1, 1) >>= f . sourceUnfoldr value
90instance NFData a => NFData (Stream Identity a) where
91 {-# INLINE rnf #-}
92 rnf xs = runIdentity $ Stream.fold (Fold.foldl' (\_ x -> rnf x) ()) xs
94o_n_space_traversable :: Int -> [Benchmark]
95o_n_space_traversable value =
96 -- Buffering operations using heap proportional to number of elements.
97 [ bgroup "traversable"
98 -- Traversable instance
99 [ benchPureSinkIO value "traverse" traversableTraverse
100 , benchPureSinkIO value "sequenceA" traversableSequenceA
101 , benchPureSinkIO value "mapM" traversableMapM
102 , benchPureSinkIO value "sequence" traversableSequence
103 ]
104 ]
106-------------------------------------------------------------------------------
107-- maps and scans
108-------------------------------------------------------------------------------
110#ifdef USE_PRELUDE
111{-# INLINE scanl' #-}
112scanl' :: MonadIO m => Int -> Stream m Int -> m ()
113scanl' n = composeN n $ Stream.scanl' (+) 0
115{-# INLINE scanlM' #-}
116scanlM' :: MonadIO m => Int -> Stream m Int -> m ()
117scanlM' n = composeN n $ Stream.scanlM' (\b a -> return $ b + a) (return 0)
119{-# INLINE scanl1' #-}
120scanl1' :: MonadIO m => Int -> Stream m Int -> m ()
121scanl1' n = composeN n $ Stream.scanl1' (+)
123{-# INLINE scanl1M' #-}
124scanl1M' :: MonadIO m => Int -> Stream m Int -> m ()
125scanl1M' n = composeN n $ Stream.scanl1M' (\b a -> return $ b + a)
126#endif
128{-# INLINE scan #-}
129scan :: MonadIO m => Int -> Stream m Int -> m ()
130scan n = composeN n $ Stream.scan FL.sum
132#ifdef USE_PRELUDE
133{-# INLINE postscanl' #-}
134postscanl' :: MonadIO m => Int -> Stream m Int -> m ()
135postscanl' n = composeN n $ Stream.postscanl' (+) 0
137{-# INLINE postscanlM' #-}
138postscanlM' :: MonadIO m => Int -> Stream m Int -> m ()
139postscanlM' n = composeN n $ Stream.postscanlM' (\b a -> return $ b + a) (return 0)
140#endif
142{-# INLINE postscan #-}
143postscan :: MonadIO m => Int -> Stream m Int -> m ()
144postscan n = composeN n $ Stream.postscan FL.sum
146{-# INLINE sequence #-}
147sequence :: MonadAsync m => Stream m (m Int) -> m ()
148sequence = Common.drain . Stream.sequence
150{-# INLINE tap #-}
151tap :: MonadIO m => Int -> Stream m Int -> m ()
152tap n = composeN n $ Stream.tap FL.sum
154#ifdef USE_PRELUDE
155{-# INLINE pollCounts #-}
156pollCounts :: Int -> Stream IO Int -> IO ()
157pollCounts n =
158 composeN n (Stream.pollCounts (const True) f)
160 where
162 f = Stream.drain . Stream.rollingMap2 (-) . Stream.delayPost 1
164{-# INLINE timestamped #-}
165timestamped :: (MonadAsync m) => Stream m Int -> m ()
166timestamped = Stream.drain . Stream.timestamped
167#endif
169{-# INLINE foldrS #-}
170foldrS :: MonadIO m => Int -> Stream m Int -> m ()
171foldrS n = composeN n $ Stream.foldrS Stream.cons Stream.nil
173{-# INLINE foldrSMap #-}
174foldrSMap :: MonadIO m => Int -> Stream m Int -> m ()
175foldrSMap n = composeN n $ Stream.foldrS (\x xs -> x + 1 `Stream.cons` xs) Stream.nil
177{-# INLINE foldrT #-}
178foldrT :: MonadIO m => Int -> Stream m Int -> m ()
179foldrT n = composeN n $ Stream.foldrT Stream.cons Stream.nil
181{-# INLINE foldrTMap #-}
182foldrTMap :: MonadIO m => Int -> Stream m Int -> m ()
183foldrTMap n = composeN n $ Stream.foldrT (\x xs -> x + 1 `Stream.cons` xs) Stream.nil
185{-# INLINE trace #-}
186trace :: MonadAsync m => Int -> Stream m Int -> m ()
187trace n = composeN n $ Stream.trace return
189o_1_space_mapping :: Int -> [Benchmark]
190o_1_space_mapping value =
191 [ bgroup
192 "mapping"
193 [
194 -- Right folds
195 benchIOSink value "foldrS" (foldrS 1)
196 , benchIOSink value "foldrSMap" (foldrSMap 1)
197 , benchIOSink value "foldrT" (foldrT 1)
198 , benchIOSink value "foldrTMap" (foldrTMap 1)
200 -- Mapping
201 , benchIOSink value "map" (mapN 1)
202 , bench "sequence" $ nfIO $ randomRIO (1, 1000) >>= \n ->
203 sequence (sourceUnfoldrAction value n)
204 , benchIOSink value "mapM" (mapM 1)
205 , benchIOSink value "tap" (tap 1)
206#ifdef USE_PRELUDE
207 , benchIOSink value "pollCounts 1 second" (pollCounts 1)
208 , benchIOSink value "timestamped" timestamped
210 -- Scanning
211 , benchIOSink value "scanl'" (scanl' 1)
212 , benchIOSink value "scanl1'" (scanl1' 1)
213 , benchIOSink value "scanlM'" (scanlM' 1)
214 , benchIOSink value "scanl1M'" (scanl1M' 1)
215 , benchIOSink value "postscanl'" (postscanl' 1)
216 , benchIOSink value "postscanlM'" (postscanlM' 1)
217#endif
218 , benchIOSink value "scan" (scan 1)
219 , benchIOSink value "postscan" (postscan 1)
220 ]
221 ]
223o_1_space_mappingX4 :: Int -> [Benchmark]
224o_1_space_mappingX4 value =
225 [ bgroup "mappingX4"
226 [ benchIOSink value "map" (mapN 4)
227 , benchIOSink value "mapM" (mapM 4)
228 , benchIOSink value "trace" (trace 4)
230#ifdef USE_PRELUDE
231 , benchIOSink value "scanl'" (scanl' 4)
232 , benchIOSink value "scanl1'" (scanl1' 4)
233 , benchIOSink value "scanlM'" (scanlM' 4)
234 , benchIOSink value "scanl1M'" (scanl1M' 4)
235 , benchIOSink value "postscanl'" (postscanl' 4)
236 , benchIOSink value "postscanlM'" (postscanlM' 4)
237#endif
238 ]
239 ]
241{-# INLINE sieveScan #-}
242sieveScan :: Monad m => Stream m Int -> Stream m Int
243sieveScan =
244 Stream.mapMaybe snd
245 . Stream.scan (FL.foldlM' (\(primes, _) n -> do
246 return $
247 let ps = takeWhile (\p -> p * p <= n) primes
248 in if all (\p -> n `mod` p /= 0) ps
249 then (primes ++ [n], Just n)
250 else (primes, Nothing)) (return ([2], Just 2)))
252o_n_space_mapping :: Int -> [Benchmark]
253o_n_space_mapping value =
254 [ bgroup "mapping"
255 [ benchIO "naive prime sieve"
256 (\n -> Stream.fold FL.sum $ sieveScan $ Stream.enumerateFromTo 2 (value + n))
257 ]
258 ]
260-------------------------------------------------------------------------------
261-- Functor
262-------------------------------------------------------------------------------
264o_1_space_functor :: Int -> [Benchmark]
265o_1_space_functor value =
266 [ bgroup "Functor"
267 [ benchIOSink value "fmap" (mapN 1)
268 , benchIOSink value "fmap x 4" (mapN 4)
269 ]
270 ]
272-------------------------------------------------------------------------------
273-- Size reducing transformations (filtering)
274-------------------------------------------------------------------------------
276{-# INLINE filterEven #-}
277filterEven :: MonadIO m => Int -> Stream m Int -> m ()
278filterEven n = composeN n $ Stream.filter even
280{-# INLINE filterAllOut #-}
281filterAllOut :: MonadIO m => Int -> Int -> Stream m Int -> m ()
282filterAllOut value n = composeN n $ Stream.filter (> (value + 1))
284{-# INLINE filterAllIn #-}
285filterAllIn :: MonadIO m => Int -> Int -> Stream m Int -> m ()
286filterAllIn value n = composeN n $ Stream.filter (<= (value + 1))
288{-# INLINE filterMEven #-}
289filterMEven :: MonadIO m => Int -> Stream m Int -> m ()
290filterMEven n = composeN n $ Stream.filterM (return . even)
292{-# INLINE filterMAllOut #-}
293filterMAllOut :: MonadIO m => Int -> Int -> Stream m Int -> m ()
294filterMAllOut value n = composeN n $ Stream.filterM (\x -> return $ x > (value + 1))
296{-# INLINE filterMAllIn #-}
297filterMAllIn :: MonadIO m => Int -> Int -> Stream m Int -> m ()
298filterMAllIn value n = composeN n $ Stream.filterM (\x -> return $ x <= (value + 1))
300{-# INLINE _takeOne #-}
301_takeOne :: MonadIO m => Int -> Stream m Int -> m ()
302_takeOne n = composeN n $ Stream.take 1
304{-# INLINE takeAll #-}
305takeAll :: MonadIO m => Int -> Int -> Stream m Int -> m ()
306takeAll value n = composeN n $ Stream.take (value + 1)
308{-# INLINE takeWhileTrue #-}
309takeWhileTrue :: MonadIO m => Int -> Int -> Stream m Int -> m ()
310takeWhileTrue value n = composeN n $ Stream.takeWhile (<= (value + 1))
312{-# INLINE takeWhileMTrue #-}
313takeWhileMTrue :: MonadIO m => Int -> Int -> Stream m Int -> m ()
314takeWhileMTrue value n = composeN n $ Stream.takeWhileM (return . (<= (value + 1)))
316{-# INLINE takeInterval #-}
317takeInterval :: Double -> Int -> Stream IO Int -> IO ()
318takeInterval i n = composeN n (Stream.takeInterval i)
320-- Inspection testing is disabled for takeInterval
321-- Enable it when looking at it throughly
322#ifdef INSPECTION
323-- inspect $ hasNoType 'takeInterval ''SPEC
324-- inspect $ hasNoTypeClasses 'takeInterval
325-- inspect $ 'takeInterval `hasNoType` ''D.Step
326#endif
328{-# INLINE dropOne #-}
329dropOne :: MonadIO m => Int -> Stream m Int -> m ()
330dropOne n = composeN n $ Stream.drop 1
332{-# INLINE dropAll #-}
333dropAll :: MonadIO m => Int -> Int -> Stream m Int -> m ()
334dropAll value n = composeN n $ Stream.drop (value + 1)
336{-# INLINE dropWhileTrue #-}
337dropWhileTrue :: MonadIO m => Int -> Int -> Stream m Int -> m ()
338dropWhileTrue value n = composeN n $ Stream.dropWhile (<= (value + 1))
340{-# INLINE dropWhileMTrue #-}
341dropWhileMTrue :: MonadIO m => Int -> Int -> Stream m Int -> m ()
342dropWhileMTrue value n = composeN n $ Stream.dropWhileM (return . (<= (value + 1)))
344{-# INLINE dropWhileFalse #-}
345dropWhileFalse :: MonadIO m => Int -> Int -> Stream m Int -> m ()
346dropWhileFalse value n = composeN n $ Stream.dropWhile (> (value + 1))
348#ifdef USE_PRELUDE
349-- XXX Decide on the time interval
350{-# INLINE _intervalsOfSum #-}
351_intervalsOfSum :: MonadAsync m => Double -> Int -> Stream m Int -> m ()
352_intervalsOfSum i n = composeN n (Stream.intervalsOf i FL.sum)
353#endif
355{-# INLINE dropInterval #-}
356dropInterval :: Double -> Int -> Stream IO Int -> IO ()
357dropInterval i n = composeN n (Stream.dropInterval i)
359-- Inspection testing is disabled for dropInterval
360-- Enable it when looking at it throughly
361#ifdef INSPECTION
362-- inspect $ hasNoTypeClasses 'dropInterval
363-- inspect $ 'dropInterval `hasNoType` ''D.Step
364#endif
366{-# INLINE findIndices #-}
367findIndices :: MonadIO m => Int -> Int -> Stream m Int -> m ()
368findIndices value n = composeN n $ Stream.findIndices (== (value + 1))
370{-# INLINE elemIndices #-}
371elemIndices :: MonadIO m => Int -> Int -> Stream m Int -> m ()
372elemIndices value n = composeN n $ Stream.elemIndices (value + 1)
374{-# INLINE deleteBy #-}
375deleteBy :: MonadIO m => Int -> Int -> Stream m Int -> m ()
376deleteBy value n = composeN n $ Stream.deleteBy (>=) (value + 1)
378-- uniq . uniq == uniq, composeN 2 ~ composeN 1
379{-# INLINE uniq #-}
380uniq :: MonadIO m => Int -> Stream m Int -> m ()
381uniq n = composeN n Stream.uniq
383{-# INLINE mapMaybe #-}
384mapMaybe :: MonadIO m => Int -> Stream m Int -> m ()
385mapMaybe n =
386 composeN n $
387 Stream.mapMaybe
388 (\x ->
389 if odd x
390 then Nothing
391 else Just x)
393{-# INLINE mapMaybeM #-}
394mapMaybeM :: MonadAsync m => Int -> Stream m Int -> m ()
395mapMaybeM n =
396 composeN n $
397 Stream.mapMaybeM
398 (\x ->
399 if odd x
400 then return Nothing
401 else return $ Just x)
403o_1_space_filtering :: Int -> [Benchmark]
404o_1_space_filtering value =
405 [ bgroup "filtering"
406 [ benchIOSink value "filter-even" (filterEven 1)
407 , benchIOSink value "filter-all-out" (filterAllOut value 1)
408 , benchIOSink value "filter-all-in" (filterAllIn value 1)
410 , benchIOSink value "filterM-even" (filterMEven 1)
411 , benchIOSink value "filterM-all-out" (filterMAllOut value 1)
412 , benchIOSink value "filterM-all-in" (filterMAllIn value 1)
414 -- Trimming
415 , benchIOSink value "take-all" (takeAll value 1)
416 , benchIOSink value "takeWhile-true" (takeWhileTrue value 1)
417 -- , benchIOSink value "takeWhileM-true" (_takeWhileMTrue value 1)
418 , benchIOSink value "drop-one" (dropOne 1)
419 , benchIOSink value "drop-all" (dropAll value 1)
420 , benchIOSink value "takeInterval-all" (takeInterval 10000 1)
421 , benchIOSink value "dropInterval-all" (dropInterval 10000 1)
422 , benchIOSink value "dropWhile-true" (dropWhileTrue value 1)
423 -- , benchIOSink value "dropWhileM-true" (_dropWhileMTrue value 1)
424 , benchIOSink
425 value
426 "dropWhile-false"
427 (dropWhileFalse value 1)
428 , benchIOSink value "deleteBy" (deleteBy value 1)
430 , benchIOSink value "uniq" (uniq 1)
432 -- Map and filter
433 , benchIOSink value "mapMaybe" (mapMaybe 1)
434 , benchIOSink value "mapMaybeM" (mapMaybeM 1)
436 -- Searching (stateful map and filter)
437 , benchIOSink value "findIndices" (findIndices value 1)
438 , benchIOSink value "elemIndices" (elemIndices value 1)
439 ]
440 ]
442o_1_space_filteringX4 :: Int -> [Benchmark]
443o_1_space_filteringX4 value =
444 [ bgroup "filteringX4"
445 [ benchIOSink value "filter-even" (filterEven 4)
446 , benchIOSink value "filter-all-out" (filterAllOut value 4)
447 , benchIOSink value "filter-all-in" (filterAllIn value 4)
449 , benchIOSink value "filterM-even" (filterMEven 4)
450 , benchIOSink value "filterM-all-out" (filterMAllOut value 4)
451 , benchIOSink value "filterM-all-in" (filterMAllIn value 4)
453 -- trimming
454 , benchIOSink value "take-all" (takeAll value 4)
455 , benchIOSink value "takeWhile-true" (takeWhileTrue value 4)
456 , benchIOSink value "takeWhileM-true" (takeWhileMTrue value 4)
457 , benchIOSink value "drop-one" (dropOne 4)
458 , benchIOSink value "drop-all" (dropAll value 4)
459 , benchIOSink value "dropWhile-true" (dropWhileTrue value 4)
460 , benchIOSink value "dropWhileM-true" (dropWhileMTrue value 4)
461 , benchIOSink
462 value
463 "dropWhile-false"
464 (dropWhileFalse value 4)
465 , benchIOSink value "deleteBy" (deleteBy value 4)
467 , benchIOSink value "uniq" (uniq 4)
469 -- map and filter
470 , benchIOSink value "mapMaybe" (mapMaybe 4)
471 , benchIOSink value "mapMaybeM" (mapMaybeM 4)
473 -- searching
474 , benchIOSink value "findIndices" (findIndices value 4)
475 , benchIOSink value "elemIndices" (elemIndices value 4)
476 ]
477 ]
479-------------------------------------------------------------------------------
480-- Size increasing transformations (insertions)
481-------------------------------------------------------------------------------
483{-# INLINE intersperse #-}
484intersperse :: MonadAsync m => Int -> Int -> Stream m Int -> m ()
485intersperse value n = composeN n $ Stream.intersperse (value + 1)
487{-# INLINE intersperseM #-}
488intersperseM :: MonadAsync m => Int -> Int -> Stream m Int -> m ()
489intersperseM value n = composeN n $ Stream.intersperseM (return $ value + 1)
491{-# INLINE insertBy #-}
492insertBy :: MonadIO m => Int -> Int -> Stream m Int -> m ()
493insertBy value n = composeN n $ Stream.insertBy compare (value + 1)
495{-# INLINE interposeSuffix #-}
496interposeSuffix :: Monad m => Int -> Int -> Stream m Int -> m ()
497interposeSuffix value n =
498 composeN n $ Stream.interposeSuffix (value + 1) Unfold.identity
500{-# INLINE intercalateSuffix #-}
501intercalateSuffix :: Monad m => Int -> Int -> Stream m Int -> m ()
502intercalateSuffix value n =
503 composeN n $ Stream.intercalateSuffix Unfold.identity (value + 1)
505o_1_space_inserting :: Int -> [Benchmark]
506o_1_space_inserting value =
507 [ bgroup "inserting"
508 [ benchIOSink value "intersperse" (intersperse value 1)
509 , benchIOSink value "intersperseM" (intersperseM value 1)
510 , benchIOSink value "insertBy" (insertBy value 1)
511 , benchIOSink value "interposeSuffix" (interposeSuffix value 1)
512 , benchIOSink value "intercalateSuffix" (intercalateSuffix value 1)
513 ]
514 ]
516o_1_space_insertingX4 :: Int -> [Benchmark]
517o_1_space_insertingX4 value =
518 [ bgroup "insertingX4"
519 [ benchIOSink value "intersperse" (intersperse value 4)
520 , benchIOSink value "insertBy" (insertBy value 4)
521 ]
522 ]
524-------------------------------------------------------------------------------
525-- Indexing
526-------------------------------------------------------------------------------
528{-# INLINE indexed #-}
529indexed :: MonadIO m => Int -> Stream m Int -> m ()
530indexed n = composeN n (fmap snd . Stream.indexed)
532{-# INLINE indexedR #-}
533indexedR :: MonadIO m => Int -> Int -> Stream m Int -> m ()
534indexedR value n = composeN n (fmap snd . Stream.indexedR value)
536o_1_space_indexing :: Int -> [Benchmark]
537o_1_space_indexing value =
538 [ bgroup "indexing"
539 [ benchIOSink value "indexed" (indexed 1)
540 , benchIOSink value "indexedR" (indexedR value 1)
541 ]
542 ]
544o_1_space_indexingX4 :: Int -> [Benchmark]
545o_1_space_indexingX4 value =
546 [ bgroup "indexingx4"
547 [ benchIOSink value "indexed" (indexed 4)
548 , benchIOSink value "indexedR" (indexedR value 4)
549 ]
550 ]
552-------------------------------------------------------------------------------
553-- Main
554-------------------------------------------------------------------------------
556-- In addition to gauge options, the number of elements in the stream can be
557-- passed using the --stream-size option.
558--
559benchmarks :: String -> Int -> [Benchmark]
560benchmarks moduleName size =
561 [ bgroup (o_1_space_prefix moduleName) $ Prelude.concat
562 [ o_1_space_functor size
563 , o_1_space_mapping size
564 , o_1_space_mappingX4 size
565 , o_1_space_filtering size
566 , o_1_space_filteringX4 size
567 , o_1_space_inserting size
568 , o_1_space_insertingX4 size
569 , o_1_space_indexing size
570 , o_1_space_indexingX4 size
571 ]
572 , bgroup (o_n_space_prefix moduleName) $ Prelude.concat
573 [ o_n_space_traversable size
574 , o_n_space_mapping size
575 ]
576 ]