d01942d · streamly · git.iko.soy

benchmark/LinearAsync.hs+1

compare with master

unifiedsplit

@@ -100,5 +100,6 @@ main =

100 -- Zip has only one parallel flavor

101 , benchIO "zip" Ops.zipAsync

102 , benchIO "zipM" Ops.zipAsyncM

103+ , benchIO "zipAp" Ops.zipAsyncAp

103 ]

104 ]

105 ]

benchmark/LinearOps.hs+10−2

@@ -19,7 +19,7 @@ import Data.Maybe (fromJust)

19import Prelude

20 (Monad, Int, (+), ($), (.), return, fmap, even, (>), (<=), (==), (>=),

21 subtract, undefined, Maybe(..), odd, Bool, not, (>>=), mapM_, curry,

22− maxBound, div, IO, compare, Double, fromIntegral, Integer)

22+ maxBound, div, IO, compare, Double, fromIntegral, Integer, (<$>), (<*>))

23import qualified Prelude as P

24import qualified Data.Foldable as F

25import qualified GHC.Exts as GHC

@@ -435,17 +435,25 @@ stripPrefix src = do

435

436{-# INLINE zipAsync #-}

437{-# INLINE zipAsyncM #-}

438+{-# INLINE zipAsyncAp #-}

438−zipAsync, zipAsyncM :: S.MonadAsync m => Stream m Int -> m ()

439+zipAsync, zipAsyncAp, zipAsyncM :: S.MonadAsync m => Stream m Int -> m ()

439

440

440zipAsync src = do

441zipAsync src = do

441 r <- S.tail src

442 r <- S.tail src

442 let src1 = fromJust r

443 let src1 = fromJust r

443 transform (S.zipAsyncWith (,) src src1)

444 transform (S.zipAsyncWith (,) src src1)

445+

444zipAsyncM src = do

446zipAsyncM src = do

445 r <- S.tail src

447 r <- S.tail src

446 let src1 = fromJust r

448 let src1 = fromJust r

447 transform (S.zipAsyncWithM (curry return) src src1)

449 transform (S.zipAsyncWithM (curry return) src src1)

448

450

451+zipAsyncAp src = do

452+ r <- S.tail src

453+ let src1 = fromJust r

454+ transform (S.zipAsyncly $ (,) <$> S.serially src

455+ <*> S.serially src1)

456+

449{-# INLINE eqBy #-}

457{-# INLINE eqBy #-}

450eqBy :: (Monad m, P.Eq a) => Stream m a -> m P.Bool

458eqBy :: (Monad m, P.Eq a) => Stream m a -> m P.Bool

451eqBy src = S.eqBy (==) src src

459eqBy src = S.eqBy (==) src src

benchmark/Nested.hs+12−6

@@ -23,7 +23,8 @@ main =

23 -- TBD Study scaling with 10, 100, 1000 loop iterations

24 defaultMain

25 [ bgroup "serially"

26+ [ benchIO "toNullAp" $ Ops.toNullAp serially

26− [ benchIO "toNull" $ Ops.toNull serially

27+ , benchIO "toNull" $ Ops.toNull serially

27 , benchIO "toList" $ Ops.toList serially

28 , benchIO "toList" $ Ops.toList serially

28 -- , benchIO "toListSome" $ Ops.toListSome serially

29 -- , benchIO "toListSome" $ Ops.toListSome serially

29 , benchIO "filterAllOut" $ Ops.filterAllOut serially

30 , benchIO "filterAllOut" $ Ops.filterAllOut serially

@@ -33,7 +34,8 @@ main =

33 ]

34 ]

34

35

35 , bgroup "wSerially"

36 , bgroup "wSerially"

37+ [ benchIO "toNullAp" $ Ops.toNullAp wSerially

36− [ benchIO "toNull" $ Ops.toNull wSerially

38+ , benchIO "toNull" $ Ops.toNull wSerially

37 , benchIO "toList" $ Ops.toList wSerially

39 , benchIO "toList" $ Ops.toList wSerially

38 -- , benchIO "toListSome" $ Ops.toListSome wSerially

40 -- , benchIO "toListSome" $ Ops.toListSome wSerially

39 , benchIO "filterAllOut" $ Ops.filterAllOut wSerially

41 , benchIO "filterAllOut" $ Ops.filterAllOut wSerially

@@ -43,7 +45,8 @@ main =

43 ]

45 ]

44

46

45 , bgroup "aheadly"

47 , bgroup "aheadly"

48+ [ benchIO "toNullAp" $ Ops.toNullAp aheadly

46− [ benchIO "toNull" $ Ops.toNull aheadly

49+ , benchIO "toNull" $ Ops.toNull aheadly

47 , benchIO "toList" $ Ops.toList aheadly

50 , benchIO "toList" $ Ops.toList aheadly

48 -- , benchIO "toListSome" $ Ops.toListSome aheadly

51 -- , benchIO "toListSome" $ Ops.toListSome aheadly

49 , benchIO "filterAllOut" $ Ops.filterAllOut aheadly

52 , benchIO "filterAllOut" $ Ops.filterAllOut aheadly

@@ -53,7 +56,8 @@ main =

53 ]

56 ]

54

57

55 , bgroup "asyncly"

58 , bgroup "asyncly"

59+ [ benchIO "toNullAp" $ Ops.toNullAp asyncly

56− [ benchIO "toNull" $ Ops.toNull asyncly

60+ , benchIO "toNull" $ Ops.toNull asyncly

57 , benchIO "toList" $ Ops.toList asyncly

61 , benchIO "toList" $ Ops.toList asyncly

58 -- , benchIO "toListSome" $ Ops.toListSome asyncly

62 -- , benchIO "toListSome" $ Ops.toListSome asyncly

59 , benchIO "filterAllOut" $ Ops.filterAllOut asyncly

63 , benchIO "filterAllOut" $ Ops.filterAllOut asyncly

@@ -63,7 +67,8 @@ main =

63 ]

67 ]

64

68

65 , bgroup "wAsyncly"

69 , bgroup "wAsyncly"

70+ [ benchIO "toNullAp" $ Ops.toNullAp wAsyncly

66− [ benchIO "toNull" $ Ops.toNull wAsyncly

71+ , benchIO "toNull" $ Ops.toNull wAsyncly

67 , benchIO "toList" $ Ops.toList wAsyncly

72 , benchIO "toList" $ Ops.toList wAsyncly

68 -- , benchIO "toListSome" $ Ops.toListSome wAsyncly

73 -- , benchIO "toListSome" $ Ops.toListSome wAsyncly

69 , benchIO "filterAllOut" $ Ops.filterAllOut wAsyncly

74 , benchIO "filterAllOut" $ Ops.filterAllOut wAsyncly

@@ -73,7 +78,8 @@ main =

73 ]

78 ]

74

79

75 , bgroup "parallely"

80 , bgroup "parallely"

81+ [ benchIO "toNullAp" $ Ops.toNullAp parallely

76− [ benchIO "toNull" $ Ops.toNull parallely

82+ , benchIO "toNull" $ Ops.toNull parallely

77 , benchIO "toList" $ Ops.toList parallely

83 , benchIO "toList" $ Ops.toList parallely

78 --, benchIO "toListSome" $ Ops.toListSome parallely

84 --, benchIO "toListSome" $ Ops.toListSome parallely

79 , benchIO "filterAllOut" $ Ops.filterAllOut parallely

85 , benchIO "filterAllOut" $ Ops.filterAllOut parallely

benchmark/NestedOps.hs+7

@@ -62,6 +62,13 @@ runToList = S.toList

62-- Benchmark ops

63-------------------------------------------------------------------------------

64

65+{-# INLINE toNullAp #-}

66+toNullAp

67+ :: (S.IsStream t, S.MonadAsync m, Monad (t m))

68+ => (t m Int -> S.SerialT m Int) -> Int -> m ()

69+toNullAp t start = runStream . t $

70+ (+) <$> source start prodCount <*> source start prodCount

71+

65{-# INLINE toNull #-}

72{-# INLINE toNull #-}

66toNull

73toNull

67 :: (S.IsStream t, S.MonadAsync m, Monad (t m))

74 :: (S.IsStream t, S.MonadAsync m, Monad (t m))

src/Streamly/Streams/Ahead.hs+9−1

@@ -686,9 +686,17 @@ instance MonadAsync m => Monad (AheadT m) where

686 return = pure

687 (>>=) = bindAhead

688

689+{-# INLINE apAhead #-}

690+{-# SPECIALIZE apAhead :: AheadT IO (a -> b) -> AheadT IO a -> AheadT IO b #-}

691+apAhead :: MonadAsync m => AheadT m (a -> b) -> AheadT m a -> AheadT m b

692+apAhead mf m = ap (K.adapt mf) (K.adapt m)

693+

694+instance (Monad m, MonadAsync m) => Applicative (AheadT m) where

695+ pure = AheadT . K.yield

696+ (<*>) = apAhead

697+

689------------------------------------------------------------------------------

698------------------------------------------------------------------------------

690-- Other instances

699-- Other instances

691------------------------------------------------------------------------------

700------------------------------------------------------------------------------

692

701

693−MONAD_APPLICATIVE_INSTANCE(AheadT,MONADPARALLEL)

694MONAD_COMMON_INSTANCES(AheadT, MONADPARALLEL)

702MONAD_COMMON_INSTANCES(AheadT, MONADPARALLEL)

src/Streamly/Streams/Async.hs+17−5

@@ -692,13 +692,19 @@ instance MonadAsync m => Monad (AsyncT m) where

692 return = pure

693 (>>=) = bindAsync

694

695+{-# INLINE apAsync #-}

696+{-# SPECIALIZE apAsync :: AsyncT IO (a -> b) -> AsyncT IO a -> AsyncT IO b #-}

697+apAsync :: MonadAsync m => AsyncT m (a -> b) -> AsyncT m a -> AsyncT m b

698+apAsync mf m = ap (adapt mf) (adapt m)

699+

700+instance (Monad m, MonadAsync m) => Applicative (AsyncT m) where

701+ pure = AsyncT . K.yield

702+ (<*>) = apAsync

703+

695------------------------------------------------------------------------------

704------------------------------------------------------------------------------

696-- Other instances

705-- Other instances

697------------------------------------------------------------------------------

706------------------------------------------------------------------------------

698

707

699−-- XXX SPECIALIZE the applicative as well?

700−

701−MONAD_APPLICATIVE_INSTANCE(AsyncT,MONADPARALLEL)

702MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)

708MONAD_COMMON_INSTANCES(AsyncT, MONADPARALLEL)

703

709

704------------------------------------------------------------------------------

710------------------------------------------------------------------------------

@@ -821,11 +827,17 @@ instance MonadAsync m => Monad (WAsyncT m) where

821 return = pure

827 return = pure

822 (>>=) = bindWAsync

828 (>>=) = bindWAsync

823

829

824−-- XXX SPECIALIZE the applicative as well?

830+{-# INLINE apWAsync #-}

831+{-# SPECIALIZE apWAsync :: WAsyncT IO (a -> b) -> WAsyncT IO a -> WAsyncT IO b #-}

832+apWAsync :: MonadAsync m => WAsyncT m (a -> b) -> WAsyncT m a -> WAsyncT m b

833+apWAsync mf m = ap (adapt mf) (adapt m)

834+

835+instance (Monad m, MonadAsync m) => Applicative (WAsyncT m) where

836+ pure = WAsyncT . K.yield

837+ (<*>) = apWAsync

825

838

826------------------------------------------------------------------------------

839------------------------------------------------------------------------------

827-- Other instances

840-- Other instances

828------------------------------------------------------------------------------

841------------------------------------------------------------------------------

829

842

830−MONAD_APPLICATIVE_INSTANCE(WAsyncT,MONADPARALLEL)

831MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)

843MONAD_COMMON_INSTANCES(WAsyncT, MONADPARALLEL)

src/Streamly/Streams/Parallel.hs+2

@@ -380,6 +380,8 @@ instance MonadAsync m => Monad (ParallelT m) where

380 return = pure

381 (>>=) = bindParallel

382

383+-- XXX Specialize the applicative instance

384+

383------------------------------------------------------------------------------

385------------------------------------------------------------------------------

384-- Other instances

386-- Other instances

385------------------------------------------------------------------------------

387------------------------------------------------------------------------------