c8d1ca3 · streamly · git.iko.soy

git.iko.soy/streamly

Fix joinOuterMap operation

The streams were used multiple times, now they are used only once.

commitc8d1ca3fc77b1eec4e235282c2d90ae27b68a2d1 authorHarendra Kumar <harendra@composewell.com> dateMon 24 Jan 2022, 18:15 (5 years ago) parent72a2b1f

3 files changed+41−40

compare with master

unifiedsplit

benchmark/Streamly/Benchmark/Prelude/Serial/NestedStream.hs+30−32 src/Streamly/Internal/Data/Stream/IsStream/Top.hs+11−7 test/Streamly/Test/Prelude/Top.hs−1

benchmark/Streamly/Benchmark/Prelude/Serial/NestedStream.hs+30−32

@@ -420,43 +420,41 @@ o_n_space_monad value =

420420toKvMap :: Int -> (Int, Int)

421421toKvMap p = (p, p)

422422

423−mkStreamLen :: (S.IsStream t, S.MonadAsync m) => Int -> t m Int

424−mkStreamLen count = sourceUnfoldrM count 0

425−

426−{-# INLINE joinInner #-}

427−joinInner :: Int -> Int -> Int -> IO ()

428−joinInner val1 val2 _ =

429− S.drain $ Internal.joinInner (==) (mkStreamLen val1) $ mkStreamLen val2

430−

431−{-# INLINE joinInnerMap #-}

432−joinInnerMap :: Int -> Int -> Int -> IO ()

433−joinInnerMap val1 val2 _ =

434− S.drain $

435− Internal.joinInnerMap

436− (fmap toKvMap (mkStreamLen val1))

437− (fmap toKvMap (mkStreamLen val2))

438−

439−{-# INLINE joinOuter #-}

440−joinOuter :: Int -> Int -> Int -> IO ()

441−joinOuter val1 val2 _ =

442− S.drain $ Internal.joinOuter (==) (mkStreamLen val1) $ mkStreamLen val2

443−

444−{-# INLINE joinOuterMap #-}

445−joinOuterMap :: Int -> Int -> Int -> IO ()

446−joinOuterMap val1 val2 _ =

447− S.drain $

448− Internal.joinOuterMap

449− (fmap toKvMap (mkStreamLen val1))

450− (fmap toKvMap (mkStreamLen val2))

423+{-# INLINE joinWith #-}

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

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

426+ -> Int

427+ -> Int

428+ -> Int

429+ -> m ()

430+joinWith j val1 val2 i =

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

432+

433+{-# INLINE joinMapWith #-}

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

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

436+ -> Int

437+ -> Int

438+ -> Int

439+ -> m ()

440+joinMapWith j val1 val2 i =

441+ S.drain

442+ $ j

443+ (fmap toKvMap (sourceUnfoldrM val1 i))

444+ (fmap toKvMap (sourceUnfoldrM val2 i))

451445

452446o_n_heap_buffering :: Int -> [Benchmark]

453447o_n_heap_buffering value =

454448 [ bgroup "buffered"

455449 [

456− benchIOSrc1 "joinInner" (joinInner sqrtVal sqrtVal)

457− , benchIOSrc1 "joinInnerMap" (joinInnerMap sqrtVal sqrtVal)

458− , benchIOSrc1 "joinOuter" (joinOuter sqrtVal sqrtVal)

459− , benchIOSrc1 "joinOuterMap" (joinOuterMap sqrtVal sqrtVal)

450+ benchIOSrc1 "joinInner"

451+ $ joinWith Internal.joinInner sqrtVal sqrtVal

452+ , benchIOSrc1 "joinInnerMap"

453+ $ joinMapWith Internal.joinInnerMap sqrtVal sqrtVal

454+ , benchIOSrc1 "joinOuter"

455+ $ joinWith Internal.joinOuter sqrtVal sqrtVal

456+ , benchIOSrc1 "joinOuterMap"

457+ $ joinMapWith Internal.joinOuterMap sqrtVal sqrtVal

460458 ]

461459 ]

462460

src/Streamly/Internal/Data/Stream/IsStream/Top.hs+11−7

@@ -385,7 +385,7 @@ leftJoin eq s1 s2 = Stream.evalStateT (return False) $ do

385385 else Stream.nil

386386 Nothing -> return (a, Nothing)

387387

388−-- | Like 'joinOuter' but uses a hashmap for efficiency.

388+-- | Like 'joinLeft' but uses a hashmap for efficiency.

389389--

390390-- Space: O(n)

391391--

@@ -470,11 +470,9 @@ joinOuter eq s1 s =

470470-- a flag. At the end go through @t m b@ and find those that are not in that

471471-- hash to return (Nothing, b).

472472--

473−-- | Like 'joinOuter' but uses a hashmap for efficiency.

473+-- | Like 'joinOuter' but uses a 'Map' for efficiency.

474474--

475−-- For space efficiency use the smaller stream as the second stream.

476−--

477−-- Space: O(n)

475+-- Space: O(m + n)

478476--

479477-- Time: O(m + n)

480478--

@@ -488,14 +486,20 @@ joinOuterMap s1 s2 =

488486 km1 <- kvFold $ IsStream.adapt s1

489487 km2 <- kvFold $ IsStream.adapt s2

490488

491− let res1 = Stream.map (joinAB km2) s1

489+ -- XXX Not sure if toList/fromList would fuse optimally. We may have to

490+ -- create a fused Map.toStream function.

491+ let res1 = Stream.map (joinAB km2) $ Stream.fromList $ Map.toList km1

492492 where

493493 joinAB km (k, a) =

494494 case k `Map.lookup` km of

495495 Just b -> (k, Just a, Just b)

496496 Nothing -> (k, Just a, Nothing)

497497

498− let res2 = Stream.mapMaybe (joinAB km1) s2

498+ -- XXX We can take advantage of the lookups in the first pass above to

499+ -- reduce the number of lookups in this pass. If we keep mutable cells

500+ -- in the second Map, we can flag it in the first pass and not do any

501+ -- lookup in the second pass if it is flagged.

502+ let res2 = Stream.mapMaybe (joinAB km1) $ Stream.fromList $ Map.toList km2

499503 where

500504 joinAB km (k, b) =

501505 case k `Map.lookup` km of

test/Streamly/Test/Prelude/Top.hs−1

@@ -84,7 +84,6 @@ joinOuter =

8484 let v2 = [ (Just i, Just j) | i <- ls0, j <- ls1, i == j ]

8585 assert (v1 == v2)

8686

87−

8887joinOuterMap :: Property

8988joinOuterMap =

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