3339c20 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Add fold and parse for array streams

3339c20Harendra Kumar5 years ago

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

compare with master

unifiedsplit

@@ -23,18 +23,26 @@ module Main

23 main

24 ) where

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

27+import Control.Monad (void)

28+import Control.Monad.Catch (MonadThrow)

26import Data.Functor.Identity (runIdentity)

29import Data.Functor.Identity (runIdentity)

27import Data.Word (Word8)

30import Data.Word (Word8)

28import System.IO (Handle)

31import System.IO (Handle)

32+import System.Random (randomRIO)

29import Prelude hiding ()

33import Prelude hiding ()

31import qualified Streamly.Prelude as Stream

35import qualified Streamly.Prelude as Stream

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

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

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

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

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

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

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

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

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

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

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

37import Gauge hiding (env)

44import Gauge hiding (env)

45+import Streamly.Prelude (SerialT, MonadAsync, IsStream)

38import Streamly.Benchmark.Common

46import Streamly.Benchmark.Common

39import Streamly.Benchmark.Common.Handle

47import Streamly.Benchmark.Common.Handle

@@ -44,6 +52,28 @@ import Streamly.Internal.Data.Stream.StreamD.Type (Step(..))

44import Test.Inspection

52import Test.Inspection

45#endif

53#endif

55+-------------------------------------------------------------------------------

56+-- Utilities

57+-------------------------------------------------------------------------------

58+

59+-- XXX these can be moved to the common module

60+

61+{-# INLINE sourceUnfoldrM #-}

62+sourceUnfoldrM :: (IsStream t, MonadAsync m) => Int -> Int -> t m Int

63+sourceUnfoldrM value n = Stream.unfoldrM step n

64+ where

65+ step cnt =

66+ if cnt > n + value

67+ then return Nothing

68+ else return (Just (cnt, cnt + 1))

69+

70+{-# INLINE benchIO #-}

71+benchIO

72+ :: NFData b

73+ => String -> (Int -> t IO a) -> (t IO a -> IO b) -> Benchmark

74+benchIO name src sink =

75+ bench name $ nfIO $ randomRIO (1,1) >>= sink . src

76+

47-------------------------------------------------------------------------------

77-------------------------------------------------------------------------------

48-- read chunked using toChunks

78-- read chunked using toChunks

49-------------------------------------------------------------------------------

79-------------------------------------------------------------------------------

@@ -183,6 +213,37 @@ o_1_space_copy_toChunks_group_ungroup env =

183 ]

213 ]

184 ]

214 ]

185

215

216+-------------------------------------------------------------------------------

217+-- Parsers

218+-------------------------------------------------------------------------------

219+

