acc3871 · streamly · git.iko.soy

git.iko.soy/streamly

Fixed insertBy to use correct state. Add deleteBy.

- Benchmarks show very poor performance of insertBy and deleteBy compared to StreamK. Should be improved in StreamD.

commitacc38719fdbdbd153f8dd9eac5c2a62e96d736be authorPranay Sashank <pranaysashank@gmail.com> dateThu 15 Nov 2018, 14:26 (8 years ago) parent7f82af4

6 files changed+96−11

compare with master

unifiedsplit

benchmark/Linear.hs+8 benchmark/LinearOps.hs+7−2 src/Streamly/Prelude.hs+17 src/Streamly/Streams/StreamD.hs+39−8 src/Streamly/Streams/StreamK.hs+18 test/Prop.hs+7−1

benchmark/Linear.hs+8

@@ -138,6 +138,10 @@ main =

138138 , benchIOSink "findIndices" (Ops.findIndices 1)

139139 , benchIOSink "elemIndices" (Ops.elemIndices 1)

140140 -- , benchIOSink "concat" Ops.concat

141+ , benchIOSink "insertLarge" (Ops.insertLarge 1)

142+ , benchIOSink "insertSmall" (Ops.insertSmall 1)

143+ , benchIOSink "deleteLarge" (Ops.deleteLarge 1)

144+ , benchIOSink "deleteSmall" (Ops.deleteSmall 1)

141145 ]

142146 , bgroup "transformationX4"

143147 [ benchIOSink "scan" (Ops.scan 4)

@@ -151,6 +155,10 @@ main =

151155 -- Ops.sequence serially (Ops.sourceUnfoldrMAction n)

152156 , benchIOSink "findIndices" (Ops.findIndices 4)

153157 , benchIOSink "elemIndices" (Ops.elemIndices 4)

158+ , benchIOSink "insertLarge" (Ops.insertLarge 4)

159+ , benchIOSink "insertSmall" (Ops.insertSmall 4)

160+ , benchIOSink "deleteLarge" (Ops.deleteLarge 4)

161+ , benchIOSink "deleteSmall" (Ops.deleteSmall 4)

154162 -- , benchIOSink "concat" Ops.concat

155163 ]

156164 , bgroup "filtering"

benchmark/LinearOps.hs+7−2

@@ -15,7 +15,7 @@ import Control.Monad (when)

1515import Data.Functor.Identity (Identity, runIdentity)

1616import Data.Maybe (fromJust)

1717import Prelude

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

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

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

2020 maxBound, div, IO, compare)

2121import qualified Prelude as P

@@ -275,7 +275,8 @@ composeN' n f =

275275scan, scanl1, map, fmap, mapMaybe, filterEven, filterAllOut,

276276 filterAllIn, takeOne, takeAll, takeWhileTrue, takeWhileMTrue, dropOne,

277277 dropAll, dropWhileTrue, dropWhileMTrue, dropWhileFalse,

278− findIndices, elemIndices

278+ findIndices, elemIndices, insertLarge, insertSmall, deleteLarge,

279+ deleteSmall

279280 :: Monad m

280281 => Int -> Stream m Int -> m ()

281282

@@ -314,6 +315,10 @@ dropWhileMTrue n = composeN n $ S.dropWhileM (return . (<= maxValue))

314315dropWhileFalse n = composeN n $ S.dropWhile (> maxValue)

315316findIndices n = composeN n $ S.findIndices (== maxValue)

316317elemIndices n = composeN n $ S.elemIndices maxValue

318+insertLarge n = composeN n $ S.insertBy compare maxValue

319+insertSmall n = composeN n $ S.insertBy compare 0

320+deleteSmall n = composeN n $ S.deleteBy (==) 0

321+deleteLarge n = composeN n $ S.deleteBy (>=) maxValue

317322

318323-------------------------------------------------------------------------------

319324-- Iteration

src/Streamly/Prelude.hs+17

@@ -205,6 +205,9 @@ module Streamly.Prelude

205205 , intersperseM

206206 , insertBy

207207

208+ -- ** Deleting

209+ , deleteBy

210+

208211 -- * Reordering

209212 , reverse

210213

@@ -1072,11 +1075,25 @@ reverse m = fromStream $ go K.nil (toStream m)

10721075intersperseM :: (IsStream t, MonadAsync m) => m a -> t m a -> t m a

10731076intersperseM = K.intersperseM

10741077

1078+-- | Inserts @x@ before the first element in the Stream that is

1079+-- less than @x@, based on the user supplied comparison function.

1080+--

