ec20f5f · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Fix/simplify intersectBySorted

ec20f5fHarendra Kumar5 years ago

benchmark/Streamly/Benchmark/Prelude/Serial/NestedStream.hs+20−21

compare with master

unifiedsplit

@@ -417,57 +417,56 @@ o_n_space_monad value =

417-- Joining

418-------------------------------------------------------------------------------

419

420−toKvMap :: Int -> (Int, Int)

420+toKv :: Int -> (Int, Int)

421−toKvMap p = (p, p)

421+toKv p = (p, p)

422

423{-# INLINE joinWith #-}

424joinWith :: (S.MonadAsync m) =>

425 ((Int -> Int -> Bool) -> SerialT m Int -> SerialT m Int -> SerialT m b)

426 -> Int

427 -> Int

428− -> Int

429 -> m ()

428 -> m ()

430−joinWith j val1 val2 i =

429+joinWith j val i =

431− S.drain $ j (==) (sourceUnfoldrM val1 i) (sourceUnfoldrM val2 i)

430+ S.drain $ j (==) (sourceUnfoldrM val i) (sourceUnfoldrM val (val `div` 2))

432

431

433{-# INLINE joinMapWith #-}

432{-# INLINE joinMapWith #-}

434joinMapWith :: (S.MonadAsync m) =>

433joinMapWith :: (S.MonadAsync m) =>

435 (SerialT m (Int, Int) -> SerialT m (Int, Int) -> SerialT m b)

434 (SerialT m (Int, Int) -> SerialT m (Int, Int) -> SerialT m b)

436 -> Int

435 -> Int

437 -> Int

436 -> Int

438− -> Int

439 -> m ()

437 -> m ()

440−joinMapWith j val1 val2 i =

438+joinMapWith j val i =

441 S.drain

439 S.drain

442 $ j

440 $ j

443− (fmap toKvMap (sourceUnfoldrM val1 i))

441+ (fmap toKv (sourceUnfoldrM val i))

444− (fmap toKvMap (sourceUnfoldrM val2 i))

442+ (fmap toKv (sourceUnfoldrM val (val `div` 2)))

445

443

446o_n_heap_buffering :: Int -> [Benchmark]

444o_n_heap_buffering :: Int -> [Benchmark]

447o_n_heap_buffering value =

445o_n_heap_buffering value =

448 [ bgroup "buffered"

446 [ bgroup "buffered"

449 [

447 [

450− benchIOSrc1 "joinInner"

448+ benchIOSrc1 "joinInner (sqrtVal)"

451− $ joinWith Internal.joinInner sqrtVal sqrtVal

449+ $ joinWith Internal.joinInner sqrtVal

452 , benchIOSrc1 "joinInnerMap"

450 , benchIOSrc1 "joinInnerMap"

453− $ joinMapWith Internal.joinInnerMap sqrtVal sqrtVal

451+ $ joinMapWith Internal.joinInnerMap halfVal

454− , benchIOSrc1 "joinLeft"

452+ , benchIOSrc1 "joinLeft (sqrtVal)"

455− $ joinWith Internal.joinLeft sqrtVal sqrtVal

453+ $ joinWith Internal.joinLeft sqrtVal

456 , benchIOSrc1 "joinLeftMap "

454 , benchIOSrc1 "joinLeftMap "

457− $ joinMapWith Internal.joinLeftMap sqrtVal sqrtVal

455+ $ joinMapWith Internal.joinLeftMap halfVal

458− , benchIOSrc1 "joinOuter"

456+ , benchIOSrc1 "joinOuter (sqrtVal)"

459− $ joinWith Internal.joinOuter sqrtVal sqrtVal

457+ $ joinWith Internal.joinOuter sqrtVal

460 , benchIOSrc1 "joinOuterMap"

458 , benchIOSrc1 "joinOuterMap"

461− $ joinMapWith Internal.joinOuterMap sqrtVal sqrtVal

459+ $ joinMapWith Internal.joinOuterMap halfVal

462− , benchIOSrc1 "intersectBy"

460+ , benchIOSrc1 "intersectBy (sqrtVal)"

463− $ joinWith Internal.intersectBy sqrtVal sqrtVal

461+ $ joinWith Internal.intersectBy sqrtVal

464 , benchIOSrc1 "intersectBySorted"

462 , benchIOSrc1 "intersectBySorted"

465− $ joinMapWith Internal.intersectBySorted sqrtVal sqrtVal

463+ $ joinMapWith (Internal.intersectBySorted compare) halfVal

466 ]

464 ]

467 ]

465 ]

468

466

469 where

467 where

470

468

469+ halfVal = value `div` 2

471 sqrtVal = round $ sqrt (fromIntegral value :: Double)

470 sqrtVal = round $ sqrt (fromIntegral value :: Double)

472

471

473-------------------------------------------------------------------------------

472-------------------------------------------------------------------------------

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

@@ -28,7 +28,7 @@ module Streamly.Internal.Data.Stream.IsStream.Top

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

29 -- necessarily sets, they may have duplicated elements.

30 , intersectBy

31− , intersectBySorted

31+ , intersectBySorted

32 , differenceBy

33 , mergeDifferenceBy

34 , unionBy

@@ -65,7 +65,6 @@ import Streamly.Internal.Data.Stream.IsStream.Common (concatM)

65import Streamly.Internal.Data.Stream.IsStream.Type

66 (IsStream(..), adapt, foldl', fromList)

67import Streamly.Internal.Data.Stream.Serial (SerialT)

68−--import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)

69import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)

68import Streamly.Internal.Data.Time.Units (NanoSecond64(..), toRelTime64)

71import qualified Data.List as List

70import qualified Data.List as List

@@ -576,7 +575,7 @@ intersectBy eq s1 s2 =

576 xs <- Stream.toListRev $ Stream.uniqBy eq $ adapt s2

575 xs <- Stream.toListRev $ Stream.uniqBy eq $ adapt s2

577 return $ Stream.filter (\x -> List.any (eq x) xs) s1

576 return $ Stream.filter (\x -> List.any (eq x) xs) s1

578

577

579−-- | Like 'intersectBy' but works only on sorted streams.

578+-- | Like 'intersectBy' but works only on streams sorted in ascending order.

580--

579--

581-- Space: O(1)

580-- Space: O(1)

582--

581--

@@ -584,10 +583,12 @@ intersectBy eq s1 s2 =

584--

583--

585-- /Pre-release/

584-- /Pre-release/

586{-# INLINE intersectBySorted #-}

585{-# INLINE intersectBySorted #-}

587−intersectBySorted :: (IsStream t, MonadIO m, Eq a) =>

586+intersectBySorted :: (IsStream t, Monad m) =>

588 (a -> a -> Ordering) -> t m a -> t m a -> t m a

587 (a -> a -> Ordering) -> t m a -> t m a -> t m a

589intersectBySorted eq s1 =

588intersectBySorted eq s1 =

589+ IsStream.fromStreamD

590− IsStream.fromStreamD . StreamD.intersectBySorted eq (IsStream.toStreamD s1) . IsStream.toStreamD

590+ . StreamD.intersectBySorted eq (IsStream.toStreamD s1)

591+ . IsStream.toStreamD

591

592

592-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2

593-- Roughly joinLeft s1 s2 = s1 `difference` s2 + s1 `intersection` s2

593

594

src/Streamly/Internal/Data/Stream/StreamD/Nesting.hs+24−34

@@ -484,57 +484,47 @@ mergeBy

484mergeBy cmp = mergeByM (\a b -> return $ cmp a b)

485

486-------------------------------------------------------------------------------

487−-- Intersection of sorted streams ---------------------------------------------

487+-- Intersection of sorted streams

488-------------------------------------------------------------------------------

489+

490+-- Assuming the streams are sorted in ascending order

489{-# INLINE_NORMAL intersectBySorted #-}

491{-# INLINE_NORMAL intersectBySorted #-}

490−intersectBySorted

492+intersectBySorted :: Monad m

491− :: (MonadIO m, Eq a)

492 => (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a

493 => (a -> a -> Ordering) -> Stream m a -> Stream m a -> Stream m a

493intersectBySorted cmp (Stream stepa ta) (Stream stepb tb) =

494intersectBySorted cmp (Stream stepa ta) (Stream stepb tb) =

494− Stream step (Just ta, Just tb, Nothing, Nothing, Nothing)

495+ Stream step

496+ ( ta -- left stream state

497+ , tb -- right stream state

498+ , Nothing -- left value

499+ , Nothing -- right value

500+ )

495

501

496 where

502 where

497− {-# INLINE_LATE step #-}

498

503

499− -- step 1

504+ {-# INLINE_LATE step #-}

505+ -- step 1, fetch the first value

500− step gst (Just sa, sb, Nothing, b, Nothing) = do

506+ step gst (sa, sb, Nothing, b) = do

501 r <- stepa gst sa

507 r <- stepa gst sa

502 return $ case r of

508 return $ case r of

503− Yield a sa' -> Skip (Just sa', sb, Just a, b, Nothing)

509+ Yield a sa' -> Skip (sa', sb, Just a, b) -- step 2/3

504− Skip sa' -> Skip (Just sa', sb, Nothing, b, Nothing)

510+ Skip sa' -> Skip (sa', sb, Nothing, b)

505 Stop -> Stop

511 Stop -> Stop

506

512

507− -- step 2

513+ -- step 2, fetch the second value

508− step gst (sa, Just sb, a, Nothing, Nothing) = do

514+ step gst (sa, sb, a@(Just _), Nothing) = do

509 r <- stepb gst sb

515 r <- stepb gst sb

510 return $ case r of

516 return $ case r of

511− Yield b sb' -> Skip (sa, Just sb', a, Just b, Nothing)

517+ Yield b sb' -> Skip (sa, sb', a, Just b) -- step 3

512− Skip sb' -> Skip (sa, Just sb', a, Nothing, Nothing)

518+ Skip sb' -> Skip (sa, sb', a, Nothing)

513 Stop -> Stop

519 Stop -> Stop

514

520

515− -- step 3

521+ -- step 3, compare the two values

516− -- both the values are available compare it

517− step _ (sa, sb, Just a, Just b, Nothing) = do

522+ step _ (sa, sb, Just a, Just b) = do

518 let res = cmp a b

523 let res = cmp a b

519 return $ case res of

524 return $ case res of

520− GT -> Skip (sa, sb, Just a, Nothing, Nothing)

525+ GT -> Skip (sa, sb, Just a, Nothing) -- step 2

521− LT -> Skip (sa, sb, Nothing, Just b, Nothing)

526+ LT -> Skip (sa, sb, Nothing, Just b) -- step 1

522− EQ -> Yield a (sa, sb, Nothing, Just a, Just b) -- step 4

527+ EQ -> Yield a (sa, sb, Nothing, Just b) -- step 1

523−

524− -- step 4

525− -- Matching element

526− step gst (Just sa, Just sb, Nothing, Just _, Just b) = do

527− r1 <- stepa gst sa

528− return $ case r1 of

529− Yield a' sa' -> do

530− if a' == b -- match with prev a

531− then Yield a' (Just sa', Just sb, Nothing, Just b, Just b) --step 1