220+{-# INLINE drainWhile #-}

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

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

223+

224+-------------------------------------------------------------------------------

225+-- Folds and parsers

226+-------------------------------------------------------------------------------

227+

228+{-# INLINE fold #-}

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

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

231+

232+{-# INLINE parseArray #-}

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

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

235+

236+o_1_space_serial_array ::

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

238+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

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

244+ $ parseArray bound

245+ ]

246+

186-------------------------------------------------------------------------------

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

187-- Driver

248-- Driver

188-------------------------------------------------------------------------------

249-------------------------------------------------------------------------------

@@ -193,13 +254,20 @@ moduleName = "Data.Array.Stream.Foreign"

193main :: IO ()

254main :: IO ()

194main = do

255main = do

195 env <- mkHandleBenchEnv

256 env <- mkHandleBenchEnv

196− defaultMain (allBenchmarks env)

257+ runWithCLIOptsEnv defaultStreamSize alloc (allBenchmarks env)

197

258

198 where

259 where

199

260

261+ alloc value = do

262+ small <- Stream.toList $ Stream.arraysOf 100 $ sourceUnfoldrM value 0

263+ big <- Stream.toList $ Stream.arraysOf value $ sourceUnfoldrM value 0

264+ return (small, big)

265+

200− allBenchmarks env =

266+ allBenchmarks env arrays value =

267+ let (arraysSmall, arraysBig) = arrays

201− [ bgroup (o_1_space_prefix moduleName)

268+ in [ bgroup (o_1_space_prefix moduleName) $ Prelude.concat

202− ( o_1_space_read_chunked env

269+ [ o_1_space_read_chunked env

270+ , o_1_space_serial_array value arraysSmall arraysBig

203− ++ o_1_space_copy_toChunks_group_ungroup env

271+ , o_1_space_copy_toChunks_group_ungroup env

204− )

272+ ]

205 ]

273 ]

hie.yaml+3−1

@@ -51,7 +51,9 @@ cradle:

51 - path: "./test/Streamly/Test/Data/Array.hs"

52 component: "test:Data.Array"

53 - path: "./test/Streamly/Test/Data/Array/Foreign.hs"

54+ component: "test:Data.Array.Foreign"

55+ - path: "./test/Streamly/Test/Data/Array/Stream/Foreign.hs"

54− component: "Data.Array.Foreign"

56+ component: "test:Data.Array.Stream.Foreign"

55 - path: "./test/Streamly/Test/Data/Fold.hs"

57 - path: "./test/Streamly/Test/Data/Fold.hs"

56 component: "test:Data.Fold"

58 component: "test:Data.Fold"

57 - path: "./test/Streamly/Test/Data/List.hs"

59 - path: "./test/Streamly/Test/Data/List.hs"

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

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

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

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

25--

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

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

27-- "hello w"

28--

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

src/Streamly/Internal/Data/Array/Stream/Foreign.hs+173−3

@@ -22,6 +22,11 @@ module Streamly.Internal.Data.Array.Stream.Foreign

22 , unlines

24 -- * Elimination

25+ -- ** Element Folds

26+ , fold

27+ , parse

28+ , parseD

29+

25 -- ** Array Folds

30 -- ** Array Folds

26 , foldArr

31 , foldArr

27 , foldArr_

32 , foldArr_

@@ -50,6 +55,9 @@ import Data.Bifunctor (second)

50import Control.Exception (assert)

55import Control.Exception (assert)

51import Control.Monad.Catch (MonadThrow, throwM)

56import Control.Monad.Catch (MonadThrow, throwM)

52import Control.Monad.IO.Class (MonadIO(..))

57import Control.Monad.IO.Class (MonadIO(..))

58+#if __GLASGOW_HASKELL__ < 808

59+import Data.Semigroup (Semigroup(..))

60+#endif

53import Data.Word (Word8)

61import Data.Word (Word8)

54import Foreign.ForeignPtr (touchForeignPtr)

62import Foreign.ForeignPtr (touchForeignPtr)

55import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

63import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

@@ -317,7 +325,65 @@ splitOnSuffix byte s =

317 D.fromStreamD $ D.splitInnerBySuffix (A.breakOn byte) A.spliceTwo $ D.toStreamD s

325 D.fromStreamD $ D.splitInnerBySuffix (A.breakOn byte) A.spliceTwo $ D.toStreamD s

318

326

319-------------------------------------------------------------------------------

327-------------------------------------------------------------------------------

320−-- Elimination

328+-- Elimination - Running folds

329+-------------------------------------------------------------------------------

330+

331+-- XXX This should be written using CPS (as foldK) if we want it to scale wrt

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

333+--

334+{-# INLINE_NORMAL foldD #-}

335+foldD :: forall m a b. (MonadIO m, Storable a) =>

336+ Fold m a b -> D.Stream m (Array a) -> m (b, D.Stream m (Array a))

337+foldD (Fold fstep initial extract) stream@(D.Stream step state) = do

338+ res <- initial

339+ case res of

340+ FL.Partial fs -> go SPEC state fs

341+ FL.Done fb -> return $! (fb, stream)

342+

343+ where

344+

345+ {-# INLINE go #-}

346+ go !_ st !fs = do

347+ r <- step defState st

348+ case r of

349+ D.Yield (Array (ForeignPtr start contents) (Ptr end)) s ->

350+ goArray SPEC s (ForeignPtr end contents) (Ptr start) fs

351+ D.Skip s -> go SPEC s fs

352+ D.Stop -> do

353+ b <- extract fs

354+ return (b, D.nil)

355+

356+ goArray !_ s fp@(ForeignPtr end _) !cur !fs

357+ | cur == Ptr end = do

358+ liftIO $ touchForeignPtr fp

359+ go SPEC s fs

360+ goArray !_ st fp@(ForeignPtr end contents) !cur !fs = do

361+ x <- liftIO $ peek cur

362+ res <- fstep fs x

363+ let elemSize = sizeOf (undefined :: a)

364+ next = cur `plusPtr` elemSize

365+ case res of

366+ FL.Done b -> do

367+ let !(Ptr curAddr) = next

368+ arr = Array (ForeignPtr curAddr contents) (Ptr end)

369+ return $! (b, D.cons arr (D.Stream step st))

370+ FL.Partial fs1 -> goArray SPEC st fp next fs1

371+

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

373+-- and the unconsumed stream.

374+--

375+-- /Internal/

376+--

377+{-# INLINE_NORMAL fold #-}

378+fold ::

379+ (MonadIO m, Storable a)

380+ => FL.Fold m a b

381+ -> SerialT m (A.Array a)

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

383+fold f s = fmap D.fromStreamD <$> foldD f (D.toStreamD s)

384+

385+-------------------------------------------------------------------------------

386+-- Fold to a single Array

321-------------------------------------------------------------------------------

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

322

388

323-- When we have to take an array partially, take the last part of the array.

389-- When we have to take an array partially, take the last part of the array.

@@ -459,7 +525,7 @@ _toArraysOf n = FL.chunksOf n (A.writeNF n) FL.toStream

459-}

525-}

460

526

461-------------------------------------------------------------------------------

527-------------------------------------------------------------------------------

462−-- Parsing

528+-- Elimination - running element parsers

463-------------------------------------------------------------------------------

529-------------------------------------------------------------------------------

464

530

465-- GHC parser does not accept {-# ANN type [] NoSpecConstr #-}, so we need

531-- GHC parser does not accept {-# ANN type [] NoSpecConstr #-}, so we need

@@ -467,6 +533,110 @@ _toArraysOf n = FL.chunksOf n (A.writeNF n) FL.toStream

467{-# ANN type List NoSpecConstr #-}

533{-# ANN type List NoSpecConstr #-}

468newtype List a = List {getList :: [a]}

534newtype List a = List {getList :: [a]}

469

535

536+-- This can be generalized to any type provided it can be unfolded to a stream

537+-- and it can be combined using a semigroup operation.

538+--

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

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

541+{-# INLINE_NORMAL parseD #-}

542+parseD ::

543+ forall m a b. (MonadIO m, MonadThrow m, Storable a)

544+ => PRD.Parser m a b

545+ -> D.Stream m (Array.Array a)

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

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

548+ res <- initial

549+ case res of

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

551+ PRD.IDone b -> return (b, stream)

552+ PRD.IError err -> throwM $ ParseError err

553+

554+ where

555+

556+ -- "backBuf" contains last few items in the stream that we may have to

557+ -- backtrack to.

558+ --

559+ -- XXX currently we are using a dumb list based approach for backtracking

560+ -- buffer. This can be replaced by a sliding/ring buffer using Data.Array.

561+ -- That will allow us more efficient random back and forth movement.

562+ {-# INLINE go #-}

563+ go !_ st backBuf !pst = do

564+ r <- step defState st

565+ case r of

566+ D.Yield (Array (ForeignPtr start contents) (Ptr end)) s ->

567+ gobuf SPEC s backBuf (ForeignPtr end contents) (Ptr start) pst

568+ D.Skip s -> go SPEC s backBuf pst

569+ D.Stop -> do

570+ b <- extract pst

571+ return (b, D.nil)

572+

573+ -- Use strictness on "cur" to keep it unboxed

574+ gobuf !_ s backBuf fp@(ForeignPtr end _) !cur !pst

575+ | cur == Ptr end = do

576+ liftIO $ touchForeignPtr fp

577+ go SPEC s backBuf pst

578+ gobuf !_ s backBuf fp@(ForeignPtr end contents) !cur !pst = do

579+ x <- liftIO $ peek cur

580+ pRes <- pstep pst x

581+ let elemSize = sizeOf (undefined :: a)

582+ next = cur `plusPtr` elemSize

583+ case pRes of

584+ PR.Partial 0 pst1 ->

585+ gobuf SPEC s (List []) fp next pst1

586+ PR.Partial n pst1 -> do

587+ assert (n <= Prelude.length (x:getList backBuf)) (return ())

588+ let src0 = Prelude.take n (x:getList backBuf)

589+ arr0 = A.fromListN n (Prelude.reverse src0)

590+ !(Ptr curAddr) = next

591+ arr1 = Array (ForeignPtr curAddr contents) (Ptr end)

592+ src = arr0 <> arr1

593+ let !(Array (ForeignPtr start cont1) (Ptr end1)) = src

594+ gobuf SPEC s (List []) (ForeignPtr end1 cont1) (Ptr start) pst1

595+ PR.Continue 0 pst1 ->

596+ gobuf SPEC s (List (x:getList backBuf)) fp next pst1

597+ PR.Continue n pst1 -> do

598+ assert (n <= Prelude.length (x:getList backBuf)) (return ())

599+ let (src0, buf1) = splitAt n (x:getList backBuf)

600+ arr0 = A.fromListN n (Prelude.reverse src0)

601+ !(Ptr curAddr) = next

602+ arr1 = Array (ForeignPtr curAddr contents) (Ptr end)

603+ src = arr0 <> arr1

604+ let !(Array (ForeignPtr start cont1) (Ptr end1)) = src

605+ gobuf SPEC s (List buf1) (ForeignPtr end1 cont1) (Ptr start) pst1

606+ PR.Done 0 b -> do

607+ let !(Ptr curAddr) = next

608+ arr = Array (ForeignPtr curAddr contents) (Ptr end)

609+ return (b, D.cons arr (D.Stream step s))

610+ PR.Done n b -> do

611+ assert (n <= Prelude.length (x:getList backBuf)) (return ())

612+ let src0 = Prelude.take n (x:getList backBuf)

613+ -- XXX create the array in reverse instead

614+ arr0 = A.fromListN n (Prelude.reverse src0)

615+ !(Ptr curAddr) = next

616+ arr1 = Array (ForeignPtr curAddr contents) (Ptr end)

617+ -- XXX Use StreamK to avoid adding arbitrary layers of

618+ -- constructors every time.

619+ str = D.cons arr0 (D.cons arr1 (D.Stream step s))

620+ return (b, str)

621+ PR.Error err -> throwM $ ParseError err

622+

623+-- | Parse an array stream using the supplied 'Parser'. Returns the parse

624+-- result and the unconsumed stream. Throws 'ParseError' if the parse fails.

625+--

626+-- /Internal/

627+--

628+{-# INLINE_NORMAL parse #-}

629+parse ::

630+ (MonadIO m, MonadThrow m, Storable a)

631+ => PRD.Parser m a b

632+ -> SerialT m (A.Array a)

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

634+parse p s = fmap D.fromStreamD <$> parseD p (D.toStreamD s)

635+

636+-------------------------------------------------------------------------------

637+-- Elimination - Running Array Folds and parsers

638+-------------------------------------------------------------------------------

639+

470{-# INLINE_NORMAL parseArrD #-}

640{-# INLINE_NORMAL parseArrD #-}

471parseArrD ::

641parseArrD ::

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

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

@@ -540,7 +710,7 @@ parseArrD (PRD.Parser pstep initial extract) stream@(D.Stream step state) = do

540{-# INLINE parseArr #-}

710{-# INLINE parseArr #-}

541parseArr ::

711parseArr ::

542 (MonadIO m, MonadThrow m, Storable a)

712 (MonadIO m, MonadThrow m, Storable a)

543− => PRD.Parser m a b

713+ => ASF.Parser m a b

544 -> SerialT m (A.Array a)

714 -> SerialT m (A.Array a)

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

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

546parseArr p s = fmap D.fromStreamD <$> parseD p (D.toStreamD s)

716parseArr p s = fmap D.fromStreamD <$> parseD p (D.toStreamD s)

streamly.cabal+1

@@ -123,6 +123,7 @@ extra-source-files:

123 test/Streamly/Test/Data/Array/Prim.hs

124 test/Streamly/Test/Data/Array/Prim/Pinned.hs

125 test/Streamly/Test/Data/Array/Foreign.hs

126+ test/Streamly/Test/Data/Array/Stream/Foreign.hs

126 test/Streamly/Test/Data/Parser/ParserD.hs

127 test/Streamly/Test/Data/Parser/ParserD.hs

127 test/Streamly/Test/FileSystem/Event.hs

128 test/Streamly/Test/FileSystem/Event.hs

128 test/Streamly/Test/FileSystem/Handle.hs

129 test/Streamly/Test/FileSystem/Handle.hs

test/Streamly/Test/Data/Array/Stream/Foreign.hsnew file+72

@@ -0,0 +1,72 @@

1+module Main (main) where

3+import Streamly.Test.Common (listEquals, chooseInt)

4+import Test.Hspec (hspec, describe)

5+import Test.Hspec.QuickCheck

6+import Test.QuickCheck (forAll, Property, vectorOf, Gen)

7+import Test.QuickCheck.Monadic (monadicIO, run)

9+import qualified Streamly.Internal.Data.Array.Foreign as Array

10+import qualified Streamly.Internal.Data.Array.Stream.Foreign as ArrayStream

11+import qualified Streamly.Internal.Data.Fold as Fold

12+import qualified Streamly.Internal.Data.Parser.ParserD as ParserD

13+import qualified Streamly.Internal.Data.Stream.IsStream as Stream

14+import qualified Test.Hspec as Hspec

15+

16+import Prelude hiding (sequence)

17+

18+#if MIN_VERSION_QuickCheck(2,14,0)

19+

20+import Test.QuickCheck (chooseAny)

21+

22+#else

23+

24+import System.Random (Random(random))

25+import Test.QuickCheck.Gen (Gen(MkGen))

26+

27+-- | Generates a random element over the natural range of `a`.

28+chooseAny :: Random a => Gen a

29+chooseAny = MkGen (\r _ -> let (x,_) = random r in x)

30+

31+#endif

32+

33+maxTestCount :: Int

34+maxTestCount = 100

35+

36+parse :: Property

37+parse = do

38+ let len = 200

39+ -- ls = input list (stream)

40+ -- clen = chunk size

41+ -- tlen = parser take size

42+ forAll

43+ ((,,)

44+ <$> vectorOf len (chooseAny :: Gen Int)

45+ <*> chooseInt (1, len)

46+ <*> chooseInt (0, len))

47+ $ \(ls, clen, tlen) ->

48+ monadicIO $ do

49+ (ls1, str) <-

50+ let input =

51+ Stream.chunksOf

52+ clen (Array.writeN clen) (Stream.fromList ls)

53+ parser = ParserD.fromFold (Fold.take tlen Fold.toList)

54+ in run $ ArrayStream.parse parser input

55+ ls2 <- run $ Stream.toList $ ArrayStream.concat str

56+ listEquals (==) (ls1 ++ ls2) ls

57+

58+-------------------------------------------------------------------------------

59+-- Main

60+-------------------------------------------------------------------------------

61+

62+moduleName :: String

63+moduleName = "Data.Array.Stream.Foreign"

64+

65+main :: IO ()

66+main =

67+ hspec $

68+ Hspec.parallel $

69+ modifyMaxSuccess (const maxTestCount) $ do

70+ describe moduleName $ do

71+ describe "Stream parsing" $ do

72+ prop "parse" parse

test/streamly-tests.cabal+5

@@ -222,6 +222,11 @@ test-suite Data.Array.Foreign

222 main-is: Streamly/Test/Data/Array/Foreign.hs

223 ghc-options: -main-is Streamly.Test.Data.Array.Foreign.main

224

225+test-suite Data.Array.Stream.Foreign

226+ import: test-options

227+ type: exitcode-stdio-1.0

228+ main-is: Streamly/Test/Data/Array/Stream/Foreign.hs

229+

225test-suite Data.Fold

230test-suite Data.Fold

226 import: test-options

231 import: test-options

227 type: exitcode-stdio-1.0

232 type: exitcode-stdio-1.0