ddfe6bd · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add a "foldFilter" combinator to serial streams

ddfe6bdHarendra Kumar5 years ago

benchmark/Streamly/Benchmark/Prelude/Serial/Transformation.hs+9

compare with master

unifiedsplit

@@ -31,6 +31,7 @@ import Test.Inspection

31#endif

33import qualified Streamly.Prelude as S

34+import qualified Streamly.Internal.Data.Stream.Serial as Serial

34import qualified Streamly.Internal.Data.Stream.IsStream as Internal

35import qualified Streamly.Internal.Data.Stream.IsStream as Internal

35import qualified Streamly.Internal.Data.Fold as FL

36import qualified Streamly.Internal.Data.Fold as FL

36import qualified Streamly.Internal.Data.Unfold as Unfold

37import qualified Streamly.Internal.Data.Unfold as Unfold

@@ -288,6 +289,10 @@ filterMAllOut value n = composeN n $ S.filterM (\x -> return $ x > (value + 1))

288filterMAllIn :: MonadIO m => Int -> Int -> SerialT m Int -> m ()

289filterMAllIn :: MonadIO m => Int -> Int -> SerialT m Int -> m ()

289filterMAllIn value n = composeN n $ S.filterM (\x -> return $ x <= (value + 1))

290filterMAllIn value n = composeN n $ S.filterM (\x -> return $ x <= (value + 1))

290

291