10751081{-# INLINE insertBy #-}

10761082insertBy ::

10771083 (IsStream t, Monad m) => (a -> a -> Ordering) -> a -> t m a -> t m a

10781084insertBy cmp x m = fromStreamS $ S.insertBy cmp x (toStreamS m)

10791085

1086+------------------------------------------------------------------------------

1087+-- Deleting

1088+------------------------------------------------------------------------------

1089+

1090+-- | Deletes the first occurence of @x@ in the Stream that satisfies the

1091+-- equality predicate.

1092+--

1093+{-# INLINE deleteBy #-}

1094+deleteBy :: (IsStream t, Monad m) => (a -> a -> Bool) -> a -> t m a -> t m a

1095+deleteBy cmp x m = fromStreamS $ S.deleteBy cmp x (toStreamS m)

1096+

10801097------------------------------------------------------------------------------

10811098-- Zipping

10821099------------------------------------------------------------------------------

src/Streamly/Streams/StreamD.hs+39−8

@@ -128,6 +128,9 @@ module Streamly.Streams.StreamD

128128 -- * Inserting

129129 , insertBy

130130

131+ -- * Deleting

132+ , deleteBy

133+

131134 -- ** Map and Filter

132135 , mapMaybe

133136 , mapMaybeM

@@ -867,22 +870,50 @@ sequence (Stream step state) = Stream step' state

867870

868871{-# INLINE_NORMAL insertBy #-}

869872insertBy :: Monad m => (a -> a -> Ordering) -> a -> Stream m a -> Stream m a

870−insertBy cmp a (Stream step state) = Stream step' (state, False)

873+insertBy cmp a (Stream step state) = Stream step' (state, False, Nothing)

871874 where

872− step' gst (st, False) = do

875+ {-# INLINE_LATE step' #-}

876+ step' gst (st, False, _) = do

873877 r <- step (rstState gst) st

874878 case r of

875879 Yield x s -> case cmp a x of

876− GT -> return $ Yield x (s, False)

877− _ -> return $ Yield a (st, True)

880+ GT -> return $ Yield x (s, False, Nothing)

881+ _ -> return $ Yield a (s, True, Just x)

882+ Skip s -> return $ Skip (s, False, Nothing)

883+ Stop -> return $ Yield a (st, True, Nothing)

884+

885+ step' _ (_, True, Nothing) = return Stop

886+

887+ step' gst (st, True, Just prev) = do

888+ r <- step (rstState gst) st

889+ case r of

890+ Yield x s -> return $ Yield prev (s, True, Just x)

891+ Skip s -> return $ Skip (s, True, Just prev)

892+ Stop -> return $ Yield prev (st, True, Nothing)

893+

894+------------------------------------------------------------------------------

895+-- Deleting

896+------------------------------------------------------------------------------

897+

898+{-# INLINE_NORMAL deleteBy #-}

899+deleteBy :: Monad m => (a -> a -> Bool) -> a -> Stream m a -> Stream m a

900+deleteBy eq x (Stream step state) = Stream step' (state, False)

901+ where

902+ {-# INLINE_LATE step' #-}

903+ step' gst (st, False) = do

904+ r <- step (rstState gst) st

905+ case r of

906+ Yield y s -> return $

907+ if eq x y then Skip (s, True) else Yield y (s, False)

878908 Skip s -> return $ Skip (s, False)

879− Stop -> return $ Yield a (st, True)

909+ Stop -> return Stop

910+

880911 step' gst (st, True) = do

881912 r <- step (rstState gst) st

882913 case r of

883− Yield x s -> return $ Yield x (s, True)

884− Skip s -> return $ Skip (s, True)

885− Stop -> return Stop

914+ Yield y s -> return $ Yield y (s, True)

915+ Skip s -> return $ Skip (s, True)

916+ Stop -> return Stop

886917

887918------------------------------------------------------------------------------

888919-- Transformation by Map and Filter

src/Streamly/Streams/StreamK.hs+18

@@ -126,6 +126,9 @@ module Streamly.Streams.StreamK

126126 , intersperseM

127127 , insertBy

128128

129+ -- ** Deleting

130+ , deleteBy

131+

129132 -- ** Map and Filter

130133 , mapMaybe

131134

@@ -990,6 +993,21 @@ insertBy cmp x m = fromStream $ go (toStream m)

990993 _ -> yld x (a `cons` r)

991994 in unStream m1 (rstState st) stop single yieldk

992995

996+------------------------------------------------------------------------------

997+-- Deleting

998+------------------------------------------------------------------------------

999+

1000+{-# INLINE deleteBy #-}

1001+deleteBy :: IsStream t => (a -> a -> Bool) -> a -> t m a -> t m a

1002+deleteBy eq x m = fromStream $ go (toStream m)

1003+ where

1004+ go m1 = Stream $ \st stp sng yld ->

1005+ let single a = if eq x a then stp else sng a

1006+ yieldk a r = if eq x a

1007+ then unStream r (rstState st) stp sng yld

1008+ else yld a (go r)

1009+ in unStream m1 (rstState st) stp single yieldk

1010+

9931011-------------------------------------------------------------------------------

9941012-- Map and Filter

9951013-------------------------------------------------------------------------------

test/Prop.hs+7−1

@@ -14,7 +14,7 @@ import Data.IORef (readIORef, modifyIORef, newIORef)

1414import Data.List

1515 (sort, foldl', scanl', findIndices, findIndex, elemIndices,

1616 elemIndex, find, insertBy, intersperse, foldl1', (\\),

17− maximumBy, minimumBy)

17+ maximumBy, minimumBy, deleteBy)

1818import Data.Maybe (mapMaybe)

1919import GHC.Word (Word8)

2020

@@ -384,6 +384,12 @@ transformOps constr desc eq t = do

384384 transform (insertBy compare 0) $ t . S.insertBy compare 0

385385 prop (desc <> " insertBy 4") $

386386 transform (insertBy compare 4) $ t . S.insertBy compare 4

387+ prop (desc <> " deleteBy (<=) maxBound") $

388+ transform (deleteBy (<=) maxBound) $ t . S.deleteBy (<=) maxBound

389+ prop (desc <> " deleteBy (<=) 0") $

390+ transform (deleteBy (<=) 0) $ t . S.deleteBy (<=) 0

391+ prop (desc <> " deleteBy (==) 4") $

392+ transform (deleteBy (==) 4) $ t . S.deleteBy (==) 4

387393

388394concurrentOps

389395 :: IsStream t