2f96dda · streamly · git.iko.soy

git.iko.soy/streamly

Split stream based IO routines in an internal module

We are exporting the `writeS` stream based IO function via `Streamly.Internal` because it provides some extra perf boost over the fold based routine.

commit2f96ddab57f87c9c12e7d18966a4861ee210ed84 authorHarendra Kumar <harendra.kumar@gmail.com> dateMon 12 Aug 2019, 05:06 (7 years ago) parent8ce6a6d

4 files changed+124−61

compare with master

unifiedsplit

src/Streamly/FileSystem/Handle.hs+5−61 src/Streamly/FileSystem/Handle/Internal.hs+114 src/Streamly/Internal.hs+4 streamly.cabal+1

src/Streamly/FileSystem/Handle.hs+5−61

@@ -86,7 +86,6 @@ module Streamly.FileSystem.Handle

8686 -- -- * Array Write

8787 , writeArray

8888 , writeArrays

89− -- , writeArraysPackedUpto

9089 , writeArraysInChunksOf

9190

9291 -- -- * Random Access (Seek)

@@ -113,17 +112,17 @@ where

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

114113import Data.Word (Word8)

115114import Foreign.ForeignPtr (withForeignPtr)

116−import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

117−import Foreign.Ptr (minusPtr, plusPtr)

115+import Foreign.Ptr (plusPtr)

118116import Foreign.Storable (Storable(..))

119117import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

120−import System.IO (Handle, hGetBufSome, hPutBuf)

118+import System.IO (Handle, hGetBufSome)

121119import Prelude hiding (read)

122120

121+import Streamly.FileSystem.Handle.Internal (writeArray)

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

124−import Streamly.Streams.Serial (SerialT)

125123import Streamly.Streams.StreamK.Type (IsStream, mkStream)

126−import Streamly.Memory.Array.Types (defaultChunkSize, shrinkToFit, lpackArraysChunksOf)

124+import Streamly.Memory.Array.Types

125+ (defaultChunkSize, shrinkToFit, lpackArraysChunksOf)

127126import Streamly.Fold (Fold)

128127-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

129128

@@ -131,7 +130,6 @@ import qualified Streamly.Fold as FL

131130import qualified Streamly.Fold.Types as FL

132131import qualified Streamly.Memory.Array as A

133132import qualified Streamly.Memory.ArrayStream as AS

134−import qualified Streamly.Prelude as S

135133import qualified Streamly.Streams.StreamD.Type as D

136134

137135-------------------------------------------------------------------------------

@@ -170,22 +168,6 @@ readArrayUpto size h = do

170168 -- XXX shrink only if the diff is significant

171169 shrinkToFit v

172170

173−-------------------------------------------------------------------------------

174−-- Array IO (output)

175−-------------------------------------------------------------------------------

176−

177−-- | Write an 'Array' to a file handle.

178−--

179−-- @since 0.7.0

