da29139 · streamly · git.iko.soy

benchmark/FileIO.hs+6

compare with master

unifiedsplit

@@ -133,6 +133,12 @@ main = do

133 $ SS.foldLines FL.drain

134 $ SS.decodeChar8

135 $ FH.read inh

136+ , mkBench "linecountU" href $ do

137+ Handles inh _ <- readIORef href

138+ S.length

139+ $ SS.foldLines FL.drain

140+ $ SS.decodeChar8

141+ $ Internal.concatMapU Internal.readU (FH.readArrays inh)

136 , mkBench "sum" href $ do

142 , mkBench "sum" href $ do

137 Handles inh _ <- readIORef href

143 Handles inh _ <- readIORef href

138 S.sum $ FH.read inh

144 S.sum $ FH.read inh

src/Streamly/Internal.hs+10

@@ -52,6 +52,8 @@ module Streamly.Internal

52 , classifySessionsBy

53 , classifyKeepAliveSessions

54

55+ , concatMapU

56+

55 -- * Streamly.Fold Experimental Exports

57 -- * Streamly.Fold Experimental Exports

56 , Fold (..)

58 , Fold (..)

57 , lmap

59 , lmap

@@ -67,6 +69,12 @@ module Streamly.Internal

67 , initialize

69 , initialize

68 , runStep

70 , runStep

69

71

72+ -- * Streamly.Unfold Experimental Exports

73+ , Unfold (..)

74+

75+ -- * Streamly.Memory.Array

76+ , readU

77+

70 -- * Streamly.FileSystem.Handle

78 -- * Streamly.FileSystem.Handle

71 , writeS

79 , writeS

72 )

80 )

@@ -75,8 +83,10 @@ where

75import Prelude hiding (break, span, splitAt)

83import Prelude hiding (break, span, splitAt)

76

84

77import Streamly.FileSystem.Handle.Internal (writeS)

85import Streamly.FileSystem.Handle.Internal (writeS)

86+import Streamly.Memory.Array.Types (readU)

78import Streamly.Streams.Combinators (inspectMode)

87import Streamly.Streams.Combinators (inspectMode)

79import Streamly.Streams.Parallel (tapAsync)

88import Streamly.Streams.Parallel (tapAsync)

89+import Streamly.Unfold.Types (Unfold(..))

80

90

81import Streamly.Prelude.Internal

91import Streamly.Prelude.Internal

82import Streamly.Fold.Types

92import Streamly.Fold.Types

src/Streamly/Memory/Array/Types.hs+27

@@ -62,6 +62,7 @@ module Streamly.Memory.Array.Types

62 , toList

63 , writeN

64 , read

65+ , readU

65

66

66 -- * Utilities

67 -- * Utilities

67 , defaultChunkSize

68 , defaultChunkSize

@@ -100,6 +101,7 @@ import GHC.IO (IO(IO), unsafePerformIO)

100import GHC.Ptr (Ptr(..))

101import GHC.Ptr (Ptr(..))

101

102

102import Streamly.Fold.Types (Fold(..))

103import Streamly.Fold.Types (Fold(..))

104+import Streamly.Unfold.Types (Unfold(..))

