2c70b45 · streamly · git.iko.soy

benchmark/Streamly/Benchmark/Prelude/Serial/Transformation2.hs+7

compare with master

unifiedsplit

@@ -20,6 +20,7 @@ import Control.Monad.IO.Class (MonadIO(..))

20import Data.Monoid (Sum(..))

21import GHC.Generics (Generic)

22

23+import qualified Data.List as List

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

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

24import qualified Streamly.Internal.Data.Stream.IsStream as Internal

25import qualified Streamly.Internal.Data.Stream.IsStream as Internal

25import qualified Streamly.Prelude as S

26import qualified Streamly.Prelude as S

@@ -174,6 +175,10 @@ reverse n = composeN n S.reverse

174reverse' :: MonadIO m => Int -> SerialT m Int -> m ()

175reverse' :: MonadIO m => Int -> SerialT m Int -> m ()

175reverse' n = composeN n Internal.reverse'

176reverse' n = composeN n Internal.reverse'

176

177

178+{-# INLINE sortBy #-}

179+sortBy :: MonadIO m => Int -> SerialT m Int -> m ()

180+sortBy n = composeN n (Internal.sortBy compare)

181+

177o_n_heap_buffering :: Int -> [Benchmark]

182o_n_heap_buffering :: Int -> [Benchmark]

178o_n_heap_buffering value =

183o_n_heap_buffering value =

179 [ bgroup "buffered"

184 [ bgroup "buffered"

@@ -181,6 +186,8 @@ o_n_heap_buffering value =

181 -- Reversing/sorting a stream

186 -- Reversing/sorting a stream

182 benchIOSink value "reverse" (reverse 1)

187 benchIOSink value "reverse" (reverse 1)

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

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

189+ , benchIOSink value "sortBy" (sortBy 1)

190+ , bench "sort Lists" $ nf (\x -> List.sort [1..x]) value

184

191

185 , benchIOSink value "mkAsync" (mkAsync serially)

192 , benchIOSink value "mkAsync" (mkAsync serially)

186 ]

193 ]

src/Streamly/Internal/Data/Stream/IsStream/Common.hs+2

@@ -8,6 +8,8 @@

8-- Stability : experimental

9-- Portability : GHC

10--

11+-- Bottom level IsStream module that can be used by all other upper level

12+-- IsStream modules.

11

13

12module Streamly.Internal.Data.Stream.IsStream.Common

14module Streamly.Internal.Data.Stream.IsStream.Common

13 (

15 (

src/Streamly/Internal/Data/Stream/IsStream/Nesting.hs+26−2

@@ -84,7 +84,8 @@ module Streamly.Internal.Data.Stream.IsStream.Nesting

84 , concatMapM

85 , concatMapWith

86 , concatSmapMWith

87− -- , bindWith

87+ , K.bindWith

88+ , concatPairsWith

88

89

89 -- *** ConcatUnfold

90 -- *** ConcatUnfold

90 -- | Unfold and flatten streams.

91 -- | Unfold and flatten streams.

@@ -440,7 +441,7 @@ roundrobin ::(IsStream t, Monad m) => t m b -> t m b -> t m b

440roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)

441roundrobin m1 m2 = fromStreamD $ D.roundRobin (toStreamD m1) (toStreamD m2)

441

442

442------------------------------------------------------------------------------

443------------------------------------------------------------------------------

443−-- Merging

444+-- Merging (sorted streams)

444------------------------------------------------------------------------------

445------------------------------------------------------------------------------

445

446

446-- | Merge two streams using a comparison function. The head elements of both

447-- | Merge two streams using a comparison function. The head elements of both

@@ -639,6 +640,29 @@ concatMapEitherWith

639concatMapEitherWith = undefined

640concatMapEitherWith = undefined

640-}

641-}

641

642

643+-- XXX Implement a StreamD version for fusion.

644+--

645+-- | Combine streams in pairs using a binary stream combinator, then combine

646+-- the resulting streams in pairs recursively until we get to a single combined

647+-- stream.

648+--

649+-- For example, you can sort a stream using merge sort like this:

650+--

651+-- >>> Stream.toList $ Stream.concatPairsWith (Stream.mergeBy compare) Stream.yield $ Stream.fromList [5,1,7,9,2]

652+-- [1,2,5,7,9]

653+--

654+-- /Caution: the stream of streams must be finite/

655+--

656+-- /Internal/

657+--

