2c70b45 · streamly · git.iko.soy

git.iko.soy/streamly

Implement concatPairswith/sortBy

commit2c70b4566019b9476b7e8aa4c5d10517318c7eb9 authorHarendra Kumar <harendra@composewell.com> dateWed 3 Feb 2021, 14:40 (6 years ago) parentd0cb8fc

6 files changed+107−2

compare with master

unifiedsplit

benchmark/Streamly/Benchmark/Prelude/Serial/Transformation2.hs+7 src/Streamly/Internal/Data/Stream/IsStream/Common.hs+2 src/Streamly/Internal/Data/Stream/IsStream/Nesting.hs+26−2 src/Streamly/Internal/Data/Stream/IsStream/Top.hs+25 src/Streamly/Internal/Data/Stream/StreamK.hs+1 src/Streamly/Internal/Data/Stream/StreamK/Type.hs+46

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

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

2020import Data.Monoid (Sum(..))

2121import GHC.Generics (Generic)

2222

23+import qualified Data.List as List

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

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

2526import qualified Streamly.Prelude as S

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

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

175176reverse' n = composeN n Internal.reverse'

176177

178+{-# INLINE sortBy #-}

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

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

181+

177182o_n_heap_buffering :: Int -> [Benchmark]

178183o_n_heap_buffering value =

179184 [ bgroup "buffered"

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

181186 -- Reversing/sorting a stream

182187 benchIOSink value "reverse" (reverse 1)

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

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

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

184191

185192 , benchIOSink value "mkAsync" (mkAsync serially)

186193 ]

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

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

88-- Stability : experimental

99-- Portability : GHC

1010--

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

12+-- IsStream modules.

1113

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

1315 (

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

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

8484 , concatMapM

8585 , concatMapWith

8686 , concatSmapMWith

87− -- , bindWith

87+ , K.bindWith

88+ , concatPairsWith

8889

8990 -- *** ConcatUnfold

9091 -- | Unfold and flatten streams.

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

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

441442

442443------------------------------------------------------------------------------

443−-- Merging

444+-- Merging (sorted streams)

444445------------------------------------------------------------------------------

445446

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

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

639640concatMapEitherWith = undefined

640641-}

641642

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+

642666------------------------------------------------------------------------------

643667-- Combine N Streams - concatUnfold

644668------------------------------------------------------------------------------

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

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

2020 , sampleBurstStart

2121 , sampleBurstEnd

2222

23+ -- ** Reordering

24+ , sortBy

25+

2326 -- * Nesting

2427 -- ** Set like operations

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

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

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

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

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

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

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

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

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

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

164168 . Stream.groupsByRolling f Fold.head

165169 . Stream.timeIndexed

166170

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+

167192------------------------------------------------------------------------------

168193-- SQL Joins

169194------------------------------------------------------------------------------

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

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

163163 , concatMapBy

164164 , concatMap

165165 , bindWith

166+ , concatPairsWith

166167 , apWith

167168 , apSerial

168169 , apSerialDiscardFst

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

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

6868 , concatMapBy

6969 , concatMap

7070 , bindWith

71+ , concatPairsWith

7172 , apWith

7273 , apSerial

7374 , apSerialDiscardFst

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

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

11191120-}

11201121

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+

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

11221168 {-# INLINE pure #-}

11231169 pure = yield