180−{-# INLINABLE writeArray #-}

181−writeArray :: Storable a => Handle -> Array a -> IO ()

182−writeArray _ arr | A.length arr == 0 = return ()

183−writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen

184− where

185− aLen =

186− let p = unsafeForeignPtrToPtr aStart

187− in aEnd `minusPtr` p

188−

189171-------------------------------------------------------------------------------

190172-- Stream of Arrays IO

191173-------------------------------------------------------------------------------

@@ -268,14 +250,6 @@ read = AS.flatten . readArrays

268250-- Writing

269251-------------------------------------------------------------------------------

270252

271−-- | Write a stream of arrays to a handle.

272−--

273−-- @since 0.7.0

274−{-# INLINE writeArraysS #-}

275−writeArraysS :: (MonadIO m, Storable a)

276− => Handle -> SerialT m (Array a) -> m ()

277−writeArraysS h m = S.mapM_ (liftIO . writeArray h) m

278−

279253-- | Write a stream of arrays to a handle.

280254--

281255-- @since 0.7.0

@@ -283,17 +257,6 @@ writeArraysS h m = S.mapM_ (liftIO . writeArray h) m

283257writeArrays :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()

284258writeArrays h = FL.drainBy (liftIO . writeArray h)

285259

286−-- | @writeArraysPackedUpto chunkSize handle stream@ writes a stream of arrays

287−-- to @handle@ after coalescing the adjacent arrays in chunks of @chunkSize@.

288−-- The chunk size is only a maximum and the actual writes could be smaller as

289−-- we do not split the arrays to fit exactly to the specified size.

290−--

291−-- @since 0.7.0

292−{-# INLINE _writeArraysPackedUpto #-}

293−_writeArraysPackedUpto :: (MonadIO m, Storable a)

294− => Int -> Handle -> SerialT m (Array a) -> m ()

295−_writeArraysPackedUpto n h xs = writeArraysS h $ AS.compact n xs

296−

297260-- | @writeArraysInChunksOf chunkSize handle@ writes a stream of arrays

298261-- to @handle@ after coalescing the adjacent arrays in chunks of @chunkSize@.

299262-- We never split an array, if a single array is bigger than the specified size

@@ -314,15 +277,6 @@ writeArraysInChunksOf n h = lpackArraysChunksOf n (writeArrays h)

314277-- do not want buffering to occur at GHC level as well. Same thing applies to

315278-- writes as well.

316279

317−-- | @writeInChunksOf chunkSize handle stream@ writes @stream@ to @handle@ in

318−-- chunks of @chunkSize@. A write is performed to the IO device as soon as we

319−-- collect the required input size.

320−--

321−-- @since 0.7.0

322−{-# INLINE writeInChunksOfS #-}

323−writeInChunksOfS :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()

324−writeInChunksOfS n h m = writeArraysS h $ AS.arraysOf n m

325−

326280-- | @writeInChunksOf chunkSize handle@ writes the input stream to @handle@ in

327281-- chunks of @chunkSize@. A write is performed to the IO device as soon as we

328282-- collect the required input size.

@@ -332,16 +286,6 @@ writeInChunksOfS n h m = writeArraysS h $ AS.arraysOf n m

332286writeInChunksOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()

333287writeInChunksOf n h = FL.lchunksOf n (A.writeN n) (writeArrays h)

334288

335−-- > write = 'writeInChunksOf' A.defaultChunkSize

336−--

337−-- | Write a byte stream to a file handle. Accumulates the input in chunks of

338−-- up to 'A.defaultChunkSize' before writing.

339−--

340−-- @since 0.7.0

341−{-# INLINE _writeS #-}

342−_writeS :: MonadIO m => Handle -> SerialT m Word8 -> m ()

343−_writeS = writeInChunksOfS defaultChunkSize

344−

345289-- > write = 'writeInChunksOf' A.defaultChunkSize

346290--

347291-- | Write a byte stream to a file handle. Accumulates the input in chunks of

src/Streamly/FileSystem/Handle/Internal.hsnew file+114

@@ -0,0 +1,114 @@

1+{-# LANGUAGE CPP #-}

2+{-# LANGUAGE BangPatterns #-}

3+{-# LANGUAGE MagicHash #-}

4+{-# LANGUAGE RecordWildCards #-}

5+{-# LANGUAGE UnboxedTuples #-}

7+#include "inline.hs"

9+-- |

10+-- Module : Streamly.FileSystem.Handle.Internal

12+--

13+-- License : BSD3

14+-- Maintainer : harendra.kumar@gmail.com

15+-- Stability : experimental

16+-- Portability : GHC

17+--

18+-- Streaming based IO write rotuines based on file handles.

19+

20+module Streamly.FileSystem.Handle.Internal

21+ (

22+ writeArray

23+

24+ -- Byte stream write

25+ , writeS

26+ , writeSInChunksOf

27+

28+ -- -- * Array stream Write

29+ , writeSArrays

30+ , writeSArraysInChunksOf

31+ )

32+where

33+

34+import Control.Monad.IO.Class (MonadIO(..))

35+import Data.Word (Word8)

36+import Foreign.ForeignPtr (withForeignPtr)

37+import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

38+import Foreign.Ptr (minusPtr)

39+import Foreign.Storable (Storable(..))

40+import System.IO (Handle, hPutBuf)

41+import Prelude hiding (read)

42+

43+import Streamly.Memory.Array.Types (Array(..))

44+import Streamly.Streams.Serial (SerialT)

45+import Streamly.Memory.Array.Types (defaultChunkSize)

46+

47+import qualified Streamly.Memory.Array as A

48+import qualified Streamly.Memory.ArrayStream as AS

49+import qualified Streamly.Prelude as S

50+

51+-------------------------------------------------------------------------------

52+-- Array IO (output)

53+-------------------------------------------------------------------------------

54+

55+-- | Write an 'Array' to a file handle.

56+--

57+-- @since 0.7.0

58+{-# INLINABLE writeArray #-}

59+writeArray :: Storable a => Handle -> Array a -> IO ()

60+writeArray _ arr | A.length arr == 0 = return ()

61+writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen

62+ where

63+ aLen =

64+ let p = unsafeForeignPtrToPtr aStart

65+ in aEnd `minusPtr` p

66+

67+-------------------------------------------------------------------------------

68+-- Stream of Arrays IO

69+-------------------------------------------------------------------------------

70+

71+-------------------------------------------------------------------------------

72+-- Writing

73+-------------------------------------------------------------------------------

74+

75+-- | Write a stream of arrays to a handle.

76+--

77+-- @since 0.7.0

78+{-# INLINE writeSArrays #-}

79+writeSArrays :: (MonadIO m, Storable a)

80+ => Handle -> SerialT m (Array a) -> m ()

81+writeSArrays h m = S.mapM_ (liftIO . writeArray h) m

82+

83+-- | @writeArraysPackedUpto chunkSize handle stream@ writes a stream of arrays

84+-- to @handle@ after coalescing the adjacent arrays in chunks of @chunkSize@.

85+-- The chunk size is only a maximum and the actual writes could be smaller as

86+-- we do not split the arrays to fit exactly to the specified size.

87+--

88+-- @since 0.7.0

89+{-# INLINE writeSArraysInChunksOf #-}

90+writeSArraysInChunksOf :: (MonadIO m, Storable a)

91+ => Int -> Handle -> SerialT m (Array a) -> m ()

92+writeSArraysInChunksOf n h xs = writeSArrays h $ AS.compact n xs

93+

94+-- | @writeSInChunksOf chunkSize handle stream@ writes @stream@ to @handle@ in

95+-- chunks of @chunkSize@. A write is performed to the IO device as soon as we

96+-- collect the required input size.

97+--

98+-- @since 0.7.0

99+{-# INLINE writeSInChunksOf #-}

100+writeSInChunksOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()

101+writeSInChunksOf n h m = writeSArrays h $ AS.arraysOf n m

102+

103+-- > write = 'writeInChunksOf' A.defaultChunkSize

104+--

105+-- | Write a byte stream to a file handle. Accumulates the input in chunks of

106+-- up to 'A.defaultChunkSize' before writing.

107+--

108+-- NOTE: This may perform better than the 'write' fold, you can try this if you

109+-- need some extra perf boost.

110+--

111+-- @since 0.7.0

112+{-# INLINE writeS #-}

113+writeS :: MonadIO m => Handle -> SerialT m Word8 -> m ()

114+writeS = writeSInChunksOf defaultChunkSize

src/Streamly/Internal.hs+4

@@ -66,11 +66,15 @@ module Streamly.Internal

6666 , duplicate

6767 , initialize

6868 , runStep

69+

70+ -- * Streamly.FileSystem.Handle

71+ , writeS

6972 )

7073where

7174

7275import Prelude hiding (break, span, splitAt)

7376

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

7478import Streamly.Streams.Combinators (inspectMode)

7579import Streamly.Streams.Parallel (tapAsync)

7680

streamly.cabal+1

@@ -219,6 +219,7 @@ library

219219 , Streamly.FileSystem.FDIO

220220 , Streamly.FileSystem.FD

221221 , Streamly.FileSystem.File

222+ , Streamly.FileSystem.Handle.Internal

222223

223224 -- Time

224225 , Streamly.Time.Units