658+{-# INLINE concatPairsWith #-}

659+concatPairsWith :: IsStream t =>

660+ (t m b -> t m b -> t m b)

661+ -> (a -> t m b)

662+ -> t m a

663+ -> t m b

664+concatPairsWith = K.concatPairsWith

665+

642------------------------------------------------------------------------------

666------------------------------------------------------------------------------

643-- Combine N Streams - concatUnfold

667-- Combine N Streams - concatUnfold

644------------------------------------------------------------------------------

668------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/IsStream/Top.hs+25

@@ -20,6 +20,9 @@ module Streamly.Internal.Data.Stream.IsStream.Top

20 , sampleBurstStart

21 , sampleBurstEnd

22

23+ -- ** Reordering

24+ , sortBy

25+

23 -- * Nesting

26 -- * Nesting

24 -- ** Set like operations

27 -- ** Set like operations

25 -- | These are not exactly set operations because streams are not

28 -- | These are not exactly set operations because streams are not

@@ -68,6 +71,7 @@ import qualified Streamly.Internal.Data.Array as Array

68import qualified Streamly.Internal.Data.Fold as Fold

71import qualified Streamly.Internal.Data.Fold as Fold

69import qualified Streamly.Internal.Data.Stream.IsStream.Lift as Stream

72import qualified Streamly.Internal.Data.Stream.IsStream.Lift as Stream

70import qualified Streamly.Internal.Data.Stream.IsStream.Eliminate as Stream

73import qualified Streamly.Internal.Data.Stream.IsStream.Eliminate as Stream

74+import qualified Streamly.Internal.Data.Stream.IsStream.Generate as Stream

71import qualified Streamly.Internal.Data.Stream.IsStream.Nesting as Stream

75import qualified Streamly.Internal.Data.Stream.IsStream.Nesting as Stream

72import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream

76import qualified Streamly.Internal.Data.Stream.IsStream.Transform as Stream

73import qualified Streamly.Internal.Data.Stream.StreamK as StreamK

77import qualified Streamly.Internal.Data.Stream.StreamK as StreamK

@@ -164,6 +168,27 @@ sampleBurstStart gap =

164 . Stream.groupsByRolling f Fold.head

168 . Stream.groupsByRolling f Fold.head

165 . Stream.timeIndexed

169 . Stream.timeIndexed

166

170

171+------------------------------------------------------------------------------

172+-- Reordering

173+------------------------------------------------------------------------------

174+--

175+-- We could possibly choose different algorithms depending on whether the

176+-- input stream is almost sorted (ascending/descending) or random. We could

177+-- serialize the stream to an array and use quicksort.

178+--

179+-- | Sort the input stream using a supplied comparison function.

180+--

181+-- /O(n) space/

182+--

183+-- Note: this is not the fastest possible implementation as of now.

184+--

185+-- /Internal/

186+--

187+{-# INLINE sortBy #-}

188+sortBy :: (IsStream t, Monad m) => (a -> a -> Ordering) -> t m a -> t m a

189+-- XXX creating StreamD and using D.mergeBy may be more efficient due to fusion

190+sortBy f = Stream.concatPairsWith (Stream.mergeBy f) Stream.yield

191+

167------------------------------------------------------------------------------

192------------------------------------------------------------------------------

168-- SQL Joins

193-- SQL Joins

169------------------------------------------------------------------------------

194------------------------------------------------------------------------------

src/Streamly/Internal/Data/Stream/StreamK.hs+1

@@ -163,6 +163,7 @@ module Streamly.Internal.Data.Stream.StreamK

163 , concatMapBy

164 , concatMap

165 , bindWith

166+ , concatPairsWith

166 , apWith

167 , apWith

167 , apSerial

168 , apSerial

168 , apSerialDiscardFst

169 , apSerialDiscardFst

src/Streamly/Internal/Data/Stream/StreamK/Type.hs+46

@@ -68,6 +68,7 @@ module Streamly.Internal.Data.Stream.StreamK.Type

68 , concatMapBy

69 , concatMap

70 , bindWith

71+ , concatPairsWith

71 , apWith

72 , apWith

72 , apSerial

73 , apSerial

73 , apSerialDiscardFst

74 , apSerialDiscardFst

@@ -1118,6 +1119,51 @@ concatMap_ f xs = buildS

1118 (\c n -> foldrSShared (\x b -> foldrSShared c b (unShare $ f x)) n xs)

1119 (\c n -> foldrSShared (\x b -> foldrSShared c b (unShare $ f x)) n xs)

1119-}

1120-}

1120

1121

1122+-- | See 'Streamly.Internal.Data.Stream.IsStream.concatPairsWith' for

1123+-- documentation.

1124+--

1125+{-# INLINE concatPairsWith #-}

1126+concatPairsWith

1127+ :: IsStream t

1128+ => (t m b -> t m b -> t m b)

1129+ -> (a -> t m b)

1130+ -> t m a

1131+ -> t m b

1132+concatPairsWith combine f = go Nothing

1133+

1134+ where

1135+

1136+ go Nothing stream =

1137+ mkStream $ \st yld sng stp ->

1138+ let foldShared = foldStreamShared st yld sng stp

1139+ single a = foldShared $ unShare (f a)

1140+ yieldk a r = foldShared $ go (Just a) r

1141+ in foldStream (adaptState st) yieldk single stp stream

1142+ go (Just a1) stream =

1143+ mkStream $ \st yld sng stp ->

1144+ let foldShared = foldStreamShared st yld sng stp

1145+ stop = foldShared $ unShare (f a1)

1146+ single a = foldShared $ unShare (f a1) `combine` f a

1147+ yieldk a r =

1148+ foldShared

1149+ $ concatPairsWith combine

1150+ (\(x,y) -> combine (unShare x) y)

1151+ $ (f a1, f a) `cons` makePairs Nothing r

1152+ in foldStream (adaptState st) yieldk single stop stream

1153+

1154+ makePairs Nothing stream =

1155+ mkStream $ \st yld sng stp ->

1156+ let foldShared = foldStreamShared st yld sng stp

1157+ single a = sng (f a, nil)

1158+ yieldk a r = foldShared $ makePairs (Just a) r

1159+ in foldStream (adaptState st) yieldk single stp stream

1160+ makePairs (Just a1) stream =

1161+ mkStream $ \st yld sng _ ->

1162+ let stop = sng (f a1, nil)

1163+ single a = sng (f a1, f a)

1164+ yieldk a r = yld (f a1, f a) (makePairs Nothing r)

1165+ in foldStream (adaptState st) yieldk single stop stream

1166+

1121instance Monad m => Applicative (Stream m) where

1167instance Monad m => Applicative (Stream m) where

1122 {-# INLINE pure #-}

1168 {-# INLINE pure #-}

1123 pure = yield

1169 pure = yield