c8d1ca3 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Fix joinOuterMap operation

c8d1ca3Harendra Kumar5 years ago

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

compare with master

unifiedsplit

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

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

421toKvMap p = (p, p)

422

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

423+{-# INLINE joinWith #-}

424−mkStreamLen count = sourceUnfoldrM count 0

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

425−

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

426−{-# INLINE joinInner #-}

426+ -> Int

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

427+ -> Int

428+ -> Int

429+ -> m ()

428−joinInner val1 val2 _ =

430+joinWith j val1 val2 i =

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

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

430−

432+

431−{-# INLINE joinInnerMap #-}

433+{-# INLINE joinMapWith #-}

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

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

433−joinInnerMap val1 val2 _ =

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

434− S.drain $

436+ -> Int

435− Internal.joinInnerMap

437+ -> Int

436− (fmap toKvMap (mkStreamLen val1))

438+ -> Int

437− (fmap toKvMap (mkStreamLen val2))

439+ -> m ()

438−

440+joinMapWith j val1 val2 i =

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 $

441+ S.drain

448− Internal.joinOuterMap

442+ $ j

449− (fmap toKvMap (mkStreamLen val1))

443+ (fmap toKvMap (sourceUnfoldrM val1 i))

450− (fmap toKvMap (mkStreamLen val2))

444+ (fmap toKvMap (sourceUnfoldrM val2 i))

451

445

452o_n_heap_buffering :: Int -> [Benchmark]

446o_n_heap_buffering :: Int -> [Benchmark]

453o_n_heap_buffering value =

447o_n_heap_buffering value =

454 [ bgroup "buffered"

448 [ bgroup "buffered"

455 [

449 [

456− benchIOSrc1 "joinInner" (joinInner sqrtVal sqrtVal)

450+ benchIOSrc1 "joinInner"

451+ $ joinWith Internal.joinInner sqrtVal sqrtVal

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

452+ , benchIOSrc1 "joinInnerMap"

453+ $ joinMapWith Internal.joinInnerMap sqrtVal sqrtVal

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

454+ , benchIOSrc1 "joinOuter"

455+ $ joinWith Internal.joinOuter sqrtVal sqrtVal

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

456+ , benchIOSrc1 "joinOuterMap"

457+ $ joinMapWith Internal.joinOuterMap sqrtVal sqrtVal

460 ]

458 ]

461 ]

459 ]

462

460

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

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

385 else Stream.nil

386 Nothing -> return (a, Nothing)

387

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

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

389--

390-- Space: O(n)

391--

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

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

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

472--

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

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

474--

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

476−--

477−-- Space: O(n)

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

478--

476--

479-- Time: O(m + n)

477-- Time: O(m + n)

480--

478--

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

488 km1 <- kvFold $ IsStream.adapt s1

486 km1 <- kvFold $ IsStream.adapt s1

489 km2 <- kvFold $ IsStream.adapt s2

487 km2 <- kvFold $ IsStream.adapt s2

490

488

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) s1

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

492 where

493 joinAB km (k, a) =

494 case k `Map.lookup` km of

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

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

497

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.

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

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

499 where

503 where

500 joinAB km (k, b) =

504 joinAB km (k, b) =

501 case k `Map.lookup` km of

505 case k `Map.lookup` km of

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

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

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

85 assert (v1 == v2)

87−

88joinOuterMap :: Property

87joinOuterMap :: Property

89joinOuterMap =

88joinOuterMap =

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

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