532− else Skip (Just sa', Just sb, Just a', Nothing, Nothing)

533−

534− Skip sa' -> Skip (Just sa', Just sb, Nothing, Nothing, Nothing)

535− Stop -> Stop

536−

537− step _ (_, _, _, _, _) = return Stop

538

528

539------------------------------------------------------------------------------

529------------------------------------------------------------------------------

540-- Combine N Streams - unfoldMany

530-- Combine N Streams - unfoldMany

streamly.cabal+1−1

@@ -101,7 +101,7 @@ extra-source-files:

101 test/Streamly/Test/Data/Array/Prim/Pinned.hs

102 test/Streamly/Test/Data/Array/Foreign.hs

103 test/Streamly/Test/Data/Array/Stream/Foreign.hs

104− test/Streamly/Test/Data/Parser/ParserD.hs

104+ test/Streamly/Test/Data/Parser/ParserD.hs

105 test/Streamly/Test/FileSystem/Event.hs

106 test/Streamly/Test/FileSystem/Event/Common.hs

107 test/Streamly/Test/FileSystem/Event/Darwin.hs

test/Streamly/Test/Prelude/Top.hs+22−29

@@ -1,8 +1,8 @@

1−module Main (main)

1+module Main (main) where

2− where

4import Data.List (elem, intersect, nub, sort)