103import Streamly.Strict (Tuple'(..))

105import Streamly.Strict (Tuple'(..))

104import Streamly.SVar (adaptState)

106import Streamly.SVar (adaptState)

105

107

@@ -388,6 +390,31 @@ byteCapacity Array{..} =

388 len = aBound `minusPtr` p

390 len = aBound `minusPtr` p

389 in assert (len >= 0) len

391 in assert (len >= 0) len

390

392

393+{-# INLINE_NORMAL readU #-}

394+readU :: forall m a. (Monad m, Storable a) => Unfold m (Array a) a

395+readU = Unfold step inject

396+ where

397+

398+ -- Note we have repurposed the Array type here

399+ inject Array{..} =

400+ return $ Array aStart (unsafeForeignPtrToPtr aStart) aEnd

401+

402+ {-# INLINE_LATE step #-}

403+ step (Array _ p end) | p == end = return D.Stop

404+ step (Array start p end) = do

405+ -- unsafeInlineIO allows us to run this in Identity monad for pure

406+ -- toList/foldr case which makes them much faster due to not

407+ -- accumulating the list and fusing better with the pure consumers.

408+ --

409+ -- This should be safe as the array contents are guaranteed to be

410+ -- evaluated/written to before we peek at them.

411+ let !x = unsafeInlineIO $ do

412+ r <- peek p

413+ touchForeignPtr start

414+ return r

415+ return $ D.Yield x

416+ (Array start (p `plusPtr` (sizeOf (undefined :: a))) end)

417+

391{-# INLINE_NORMAL toStreamD #-}

418{-# INLINE_NORMAL toStreamD #-}

392toStreamD :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a

419toStreamD :: forall m a. (Monad m, Storable a) => Array a -> D.Stream m a

393toStreamD Array{..} =

420toStreamD Array{..} =

src/Streamly/Memory/ArrayStream.hs+2

@@ -68,6 +68,8 @@ import qualified Streamly.Streams.Prelude as P

68{-# INLINE concat #-}

69concat :: (IsStream t, MonadIO m, Storable a) => t m (Array a) -> t m a

70concat m = D.fromStreamD $ A.flattenArrays (D.toStreamD m)

71+-- concat m = D.fromStreamD $ D.concatMap A.toStreamD (D.toStreamD m)

72+-- concat m = D.fromStreamD $ D.concatMapU A.readU (D.toStreamD m)

71

73

72-- XXX should we have a reverseArrays API to reverse the stream of arrays

74-- XXX should we have a reverseArrays API to reverse the stream of arrays

73-- instead?

75-- instead?

src/Streamly/Prelude/Internal.hs+11

@@ -231,6 +231,7 @@ module Streamly.Prelude.Internal

231

232 -- ** Nested Streams

233 , concatMapM

234+ , concatMapU

234 , concatMap

235 , concatMap

235 , concatMapWith

236 , concatMapWith

236 -- , interposeBy

237 -- , interposeBy

@@ -379,6 +380,7 @@ import qualified System.IO as IO

379

380

380import Streamly.Enumeration (Enumerable(..), enumerate, enumerateTo)

381import Streamly.Enumeration (Enumerable(..), enumerate, enumerateTo)

381import Streamly.Fold.Types (Fold (..))

382import Streamly.Fold.Types (Fold (..))

383+import Streamly.Unfold.Types (Unfold)

382import Streamly.Memory.Array.Types (Array)

384import Streamly.Memory.Array.Types (Array)

383-- import Streamly.Memory.Ring (Ring)

385-- import Streamly.Memory.Ring (Ring)

384import Streamly.SVar (MonadAsync, defState)

386import Streamly.SVar (MonadAsync, defState)

@@ -2099,6 +2101,15 @@ concatMap f m = fromStreamD $ D.concatMap (toStreamD . f) (toStreamD m)

2099concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b

2101concatMapM :: (IsStream t, Monad m) => (a -> m (t m b)) -> t m a -> t m b

2100concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)

2102concatMapM f m = fromStreamD $ D.concatMapM (fmap toStreamD . f) (toStreamD m)

2101

2103

2104+-- | Like 'concatMap' but uses an 'Unfold' for stream generation. Unlike

2105+-- 'concatMap' this can fuse the 'Unfold' code with the inner loop and

2106+-- therefore provide many times better performance.

2107+--

2108+-- @since 0.7.0

2109+{-# INLINE concatMapU #-}

2110+concatMapU ::(IsStream t, Monad m) => Unfold m a b -> t m a -> t m b

2111+concatMapU u m = fromStreamD $ D.concatMapU u (toStreamD m)

2112+

2102{-

2113{-

2103-- | A generalization of insertBy to inserting sequences.

2114-- | A generalization of insertBy to inserting sequences.

2104interposeBy :: (IsStream t, Monad m)

2115interposeBy :: (IsStream t, Monad m)

src/Streamly/Unfold.hs+4

@@ -2,6 +2,7 @@

2{-# LANGUAGE CPP #-}

3{-# LANGUAGE ExistentialQuantification #-}

4{-# LANGUAGE FlexibleContexts #-}

5+{-# LANGUAGE PatternSynonyms #-}

5{-# LANGUAGE RankNTypes #-}

6{-# LANGUAGE RankNTypes #-}

6{-# LANGUAGE RecordWildCards #-}

7{-# LANGUAGE RecordWildCards #-}

7{-# LANGUAGE ScopedTypeVariables #-}

8{-# LANGUAGE ScopedTypeVariables #-}

@@ -34,6 +35,9 @@ module Streamly.Unfold

34where

35where

35

36

36import Streamly.Streams.StreamD.Type (Stream(..), Step(..))

37import Streamly.Streams.StreamD.Type (Stream(..), Step(..))

38+#if __GLASGOW_HASKELL__ < 800

39+import Streamly.Streams.StreamD.Type (pattern Stream)

40+#endif

37import Streamly.Unfold.Types (Unfold(..))

41import Streamly.Unfold.Types (Unfold(..))

38

42

39-------------------------------------------------------------------------------

43-------------------------------------------------------------------------------