107bebd · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Rename ArrayStream fold/parse operations

107bebdHarendra Kumar4 years ago

benchmark/Streamly/Benchmark/Data/Array/Stream/Foreign.hs+10−10

compare with master

unifiedsplit

@@ -25,7 +25,7 @@ module Main

26import Control.DeepSeq (NFData(..))

27import Control.Monad (void)

28−import Control.Monad.Catch (MonadThrow)

28+import Control.Monad.Catch (MonadCatch)

29import Data.Functor.Identity (runIdentity)

30import Data.Word (Word8)

31import System.IO (Handle)

@@ -36,7 +36,7 @@ import qualified Streamly.Prelude as Stream

36import qualified Streamly.Internal.Data.Array.Foreign as Array

37import qualified Streamly.Internal.Data.Array.Stream.Foreign as ArrayStream

38import qualified Streamly.Internal.Data.Fold as Fold

39−import qualified Streamly.Internal.Data.Parser.ParserD as ParserD

39+import qualified Streamly.Internal.Data.Parser as Parser

40import qualified Streamly.Internal.Data.Stream.IsStream as Stream (arraysOf)

41import qualified Streamly.Internal.FileSystem.Handle as Handle

42import qualified Streamly.Internal.Unicode.Stream as Unicode

@@ -218,8 +218,8 @@ o_1_space_copy_toChunks_group_ungroup env =

218-------------------------------------------------------------------------------

219