292+{-# INLINE foldFilterEven #-}

293+foldFilterEven :: MonadIO m => Int -> SerialT m Int -> m ()

294+foldFilterEven n = composeN n $ Serial.foldFilter (FL.satisfy even)

295+

291{-# INLINE _takeOne #-}

296{-# INLINE _takeOne #-}

292_takeOne :: MonadIO m => Int -> SerialT m Int -> m ()

297_takeOne :: MonadIO m => Int -> SerialT m Int -> m ()

293_takeOne n = composeN n $ S.take 1

298_takeOne n = composeN n $ S.take 1

@@ -396,6 +401,8 @@ o_1_space_filtering value =

396 , benchIOSink value "filterM-all-out" (filterMAllOut value 1)

401 , benchIOSink value "filterM-all-out" (filterMAllOut value 1)

397 , benchIOSink value "filterM-all-in" (filterMAllIn value 1)

402 , benchIOSink value "filterM-all-in" (filterMAllIn value 1)

398

403

404+ , benchIOSink value "foldFilter-even" (foldFilterEven 1)

405+

399 -- Trimming

406 -- Trimming

400 , benchIOSink value "take-all" (takeAll value 1)

407 , benchIOSink value "take-all" (takeAll value 1)

401 , benchIOSink

408 , benchIOSink

@@ -441,6 +448,8 @@ o_1_space_filteringX4 value =

441 , benchIOSink value "filterM-all-out" (filterMAllOut value 4)

448 , benchIOSink value "filterM-all-out" (filterMAllOut value 4)

442 , benchIOSink value "filterM-all-in" (filterMAllIn value 4)

449 , benchIOSink value "filterM-all-in" (filterMAllIn value 4)

443

450

451+ , benchIOSink value "foldFilter-even" (foldFilterEven 4)

452+

444 -- trimming

453 -- trimming

445 , benchIOSink value "take-all" (takeAll value 4)

454 , benchIOSink value "take-all" (takeAll value 4)

446 , benchIOSink value "takeWhile-true" (takeWhileTrue value 4)

455 , benchIOSink value "takeWhile-true" (takeWhileTrue value 4)

core/src/Streamly/Internal/Data/Stream/Serial.hs+20

@@ -42,6 +42,7 @@ module Streamly.Internal.Data.Stream.Serial

42 -- * Transformation

43 , map

44 , mapM

45+ , foldFilter

46 -- * Deprecated

47 -- * Deprecated

47 , StreamT

48 , StreamT

@@ -72,12 +73,14 @@ import Text.Read

72 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec

73 ( Lexeme(Ident), lexP, parens, prec, readPrec, readListPrec

73 , readListPrecDefault)

74 , readListPrecDefault)

74import Streamly.Internal.BaseCompat ((#.), errorWithoutStackTrace, oneShot)

75import Streamly.Internal.BaseCompat ((#.), errorWithoutStackTrace, oneShot)

76+import Streamly.Internal.Data.Fold.Type (Fold)

75import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)

77import Streamly.Internal.Data.Maybe.Strict (Maybe'(..), toMaybe)

76import Streamly.Internal.Data.Stream.StreamK.Type

78import Streamly.Internal.Data.Stream.StreamK.Type

77 (Stream, mkStream, foldStream)

79 (Stream, mkStream, foldStream)

79import qualified Streamly.Internal.Data.Stream.Common as P

81import qualified Streamly.Internal.Data.Stream.Common as P

80import qualified Streamly.Internal.Data.Stream.StreamD.Generate as D

82import qualified Streamly.Internal.Data.Stream.StreamD.Generate as D

83+import qualified Streamly.Internal.Data.Stream.StreamD.Transform as D

81import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

84import qualified Streamly.Internal.Data.Stream.StreamD.Type as D

82import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

85import qualified Streamly.Internal.Data.Stream.StreamK.Type as K

@@ -259,6 +262,23 @@ NFDATA1_INSTANCE(SerialT)

259FOLDABLE_INSTANCE(SerialT)

262FOLDABLE_INSTANCE(SerialT)

260TRAVERSABLE_INSTANCE(SerialT)

263TRAVERSABLE_INSTANCE(SerialT)

261

264

265+{-# INLINE toStreamD #-}

266+toStreamD :: Applicative m => SerialT m a -> D.Stream m a

267+toStreamD (SerialT m) = D.fromStreamK m

268+

269+{-# INLINE fromStreamD #-}

270+fromStreamD :: Monad m => D.Stream m a -> SerialT m a

271+fromStreamD m = SerialT $ D.toStreamK m

272+

273+-- | Use a filtering fold on a stream.

274+--

275+-- > Stream.sum $ Stream.foldFilter (Fold.satisfy (> 5)) $ Stream.fromList [1..10]

276+-- 40

277+--

278+{-# INLINE foldFilter #-}

279+foldFilter :: Monad m => Fold m a (Maybe b) -> SerialT m a -> SerialT m b

280+foldFilter p = fromStreamD . D.foldFilter p . toStreamD

281+

262------------------------------------------------------------------------------

282------------------------------------------------------------------------------

263-- WSerialT

283-- WSerialT

264------------------------------------------------------------------------------

284------------------------------------------------------------------------------

core/src/Streamly/Internal/Data/Stream/StreamD/Transform.hs+14

@@ -70,6 +70,7 @@ module Streamly.Internal.Data.Stream.StreamD.Transform

70 -- | Produce a subset of the stream.

71 , filter

72 , filterM

73+ , foldFilter

73 , deleteBy

74 , deleteBy

74 , uniq

75 , uniq

@@ -741,6 +742,19 @@ scanl1' f = scanl1M' (\x y -> return (f x y))

741-- Filtering

742-- Filtering

742-------------------------------------------------------------------------------

743-------------------------------------------------------------------------------

743

744

745+-- XXX nested foldMany does not fuse, therefore, this may not be very useful as

746+-- a filter unless that is fixed.

747+--

748+-- | Use a filtering fold on a stream.

749+--

750+-- > Stream.sum $ Stream.foldFilter (Fold.satisfy (> 5)) $ Stream.fromList [1..10]

751+-- 40

752+--

753+-- /Pre-release/

754+{-# INLINE foldFilter #-}

755+foldFilter :: Monad m => Fold m a (Maybe b) -> Stream m a -> Stream m b

756+foldFilter f = catMaybes . foldMany f

757+

744-- Adapted from the vector package

758-- Adapted from the vector package

745{-# INLINE_NORMAL filterM #-}

759{-# INLINE_NORMAL filterM #-}

746filterM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a

760filterM :: Monad m => (a -> m Bool) -> Stream m a -> Stream m a