3import Data.List (elem, intersect, nub, sort)

5import Data.Maybe (isNothing)

4import Data.Maybe (isNothing)

5+import Streamly.Prelude (SerialT)

6import Test.QuickCheck

7 ( Gen

8 , Property

@@ -169,8 +169,16 @@ joinLeftMap =

169 let v2 = joinLeftList ls0 ls1

170 assert (v1 == v2)

171

172−intersectBy :: Property

172+intersectBy ::

173+ ([Int] -> [Int])

174+ -> ( (Int -> Int -> a)

175+ -> SerialT IO Int

176+ -> SerialT IO Int

177+ -> SerialT IO Int

178+ )

179+ -> (Int -> Int -> a)

180+ -> Property

173−intersectBy =

181+intersectBy _srt intersectFunc cmp =

174 forAll (listOf (chooseInt (min_value, max_value))) $ \ls0 ->

182 forAll (listOf (chooseInt (min_value, max_value))) $ \ls0 ->

175 forAll (listOf (chooseInt (min_value, max_value))) $ \ls1 ->

183 forAll (listOf (chooseInt (min_value, max_value))) $ \ls1 ->

176 monadicIO $ action (sort ls0) (sort ls1)

184 monadicIO $ action (sort ls0) (sort ls1)

@@ -181,33 +189,17 @@ intersectBy =

181 v1 <-

189 v1 <-

182 run

190 run

183 $ S.toList

191 $ S.toList

184− $ Top.intersectBy

192+ $ intersectFunc

185− (==)

193+ cmp

186 (S.fromList ls0)

194 (S.fromList ls0)

187 (S.fromList ls1)

195 (S.fromList ls1)

188− let v2 = intersect ls0 ls1

189− assert (v1 == sort v2)

190−

191−intersectBySorted :: Property

192−intersectBySorted =

193− forAll (listOf (chooseInt (min_value, max_value))) $ \ls0 ->

194− forAll (listOf (chooseInt (min_value, max_value))) $ \ls1 ->

195− monadicIO $ action (sort ls0) (sort ls1)

196−

197− where

198−

199− action ls0 ls1 = do

200− v1 <-

201− run

202− $ S.toList

203− $ Top.intersectBySorted

204− compare

205− (S.fromList ls0)

206− (S.fromList ls1)

207− let v2 = intersect ls0 ls1

196+ let v2 = ls0 `intersect` ls1

208 assert (v1 == sort v2)

197 assert (v1 == sort v2)

209

198

210-------------------------------------------------------------------------------

199-------------------------------------------------------------------------------

200+-- Main

201+-------------------------------------------------------------------------------

202+

211moduleName :: String

203moduleName :: String

212moduleName = "Prelude.Top"

204moduleName = "Prelude.Top"

213

205

@@ -215,7 +207,6 @@ main :: IO ()

215main = hspec $ do

207main = hspec $ do

216 describe moduleName $ do

208 describe moduleName $ do

217 -- Joins

209 -- Joins

218−

219 prop "joinInner" Main.joinInner

210 prop "joinInner" Main.joinInner

220 prop "joinInnerMap" Main.joinInnerMap

211 prop "joinInnerMap" Main.joinInnerMap

221 -- XXX currently API is broken https://github.com/composewell/streamly/issues/1032

212 -- XXX currently API is broken https://github.com/composewell/streamly/issues/1032

@@ -224,5 +215,7 @@ main = hspec $ do

224 prop "joinLeft" Main.joinLeft

215 prop "joinLeft" Main.joinLeft

225 prop "joinLeftMap" Main.joinLeftMap

216 prop "joinLeftMap" Main.joinLeftMap

226 -- intersect

217 -- intersect

218+ -- XXX currently API is broken https://github.com/composewell/streamly/issues/1471

227− prop "intersectBy" Main.intersectBy

219+ --prop "intersectBy" (intersectBy id Top.intersectBy (==))

228− prop "intersectBySorted" Main.intersectBySorted

220+ prop "intersectBySorted"

221+ (intersectBy sort Top.intersectBySorted compare)