220{-# INLINE drainWhile #-}

221−drainWhile :: MonadThrow m => (a -> Bool) -> ParserD.Parser m a ()

221+drainWhile :: MonadCatch m => (a -> Bool) -> Parser.Parser m a ()

222−drainWhile p = ParserD.takeWhile p Fold.drain

222+drainWhile p = Parser.takeWhile p Fold.drain

223

224-------------------------------------------------------------------------------

225-- Folds and parsers

@@ -227,11 +227,11 @@ drainWhile p = ParserD.takeWhile p Fold.drain

227

228{-# INLINE fold #-}

229fold :: SerialT IO (Array.Array Int) -> IO ()

230−fold s = void $ ArrayStream.fold Fold.drain s

230+fold s = void $ ArrayStream.foldBreak Fold.drain s

231

232−{-# INLINE parseArray #-}

232+{-# INLINE parse #-}

233−parseArray :: Int -> SerialT IO (Array.Array Int) -> IO ()

233+parse :: Int -> SerialT IO (Array.Array Int) -> IO ()

234−parseArray value s = void $ ArrayStream.parse (drainWhile (< value)) s

234+parse value s = void $ ArrayStream.parseBreak (drainWhile (< value)) s

235

236o_1_space_serial_array ::

237 Int -> [Array.Array Int] -> [Array.Array Int] -> [Benchmark]

@@ -239,9 +239,9 @@ o_1_space_serial_array bound arraysSmall arraysBig =

239 [ benchIO "fold (of 100)" (\_ -> Stream.fromList arraysSmall) fold

240 , benchIO "fold (single)" (\_ -> Stream.fromList arraysBig) fold

241 , benchIO "parse (of 100)" (\_ -> Stream.fromList arraysSmall)

242− $ parseArray bound

242+ $ parse bound

243 , benchIO "parse (single)" (\_ -> Stream.fromList arraysBig)

244− $ parseArray bound

244+ $ parse bound

245 ]

246

247-------------------------------------------------------------------------------

core/src/Streamly/Internal/Data/Array/Stream/Fold/Foreign.hs+3−1

@@ -23,7 +23,9 @@

23-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream (arraysOf)

24-- >>> import qualified Streamly.Prelude as Stream

25--

26+-- >>> f = ArrayFold.fromFold (Fold.take 7 Fold.toList)

26−-- >>> ArrayStream.foldArr (ArrayFold.fromFold (Fold.take 7 Fold.toList)) $ Stream.arraysOf 5 $ Stream.fromList "hello world"

27+-- >>> s = Stream.arraysOf 5 $ Stream.fromList "hello world"

28+-- >>> ArrayStream.runArrayFold f s

27-- "hello w"

29-- "hello w"

28--

30--

29module Streamly.Internal.Data.Array.Stream.Fold.Foreign

31module Streamly.Internal.Data.Array.Stream.Fold.Foreign

src/Streamly/Internal/Data/Array/Stream/Foreign.hs+56−33

@@ -23,16 +23,16 @@ module Streamly.Internal.Data.Array.Stream.Foreign

24 -- * Elimination

25 -- ** Element Folds

26− , fold

26+ , foldBreak

27− , parse

27+ , parseBreak

28− , parseD

28+ , parseBreakD

30 -- ** Array Folds

31− , foldArr

31+ , runArrayFold

32− , foldArr_

32+ , runArrayFoldBreak

33 -- , parseArr

34− , parseArrD

34+ , runArrayParserDBreak

35− , foldArrMany

35+ , runArrayFoldMany

37 , toArray

@@ -82,6 +82,8 @@ import Streamly.Internal.Data.Array.Foreign.Mut.Type

83import qualified Streamly.Internal.Data.Array.Foreign as A

84import qualified Streamly.Internal.Data.Array.Foreign as Array

85+-- import qualified Streamly.Internal.Data.Array.Stream.Fold.Foreign

86+-- as ArrayFold

85import qualified Streamly.Internal.Data.Array.Foreign.Type as A

87import qualified Streamly.Internal.Data.Array.Foreign.Type as A

86import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA

88import qualified Streamly.Internal.Data.Array.Foreign.Mut.Type as MA

87import qualified Streamly.Internal.Data.Array.Stream.Mut.Foreign as AS

89import qualified Streamly.Internal.Data.Array.Stream.Mut.Foreign as AS

@@ -359,15 +361,20 @@ foldD (Fold fstep initial extract) stream@(D.Stream step state) = do

359-- | Fold an array stream using the supplied 'Fold'. Returns the fold result

361-- | Fold an array stream using the supplied 'Fold'. Returns the fold result

360-- and the unconsumed stream.

362-- and the unconsumed stream.

361--

363--

364+-- > foldBreak f = runArrayFoldBreak (ArrayFold.fromFold f)

365+--

362-- /Internal/

366-- /Internal/

363--

367--

364−{-# INLINE_NORMAL fold #-}

368+{-# INLINE_NORMAL foldBreak #-}

365−fold ::

369+foldBreak ::

366 (MonadIO m, Storable a)

370 (MonadIO m, Storable a)

367 => FL.Fold m a b

371 => FL.Fold m a b

368 -> SerialT m (A.Array a)

372 -> SerialT m (A.Array a)

369 -> m (b, SerialT m (A.Array a))

373 -> m (b, SerialT m (A.Array a))

370−fold f s = fmap fromStreamD <$> foldD f (toStreamD s)

374+foldBreak f s = fmap fromStreamD <$> foldD f (toStreamD s)

375+-- If fold performs better than runArrayFoldBreak we can rewrite runArrayFoldBreak to

376+-- fold.

377+-- fold f = runArrayFoldBreak (ArrayFold.fromFold f)

371

378

372-------------------------------------------------------------------------------

379-------------------------------------------------------------------------------

373-- Fold to a single Array

380-- Fold to a single Array

@@ -524,13 +531,15 @@ newtype List a = List {getList :: [a]}

524--

531--

525-- XXX This should be written using CPS (as parseK) if we want it to scale wrt

532-- XXX This should be written using CPS (as parseK) if we want it to scale wrt

526-- to the number of times it can be called on the same stream.

533-- to the number of times it can be called on the same stream.

527−{-# INLINE_NORMAL parseD #-}

534+{-# INLINE_NORMAL parseBreakD #-}

528−parseD ::

535+parseBreakD ::

529 forall m a b. (MonadIO m, MonadThrow m, Storable a)

536 forall m a b. (MonadIO m, MonadThrow m, Storable a)

530 => PRD.Parser m a b

537 => PRD.Parser m a b

531 -> D.Stream m (Array.Array a)

538 -> D.Stream m (Array.Array a)

532 -> m (b, D.Stream m (Array.Array a))

539 -> m (b, D.Stream m (Array.Array a))

540+parseBreakD

533−parseD (PRD.Parser pstep initial extract) stream@(D.Stream step state) = do

541+ (PRD.Parser pstep initial extract) stream@(D.Stream step state) = do

542+

534 res <- initial

543 res <- initial

535 case res of

544 case res of

536 PRD.IPartial s -> go SPEC state (List []) s

545 PRD.IPartial s -> go SPEC state (List []) s

@@ -608,25 +617,36 @@ parseD (PRD.Parser pstep initial extract) stream@(D.Stream step state) = do

608--

617--

609-- /Internal/

618-- /Internal/

610--

619--

611−{-# INLINE_NORMAL parse #-}

620+{-# INLINE_NORMAL parseBreak #-}

612−parse ::

621+parseBreak ::

613 (MonadIO m, MonadThrow m, Storable a)

622 (MonadIO m, MonadThrow m, Storable a)

614 => PR.Parser m a b

623 => PR.Parser m a b

615 -> SerialT m (A.Array a)

624 -> SerialT m (A.Array a)

616 -> m (b, SerialT m (A.Array a))

625 -> m (b, SerialT m (A.Array a))

626+parseBreak p s =

617−parse p s = fmap fromStreamD <$> parseD (PRD.fromParserK p) (toStreamD s)

627+ fmap fromStreamD <$> parseBreakD (PRD.fromParserK p) (toStreamD s)

618

628

619-------------------------------------------------------------------------------

629-------------------------------------------------------------------------------

620-- Elimination - Running Array Folds and parsers

630-- Elimination - Running Array Folds and parsers

621-------------------------------------------------------------------------------

631-------------------------------------------------------------------------------

622

632

633+-- | Note that this is not the same as using a @Parser m (Array a) b@ with the

634+-- regular "Streamly.Internal.Data.IsStream.parse" function. The regular parse

635+-- would consume the input arrays as single unit. This parser parses in the way

636+-- as described in the ArrayFold module. The input arrays are treated as @n@

637+-- element units and can be consumed partially. The remaining elements are

638+-- inserted in the source stream as an array.

639+--

623−{-# INLINE_NORMAL parseArrD #-}

640+{-# INLINE_NORMAL runArrayParserDBreak #-}

624−parseArrD ::

641+runArrayParserDBreak ::

625 forall m a b. (MonadIO m, MonadThrow m, Storable a)

642 forall m a b. (MonadIO m, MonadThrow m, Storable a)

626 => PRD.Parser m (Array a) b

643 => PRD.Parser m (Array a) b

627 -> D.Stream m (Array.Array a)

644 -> D.Stream m (Array.Array a)

628 -> m (b, D.Stream m (Array.Array a))

645 -> m (b, D.Stream m (Array.Array a))

646+runArrayParserDBreak

629−parseArrD (PRD.Parser pstep initial extract) stream@(D.Stream step state) = do

647+ (PRD.Parser pstep initial extract)

648+ stream@(D.Stream step state) = do

649+

630 res <- initial

650 res <- initial

631 case res of

651 case res of

632 PRD.IPartial s -> go SPEC state (List []) s

652 PRD.IPartial s -> go SPEC state (List []) s

@@ -695,26 +715,27 @@ parseArr ::

695 => ASF.Parser m a b

715 => ASF.Parser m a b

696 -> SerialT m (A.Array a)

716 -> SerialT m (A.Array a)

697 -> m (b, SerialT m (A.Array a))

717 -> m (b, SerialT m (A.Array a))

698−parseArr p s = fmap fromStreamD <$> parseD p (toStreamD s)

718+parseArr p s = fmap fromStreamD <$> parseBreakD p (toStreamD s)

699-}

719-}

700

720

701-- | Fold an array stream using the supplied array stream 'Fold'.

721-- | Fold an array stream using the supplied array stream 'Fold'.

702--

722--

703-- /Pre-release/

723-- /Pre-release/

704--

724--

705−{-# INLINE foldArr #-}

725+{-# INLINE runArrayFold #-}

706−foldArr :: (MonadIO m, MonadThrow m, Storable a) =>

726+runArrayFold :: (MonadIO m, MonadThrow m, Storable a) =>

707 ArrayFold m a b -> SerialT m (A.Array a) -> m b

727 ArrayFold m a b -> SerialT m (A.Array a) -> m b

708−foldArr (ArrayFold p) s = fst <$> parseArrD p (toStreamD s)

728+runArrayFold (ArrayFold p) s = fst <$> runArrayParserDBreak p (toStreamD s)

709

729

710-- | Like 'fold' but also returns the remaining stream.

730-- | Like 'fold' but also returns the remaining stream.

711--

731--

712-- /Pre-release/

732-- /Pre-release/

713--

733--

714−{-# INLINE foldArr_ #-}

734+{-# INLINE runArrayFoldBreak #-}

715−foldArr_ :: (MonadIO m, MonadThrow m, Storable a) =>

735+runArrayFoldBreak :: (MonadIO m, MonadThrow m, Storable a) =>

716 ArrayFold m a b -> SerialT m (A.Array a) -> m (b, SerialT m (A.Array a))

736 ArrayFold m a b -> SerialT m (A.Array a) -> m (b, SerialT m (A.Array a))

737+runArrayFoldBreak (ArrayFold p) s =

717−foldArr_ (ArrayFold p) s = second fromStreamD <$> parseArrD p (toStreamD s)

738+ second fromStreamD <$> runArrayParserDBreak p (toStreamD s)

718

739

719{-# ANN type ParseChunksState Fuse #-}

740{-# ANN type ParseChunksState Fuse #-}

720data ParseChunksState x inpBuf st pst =

741data ParseChunksState x inpBuf st pst =

@@ -724,13 +745,15 @@ data ParseChunksState x inpBuf st pst =

724 | ParseChunksBuf inpBuf st inpBuf !pst

745 | ParseChunksBuf inpBuf st inpBuf !pst

725 | ParseChunksYield x (ParseChunksState x inpBuf st pst)

746 | ParseChunksYield x (ParseChunksState x inpBuf st pst)

726

747

727−{-# INLINE_NORMAL foldArrManyD #-}

748+{-# INLINE_NORMAL runArrayFoldManyD #-}

728−foldArrManyD

749+runArrayFoldManyD

729 :: (MonadThrow m, Storable a)

750 :: (MonadThrow m, Storable a)

730 => ArrayFold m a b

751 => ArrayFold m a b

731 -> D.Stream m (Array a)

752 -> D.Stream m (Array a)

732 -> D.Stream m b

753 -> D.Stream m b

754+runArrayFoldManyD

733−foldArrManyD (ArrayFold (PRD.Parser pstep initial extract)) (D.Stream step state) =

755+ (ArrayFold (PRD.Parser pstep initial extract)) (D.Stream step state) =

756+

734 D.Stream stepOuter (ParseChunksInit [] state)

757 D.Stream stepOuter (ParseChunksInit [] state)

735

758

736 where

759 where

@@ -844,16 +867,16 @@ foldArrManyD (ArrayFold (PRD.Parser pstep initial extract)) (D.Stream step state

844

867

845 stepOuter _ (ParseChunksYield a next) = return $ D.Yield a next

868 stepOuter _ (ParseChunksYield a next) = return $ D.Yield a next

846

869

847−-- | Apply an array stream 'Fold' repeatedly on an array stream and emit the

870+-- | Apply an 'ArrayFold' repeatedly on an array stream and emit the

848-- fold outputs in the output stream.

871-- fold outputs in the output stream.

849--

872--

850-- See "Streamly.Prelude.foldMany" for more details.

873-- See "Streamly.Prelude.foldMany" for more details.

851--

874--

852-- /Pre-release/

875-- /Pre-release/

853−{-# INLINE foldArrMany #-}

876+{-# INLINE runArrayFoldMany #-}

854−foldArrMany

877+runArrayFoldMany

855 :: (IsStream t, MonadThrow m, Storable a)

878 :: (IsStream t, MonadThrow m, Storable a)

856 => ArrayFold m a b

879 => ArrayFold m a b

857 -> t m (Array a)

880 -> t m (Array a)

858 -> t m b

881 -> t m b

859−foldArrMany p m = fromStreamD $ foldArrManyD p (toStreamD m)

882+runArrayFoldMany p m = fromStreamD $ runArrayFoldManyD p (toStreamD m)