6e52834 · streamly · git.iko.soy

git.iko.soy/streamly

Fix groupByRollingEither

* Change the signature to return different types in Either * Fix a case to return correct Either value * Fix and enhance benchmarks * Implement sortBy using groupByRollingEither

commit6e52834cb26d201949017934287aa30ca174f75a authorHarendra Kumar <harendra@composewell.com> dateSun 7 Nov 2021, 00:33 (5 years ago) parentb926110

6 files changed+69−28

compare with master

unifiedsplit

benchmark/Streamly/Benchmark/Data/Parser/ParserD.hs+25−12 benchmark/Streamly/Benchmark/Prelude/Merge.hs+21 src/Streamly/Internal/Data/Parser.hs+1−1 src/Streamly/Internal/Data/Parser/ParserD.hs+9−9 src/Streamly/Internal/Data/Stream/IsStream/Top.hs+12−5 test/Streamly/Test/Prelude/Serial.hs+1−1

benchmark/Streamly/Benchmark/Data/Parser/ParserD.hs+25−12

@@ -60,6 +60,17 @@ benchIOSink

6060benchIOSink value name f =

6161 bench name $ nfIO $ randomRIO (1,1) >>= f . sourceUnfoldrM value

6262

63+-- Make the input unsorted.

64+{-# INLINE benchIOSinkRandom #-}

65+benchIOSinkRandom

66+ :: (IsStream t, NFData b)

67+ => Int -> String -> (t IO Int -> IO b) -> Benchmark

68+benchIOSinkRandom value name f =

69+ bench name $ nfIO $ randomRIO (1,1)

70+ >>= f

71+ . S.map (\x -> if even x then x + 2 else x)

72+ . sourceUnfoldrM value

73+

6374-------------------------------------------------------------------------------

6475-- Parsers

6576-------------------------------------------------------------------------------

@@ -218,10 +229,10 @@ parseManyGroupsRolling b =

218229

219230{-# INLINE parseManyGroupsRollingEither #-}

220231parseManyGroupsRollingEither :: (MonadThrow m, MonadCatch m) =>

221− Bool -> SerialT m Int -> m ()

222−parseManyGroupsRollingEither b =

223− IP.drain .

224− IP.parseManyD (PR.groupByRollingEither (\_ _ -> b) FL.drain FL.drain)

232+ (Int -> Int -> Bool) -> SerialT m Int -> m ()

233+parseManyGroupsRollingEither cmp =

234+ IP.drain

235+ . IP.parseManyD (PR.groupByRollingEither cmp FL.drain FL.drain)

225236

226237-------------------------------------------------------------------------------

227238-- Parsing with unfolds

@@ -334,16 +345,18 @@ o_1_space_serial_nested :: Int -> [Benchmark]

334345o_1_space_serial_nested value =

335346 [ benchIOSink value "parseMany" $ parseMany value

336347 , benchIOSink value "parseMany groupBy (bound groups)"

337− $ (parseManyGroups False)

348+ $ parseManyGroups False

338349 , benchIOSink value "parseMany groupRollingBy (bound groups)"

339− $ (parseManyGroupsRolling False)

340− , benchIOSink value "parseMany groupBy (1 group)" $ (parseManyGroups True)

350+ $ parseManyGroupsRolling False

351+ , benchIOSink value "parseMany groupBy (1 group)" $ parseManyGroups True

341352 , benchIOSink value "parseMany groupRollingBy (1 group)"

342− $ (parseManyGroupsRolling True)

343− , benchIOSink value "parseMany groupRollingByEither (1 group)"

344− $ (parseManyGroupsRollingEither True)

345− , benchIOSink value "parseMany groupRollingByEither (bound groups)"

346− $ (parseManyGroupsRollingEither False)

353+ $ parseManyGroupsRolling True

354+ , benchIOSink value "parseMany groupRollingByEither (Left)"

355+ $ parseManyGroupsRollingEither (<)

356+ , benchIOSink value "parseMany groupRollingByEither (Right)"

357+ $ parseManyGroupsRollingEither (>)

358+ , benchIOSinkRandom value "parseMany groupRollingByEither (Alternating)"

359+ $ parseManyGroupsRollingEither (>)

347360 ]

348361

349362o_1_space_serial_unfold :: Int -> [Array.Array Int] -> [Benchmark]

benchmark/Streamly/Benchmark/Prelude/Merge.hs+21

@@ -21,6 +21,7 @@

2121#endif

2222

2323import Streamly.Internal.Data.Stream.IsStream (SerialT)

24+import qualified Data.List as List

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

2526-- import qualified Streamly.Internal.Data.Unfold as Unfold

2627

@@ -194,6 +195,7 @@ inspect $ hasNoTypeClasses 'unfoldManyMergeBy

194195sortBy :: (Int -> Int -> Ordering) -> SerialT IO Int -> IO ()

195196sortBy cmp = Stream.drain . Stream.sortBy cmp

196197

198+{-

197199-- For fair comparison with concatPairs, removed sorted segmentation

198200{-# INLINE listSortBy #-}

199201listSortBy :: (a -> a -> Ordering) -> [a] -> [a]

@@ -213,6 +215,7 @@ listSortBy cmp = mergeAll . sequences

213215 | otherwise = a:merge as' bs

214216 merge [] bs = bs

215217 merge as [] = as

218+-}

216219

217220o_n_heap_concat :: Int -> [Benchmark]

218221o_n_heap_concat value =

@@ -246,10 +249,28 @@ o_n_heap_concat value =

246249 , bgroup "sorting"

247250 [ benchIOSink value "sortBy compare" (sortBy compare)

248251 , benchIOSink value "sortBy (flip compare)" (sortBy (flip compare))

252+ , benchIOSink value "sortBy compare randomized"

253+ (sortBy compare . Stream.map (\x -> if even x then x + 2 else x))

254+ {-

249255 , bench "sortByLists compare"

250256 $ nf (\x -> listSortBy compare [1..x]) value

251257 , bench "sortByLists (flip compare)"

252258 $ nf (\x -> listSortBy (flip compare) [1..x]) value

259+ , bench "sortByLists compare randomized"

260+ $ nf (\x -> listSortBy compare

261+ (map (\n -> if even n then n + 2 else n) [1..x])

262+ )

263+ value

264+ -}

265+ , bench "sortByLists compare"

266+ $ nf (\x -> List.sortBy compare [1..x]) value

267+ , bench "sortByLists (flip compare)"

268+ $ nf (\x -> List.sortBy (flip compare) [1..x]) value

269+ , bench "sortByLists compare randomized"

270+ $ nf (\x -> List.sortBy compare

271+ (map (\n -> if even n then n + 2 else n) [1..x])

272+ )

273+ value

253274 ]

254275 ]

255276

src/Streamly/Internal/Data/Parser.hs+1−1

@@ -722,7 +722,7 @@ groupByRolling eq = K.toParserK . D.groupByRolling eq

722722-- /Unimplemented/

723723{-# INLINABLE groupByRollingEither #-}

724724groupByRollingEither :: MonadCatch m =>

725− (a -> a -> Bool) -> Fold m a b -> Fold m a b -> Parser m a (Either b b)

725+ (a -> a -> Bool) -> Fold m a b -> Fold m a c -> Parser m a (Either b c)

726726groupByRollingEither eq f1 = K.toParserK . D.groupByRollingEither eq f1

727727

728728-- | Match the given sequence of elements using the given comparison function.

src/Streamly/Internal/Data/Parser/ParserD.hs+9−9

@@ -734,13 +734,6 @@ data GroupByState a s

734734 = GroupByInit !s

735735 | GroupByGrouping !a !s

736736

737−{-# ANN type GroupByStatePair Fuse #-}

738−data GroupByStatePair a s1 s2

739− = GroupByInitPair !s1 !s2

740− | GroupByGroupingPair !a !s1 !s2

741− | GroupByGroupingPairL !a !s1 !s2

742− | GroupByGroupingPairR !a !s1 !s2

743−

744737-- | See 'Streamly.Internal.Data.Parser.groupBy'.

745738--

746739{-# INLINE groupBy #-}

@@ -805,9 +798,16 @@ groupByRolling eq (Fold fstep finitial fextract) = Parser step initial extract

805798 extract (GroupByInit s) = fextract s

806799 extract (GroupByGrouping _ s) = fextract s

807800

801+{-# ANN type GroupByStatePair Fuse #-}

802+data GroupByStatePair a s1 s2

803+ = GroupByInitPair !s1 !s2

804+ | GroupByGroupingPair !a !s1 !s2

805+ | GroupByGroupingPairL !a !s1 !s2

806+ | GroupByGroupingPairR !a !s1 !s2

807+

808808{-# INLINABLE groupByRollingEither #-}

809809groupByRollingEither :: MonadCatch m =>

810− (a -> a -> Bool) -> Fold m a b -> Fold m a b -> Parser m a (Either b b)

810+ (a -> a -> Bool) -> Fold m a b -> Fold m a c -> Parser m a (Either b c)

811811groupByRollingEither

812812 eq

813813 (Fold fstep1 finitial1 fextract1)

@@ -857,7 +857,7 @@ groupByRollingEither

857857 FL.Partial s1 ->

858858 case res2 of

859859 FL.Partial s2 -> IPartial $ GroupByInitPair s1 s2

860− FL.Done b -> IDone (Left b)

860+ FL.Done b -> IDone (Right b)

861861 FL.Done b -> IDone (Left b)

862862

863863 step (GroupByInitPair s1 s2) a = grouper s1 s2 a

src/Streamly/Internal/Data/Stream/IsStream/Top.hs+12−5

@@ -50,6 +50,7 @@ where

5050

5151#include "inline.hs"

5252

53+import Control.Monad.Catch (MonadCatch)

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

5455import Control.Monad.Trans.Class (lift)

5556import Control.Monad.Trans.State.Strict (get, put)

@@ -69,6 +70,7 @@ import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)

6970import qualified Data.List as List

7071import qualified Streamly.Internal.Data.Array as Array

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

73+import qualified Streamly.Internal.Data.Parser as Parser

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

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

7476import qualified Streamly.Internal.Data.Stream.IsStream.Generate as Stream

@@ -181,9 +183,6 @@ sampleBurstStart gap =

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

182184-- serialize the stream to an array and use quicksort.

183185--

184−-- Use 'groupByRollingEither (\x -> (< GT) . f x) Fold.toList Fold.toListRev'

185−-- to generate sorted segments before merging. Compare the perf.

186−--

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

188187--

189188-- /O(n) space/

@@ -193,8 +192,16 @@ sampleBurstStart gap =

193192-- /Pre-release/

194193--

195194{-# INLINE sortBy #-}

196−sortBy :: IsStream t => (a -> a -> Ordering) -> t m a -> t m a

197−sortBy f = Stream.concatPairsWith (Stream.mergeBy f) Stream.fromPure

195+sortBy :: MonadCatch m => (a -> a -> Ordering) -> SerialT m a -> SerialT m a

196+-- sortBy f = Stream.concatPairsWith (Stream.mergeBy f) Stream.fromPure

197+sortBy cmp =

198+ let p =

199+ Parser.groupByRollingEither

200+ (\x -> (< GT) . cmp x)

201+ Fold.toStreamRev

202+ Fold.toStream

203+ in Stream.concatPairsWith (Stream.mergeBy cmp) id

204+ . Stream.parseMany (fmap (either id id) p)

198205

199206------------------------------------------------------------------------------

200207-- SQL Joins

test/Streamly/Test/Prelude/Serial.hs+1−1

@@ -505,7 +505,7 @@ foldIterateM =

505505sortBy :: Property

506506sortBy = forAll (listOf (chooseInt (0, max_length))) $ \lst -> monadicIO $ do

507507 let s1 = sort lst

508− s2 <- S.toList $ IS.sortBy compare $ S.fromList lst

508+ s2 <- run $ S.toList $ IS.sortBy compare $ S.fromList lst

509509 assert $ s1 == s2

510510

511511moduleName :: String