2f96dda · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Split stream based IO routines in an internal module

2f96ddaHarendra Kumar7 years ago

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

compare with master

unifiedsplit

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

86 -- -- * Array Write

87 , writeArray

88 , writeArrays

89− -- , writeArraysPackedUpto

90 , writeArraysInChunksOf

89 , writeArraysInChunksOf

92 -- -- * Random Access (Seek)

91 -- -- * Random Access (Seek)

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

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

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

114import Data.Word (Word8)

113import Data.Word (Word8)

115import Foreign.ForeignPtr (withForeignPtr)

114import Foreign.ForeignPtr (withForeignPtr)

116−import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

115+import Foreign.Ptr (plusPtr)

117−import Foreign.Ptr (minusPtr, plusPtr)

118import Foreign.Storable (Storable(..))

116import Foreign.Storable (Storable(..))

119import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

117import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

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

118+import System.IO (Handle, hGetBufSome)

121import Prelude hiding (read)

119import Prelude hiding (read)

122

120

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

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

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

124−import Streamly.Streams.Serial (SerialT)

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

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

124+import Streamly.Memory.Array.Types

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

125+ (defaultChunkSize, shrinkToFit, lpackArraysChunksOf)

127import Streamly.Fold (Fold)

126import Streamly.Fold (Fold)

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

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

129

128

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

131import qualified Streamly.Fold.Types as FL

130import qualified Streamly.Fold.Types as FL

132import qualified Streamly.Memory.Array as A

131import qualified Streamly.Memory.Array as A

133import qualified Streamly.Memory.ArrayStream as AS

132import qualified Streamly.Memory.ArrayStream as AS

134−import qualified Streamly.Prelude as S

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

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

136

134

137-------------------------------------------------------------------------------

135-------------------------------------------------------------------------------

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

170 -- XXX shrink only if the diff is significant

168 -- XXX shrink only if the diff is significant

171 shrinkToFit v

169 shrinkToFit v

172

170

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−

189-------------------------------------------------------------------------------

171-------------------------------------------------------------------------------

190-- Stream of Arrays IO

172-- Stream of Arrays IO

191-------------------------------------------------------------------------------

173-------------------------------------------------------------------------------

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

268-- Writing

250-- Writing

269-------------------------------------------------------------------------------

251-------------------------------------------------------------------------------

270

252

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−

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

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

280--

254--

281-- @since 0.7.0

255-- @since 0.7.0

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

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

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

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

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

285

259

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−

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

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

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

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

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

262-- 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)

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

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

315-- writes as well.

278-- writes as well.

316

279

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−

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

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

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

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

328-- collect the required input size.

282-- collect the required input size.

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

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

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

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

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

334

288

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−

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

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

346--

290--

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

291-- | 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

66 , duplicate

67 , initialize

68 , runStep

69+

70+ -- * Streamly.FileSystem.Handle

71+ , writeS

69 )

72 )

70where

73where

72import Prelude hiding (break, span, splitAt)

75import Prelude hiding (break, span, splitAt)

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

74import Streamly.Streams.Combinators (inspectMode)

78import Streamly.Streams.Combinators (inspectMode)

75import Streamly.Streams.Parallel (tapAsync)

79import Streamly.Streams.Parallel (tapAsync)

streamly.cabal+1

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

219 , Streamly.FileSystem.FDIO

220 , Streamly.FileSystem.FD

221 , Streamly.FileSystem.File

222+ , Streamly.FileSystem.Handle.Internal

222

223

223 -- Time

224 -- Time

224 , Streamly.Time.Units

225 , Streamly.Time.Units