6f517f7 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Consolidate FileSystem.Handle with Internal.FileSystem.Handle

6f517f7Harendra Kumar7 years ago

src/Streamly/FileSystem/Handle.hs+1−366

compare with master

unifiedsplit

@@ -109,370 +109,5 @@ module Streamly.FileSystem.Handle

109 )

110where

111

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

112+import Streamly.Internal.FileSystem.Handle

113−import Data.Word (Word8)

114−import Foreign.ForeignPtr (withForeignPtr)

115−import Foreign.Ptr (plusPtr)

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

117−import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

118−import System.IO (Handle, hGetBufSome)

119import Prelude hiding (read)

113import Prelude hiding (read)

120−

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

122−import Streamly.Internal.Memory.Array.Types (Array(..), writeNUnsafe)

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

124−import Streamly.Internal.Memory.Array.Types

125− (defaultChunkSize, shrinkToFit, lpackArraysChunksOf)

126−import Streamly.Data.Fold (Fold)

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

128−

129−import qualified Streamly.Data.Fold as FL

130−import qualified Streamly.Internal.Data.Fold.Types as FL

131−import qualified Streamly.Memory.Array as A

132−import qualified Streamly.Internal.Memory.ArrayStream as AS

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

134−

135−-------------------------------------------------------------------------------

136−-- References

137−-------------------------------------------------------------------------------

138−--

139−-- The following references may be useful to build an understanding about the

140−-- file API design:

141−--

142−-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.

143−-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations

144−-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.

145−-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.

146−-- https://www.w3.org/TR/FileAPI/ for http file API.

147−

148−-------------------------------------------------------------------------------

149−-- Array IO (Input)

150−-------------------------------------------------------------------------------

151−

152−-- | Read a 'ByteArray' from a file handle. If no data is available on the

153−-- handle it blocks until some data becomes available. If data is available

154−-- then it immediately returns that data without blocking. It reads a maximum

155−-- of up to the size requested.

156−{-# INLINABLE readArrayUpto #-}

157−readArrayUpto :: Int -> Handle -> IO (Array Word8)

158−readArrayUpto size h = do

159− ptr <- mallocPlainForeignPtrBytes size

160− -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))

161− withForeignPtr ptr $ \p -> do

162− n <- hGetBufSome h p size

163− let v = Array

164− { aStart = ptr

165− , aEnd = p `plusPtr` n

166− , aBound = p `plusPtr` size

167− }

168− -- XXX shrink only if the diff is significant

169− shrinkToFit v

170−

171−-------------------------------------------------------------------------------

172−-- Stream of Arrays IO

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

174−

175−-- | @readArraysOf size h@ reads a stream of arrays from file handle @h@.

176−-- The maximum size of a single array is specified by @size@. The actual size

177−-- read may be less than or equal to @size@.

178−{-# INLINABLE _readArraysOf #-}

179−_readArraysOf :: (IsStream t, MonadIO m)

180− => Int -> Handle -> t m (Array Word8)

181−_readArraysOf size h = go

182− where

183− -- XXX use cons/nil instead

184− go = mkStream $ \_ yld _ stp -> do

185− arr <- liftIO $ readArrayUpto size h

186− if A.length arr == 0

187− then stp

188− else yld arr go

189−

190−-- | @readArraysOf size handle@ reads a stream of arrays from the file

191−-- handle @handle@. The maximum size of a single array is limited to @size@.

192−-- The actual size read may be less than or equal to @size@.

193−--

194−-- @since 0.7.0

195−{-# INLINE_NORMAL readArraysOf #-}

196−readArraysOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)

197−readArraysOf size h = D.fromStreamD (D.Stream step ())

198− where

199− {-# INLINE_LATE step #-}

200− step _ _ = do

201− arr <- liftIO $ readArrayUpto size h

202− return $

203− case A.length arr of

204− 0 -> D.Stop

205− _ -> D.Yield arr ()

206−

207−-- XXX read 'Array a' instead of Word8

208−--

209−-- | @readArrays handle@ reads a stream of arrays from the specified file

210−-- handle. The maximum size of a single array is limited to

211−-- @defaultChunkSize@. The actual size read may be less than or equal to

212−-- @defaultChunkSize@.

213−--

214−-- > readArrays = readArraysOf defaultChunkSize

215−--

216−-- @since 0.7.0

217−{-# INLINE readArrays #-}

218−readArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)

219−readArrays = readArraysOf defaultChunkSize

220−

221−-------------------------------------------------------------------------------

222−-- Read File to Stream

223−-------------------------------------------------------------------------------

224−

225−-- TODO for concurrent streams implement readahead IO. We can send multiple

226−-- read requests at the same time. For serial case we can use async IO. We can

227−-- also control the read throughput in mbps or IOPS.

228−

229−-- | @readInChunksOf chunkSize handle@ reads a byte stream from a file

230−-- handle, reads are performed in chunks of up to @chunkSize@.

231−--

232−-- @since 0.7.0

233−{-# INLINE readInChunksOf #-}

234−readInChunksOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8

235−readInChunksOf chunkSize h = AS.concat $ readArraysOf chunkSize h

236−

237−-- TODO

238−-- Generate a stream of elements of the given type from a file 'Handle'.

239−-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a

240−--

241−-- > read = 'readInChunksOf' A.defaultChunkSize

242−-- | Generate a byte stream from a file 'Handle'.

243−--

244−-- @since 0.7.0

245−{-# INLINE read #-}

246−read :: (IsStream t, MonadIO m) => Handle -> t m Word8

247−read = AS.concat . readArrays

248−

249−-------------------------------------------------------------------------------

250−-- Writing

251−-------------------------------------------------------------------------------

252−

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

254−--

255−-- @since 0.7.0

256−{-# INLINE writeArrays #-}

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

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

259−

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

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

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

263−-- it emitted as it is. Multiple arrays are coalesed as long as the total size

264−-- remains below the specified size.

265−--

266−-- @since 0.7.0

267−{-# INLINE writeArraysInChunksOf #-}

268−writeArraysInChunksOf :: (MonadIO m, Storable a)

269− => Int -> Handle -> Fold m (Array a) ()

270−writeArraysInChunksOf n h = lpackArraysChunksOf n (writeArrays h)

271−

272−-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

273−--

274−-- XXX test this

275−-- Note that if you use a chunk size less than 8K (GHC's default buffer

276−-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you

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

278−-- writes as well.

279−

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

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

282−-- collect the required input size.

283−--

284−-- @since 0.7.0

285−{-# INLINE writeInChunksOf #-}

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

287−writeInChunksOf n h = FL.lchunksOf n (writeNUnsafe n) (writeArrays h)

288−

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

290−--

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

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

293−--

294−-- @since 0.7.0

295−{-# INLINE write #-}

296−write :: MonadIO m => Handle -> Fold m Word8 ()

297−write = writeInChunksOf defaultChunkSize

298−

299−{-

300−{-# INLINE write #-}

301−write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()

302−write = toHandleWith A.defaultChunkSize

303−-}

304−

305−-- XXX mmap a file into an array. This could be useful for in-place operations

306−-- on a file. For example, we can quicksort the contents of a file by mmapping

307−-- it.

308−

309−-------------------------------------------------------------------------------

310−-- IO with encoding/decoding Unicode characters

311−-------------------------------------------------------------------------------

312−

313−{-

314−-- |

315−-- > readUtf8 = decodeUtf8 . read

316−--

317−-- Read a UTF8 encoded stream of unicode characters from a file handle.

318−--

319−-- @since 0.7.0

320−{-# INLINE readUtf8 #-}

321−readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char

322−readUtf8 = decodeUtf8 . read

323−

324−-- |

325−-- > writeUtf8 h s = write h $ encodeUtf8 s

326−--

327−-- Encode a stream of unicode characters to UTF8 and write it to the given file

328−-- handle. Default block buffering applies to the writes.

329−--

330−-- @since 0.7.0

331−{-# INLINE writeUtf8 #-}

332−writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()

333−writeUtf8 h s = write h $ encodeUtf8 s

334−

335−-- | Write a stream of unicode characters after encoding to UTF-8 in chunks

336−-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds

337−-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written

338−-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'

339−-- option.

340−--

341−-- @since 0.7.0

342−{-# INLINE writeUtf8ByLines #-}

343−writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()

344−writeUtf8ByLines = undefined

345−

346−-- | Read UTF-8 lines from a file handle and apply the specified fold to each

347−-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.

348−--

349−-- @since 0.7.0

350−{-# INLINE readLines #-}

351−readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b

352−readLines h f = foldLines (readUtf8 h) f

353−

354−-------------------------------------------------------------------------------

355−-- Framing on a sequence

356−-------------------------------------------------------------------------------

357−

358−-- | Read a stream from a file handle and split it into frames delimited by

359−-- the specified sequence of elements. The supplied fold is applied on each

360−-- frame.

361−--

362−-- @since 0.7.0

363−{-# INLINE readFrames #-}

364−readFrames :: (IsStream t, MonadIO m, Storable a)

365− => Array a -> Handle -> Fold m a b -> t m b

366−readFrames = undefined -- foldFrames . read

367−

368−-- | Write a stream to the given file handle buffering up to frames separated

369−-- by the given sequence or up to a maximum of @defaultChunkSize@.

370−--

371−-- @since 0.7.0

372−{-# INLINE writeByFrames #-}

373−writeByFrames :: (IsStream t, MonadIO m, Storable a)

374− => Array a -> Handle -> t m a -> m ()

375−writeByFrames = undefined

376−

377−-------------------------------------------------------------------------------

378−-- Framing by time

379−-------------------------------------------------------------------------------

380−

381−-- | Write collecting the input in sessions of n seconds or if chunkSize

382−-- gets exceeded.

383−{-# INLINE writeByChunksOrSessionsOf #-}

384−writeByChunksOrSessionsOf :: MonadIO m

385− => Int -> Double -> Handle -> SerialT m Word8 -> m ()

386−writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined

387−

388−-- | Write collecting the input in sessions of n seconds or if defaultChunkSize

389−-- gets exceeded.

390−{-# INLINE writeBySessionsOf #-}

391−writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()

392−writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n

393−

394−-------------------------------------------------------------------------------

395−-- Random Access IO (Seek)

396−-------------------------------------------------------------------------------

397−

398−-- XXX handles could be shared, so we may not want to use the handle state at

399−-- all for these APIs. we can use pread and pwrite instead. On windows we will

400−-- need to use readFile/writeFile with an offset argument.

401−

402−-------------------------------------------------------------------------------

403−

404−-- | Read the element at the given index treating the file as an array.

405−--

406−-- @since 0.7.0

407−{-# INLINE readIndex #-}

408−readIndex :: Storable a => Handle -> Int -> Maybe a

409−readIndex arr i = undefined

410−

411−-- NOTE: To represent a range to read we have chosen (start, size) instead of

412−-- (start, end). This helps in removing the ambiguity of whether "end" is

413−-- included in the range or not.

414−--

415−-- We could avoid specifying the range to be read and instead use "take size"

416−-- on the stream, but it may end up reading more and then consume it partially.

417−

418−-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes

419−-- from @handle@ starting at the offset @pos@ from the beginning of the file.

420−--

421−-- Reads are performed in chunks of size @chunkSize@. For block devices, to

422−-- avoid reading partial blocks @chunkSize@ must align with the block size of

423−-- the underlying device. If the underlying block size is unknown, it is a good

424−-- idea to keep it a multiple 4KiB. This API ensures that the start of each

425−-- chunk is aligned with @chunkSize@ from second chunk onwards.

426−--

427−{-# INLINE readSliceWith #-}

428−readSliceWith :: (IsStream t, MonadIO m, Storable a)

429− => Int -> Handle -> Int -> Int -> t m a

430−readSliceWith chunkSize h pos len = undefined

431−

432−-- | @readSlice h i count@ streams a slice from the file handle @h@ starting

433−-- at index @i@ and reading up to @count@ elements in the forward direction

434−-- ending at the index @i + count - 1@.

435−--

436−-- @since 0.7.0

437−{-# INLINE readSlice #-}

438−readSlice :: (IsStream t, MonadIO m, Storable a)

439− => Handle -> Int -> Int -> t m a

440−readSlice = readSliceWith A.defaultChunkSize

441−

442−-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting

443−-- at index @i@ and reading up to @count@ elements in the reverse direction

444−-- ending at the index @i - count + 1@.

445−--

446−-- @since 0.7.0

447−{-# INLINE readSliceRev #-}

448−readSliceRev :: (IsStream t, MonadIO m, Storable a)

449− => Handle -> Int -> Int -> t m a

450−readSliceRev h i count = undefined

451−

452−-- | Write the given element at the given index in the file.

453−--

454−-- @since 0.7.0

455−{-# INLINE writeIndex #-}

456−writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()

457−writeIndex h i a = undefined

458−

459−-- | @writeSlice h i count stream@ writes a stream to the file handle @h@

460−-- starting at index @i@ and writing up to @count@ elements in the forward

461−-- direction ending at the index @i + count - 1@.

462−--

463−-- @since 0.7.0

464−{-# INLINE writeSlice #-}

465−writeSlice :: (IsStream t, Monad m, Storable a)

466− => Handle -> Int -> Int -> t m a -> m ()

467−writeSlice h i len s = undefined

468−

469−-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@

470−-- starting at index @i@ and writing up to @count@ elements in the reverse

471−-- direction ending at the index @i - count + 1@.

472−--

473−-- @since 0.7.0

474−{-# INLINE writeSliceRev #-}

475−writeSliceRev :: (IsStream t, Monad m, Storable a)

476− => Handle -> Int -> Int -> t m a -> m ()

477−writeSliceRev arr i len s = undefined

478−-}

src/Streamly/Internal/FileSystem/Handle.hs+406−10

@@ -11,23 +11,66 @@

12--

13-- License : BSD3

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

14+-- Maintainer : streamly@composewell.com

15-- Stability : experimental

16-- Portability : GHC

17−--

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

20module Streamly.Internal.FileSystem.Handle

18module Streamly.Internal.FileSystem.Handle

21 (

19 (

22− writeArray

20+ -- ** Read from Handle

21+ read

22+ -- , readUtf8

23+ -- , readLines

24+ -- , readFrames

25+ , readInChunksOf

26+

27+ -- -- * Array Read

28+ -- , readArrayUpto

29+ -- , readArrayOf

30+

31+ , readArraysOf

32+ -- , readArraysOf

33+ , readArrays

35+ -- ** Write to Handle

36+ -- Byte stream write (Folds)

37+ , write

38+ -- , writeUtf8

39+ -- , writeUtf8ByLines

40+ -- , writeByFrames

41+ , writeInChunksOf

42+

24− -- Byte stream write

43+ -- Byte stream write (Streams)

25 , writeS

44 , writeS

26 , writeSInChunksOf

45 , writeSInChunksOf

47+ -- -- * Array Write

48+ , writeArray

49+ , writeArrays

50+ , writeArraysInChunksOf

51+

28 -- -- * Array stream Write

52 -- -- * Array stream Write

29 , writeSArrays

53 , writeSArrays

30 , writeSArraysInChunksOf

54 , writeSArraysInChunksOf

55+

56+ -- -- * Random Access (Seek)

57+ -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or

58+ -- files that have random access or seek capability. This type of devices

59+ -- include disks, files, memory devices and exclude terminals, pipes,

60+ -- sockets and fifos.

61+ --

62+ -- , readIndex

63+ -- , readSlice

64+ -- , readSliceRev

65+ -- , readAt -- read from a given position to th end of file

66+ -- , readSliceArrayUpto

67+ -- , readSliceArrayOf

68+

69+ -- , writeIndex

70+ -- , writeSlice

71+ -- , writeSliceRev

72+ -- , writeAt -- start writing at the given position

73+ -- , writeSliceArray

31 )

74 )

32where

75where

@@ -35,18 +78,144 @@ import Control.Monad.IO.Class (MonadIO(..))

35import Data.Word (Word8)

78import Data.Word (Word8)

36import Foreign.ForeignPtr (withForeignPtr)

79import Foreign.ForeignPtr (withForeignPtr)

37import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

80import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)

38−import Foreign.Ptr (minusPtr)

81+import Foreign.Ptr (minusPtr, plusPtr)

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

82import Foreign.Storable (Storable(..))

83+import GHC.ForeignPtr (mallocPlainForeignPtrBytes)

40−import System.IO (Handle, hPutBuf)

84+import System.IO (Handle, hGetBufSome, hPutBuf)

41import Prelude hiding (read)

85import Prelude hiding (read)

87+import Streamly.Data.Fold (Fold)

43−import Streamly.Internal.Memory.Array.Types (Array(..))

88+import Streamly.Internal.Memory.Array.Types

89+ (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,

90+ lpackArraysChunksOf)

44import Streamly.Streams.Serial (SerialT)

91import Streamly.Streams.Serial (SerialT)

45−import Streamly.Internal.Memory.Array.Types (defaultChunkSize)

92+import Streamly.Streams.StreamK.Type (IsStream, mkStream)

93+-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)

47−import qualified Streamly.Memory.Array as A

95+import qualified Streamly.Data.Fold as FL

96+import qualified Streamly.Internal.Data.Fold.Types as FL

48import qualified Streamly.Internal.Memory.ArrayStream as AS

97import qualified Streamly.Internal.Memory.ArrayStream as AS

49import qualified Streamly.Internal.Prelude as S

98import qualified Streamly.Internal.Prelude as S

99+import qualified Streamly.Memory.Array as A

100+import qualified Streamly.Streams.StreamD.Type as D

101+

102+-------------------------------------------------------------------------------

103+-- References

104+-------------------------------------------------------------------------------

105+--

106+-- The following references may be useful to build an understanding about the

107+-- file API design:

108+--

109+-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.

110+-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations

111+-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.

112+-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.

113+-- https://www.w3.org/TR/FileAPI/ for http file API.

114+

115+-------------------------------------------------------------------------------

116+-- Array IO (Input)

117+-------------------------------------------------------------------------------

118+

119+-- | Read a 'ByteArray' from a file handle. If no data is available on the

120+-- handle it blocks until some data becomes available. If data is available

121+-- then it immediately returns that data without blocking. It reads a maximum

122+-- of up to the size requested.

123+{-# INLINABLE readArrayUpto #-}

124+readArrayUpto :: Int -> Handle -> IO (Array Word8)

125+readArrayUpto size h = do

126+ ptr <- mallocPlainForeignPtrBytes size

127+ -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))

128+ withForeignPtr ptr $ \p -> do

129+ n <- hGetBufSome h p size

130+ let v = Array

131+ { aStart = ptr

132+ , aEnd = p `plusPtr` n

133+ , aBound = p `plusPtr` size

134+ }

135+ -- XXX shrink only if the diff is significant

136+ shrinkToFit v

137+

138+-------------------------------------------------------------------------------

139+-- Stream of Arrays IO

140+-------------------------------------------------------------------------------

141+

142+-- | @readArraysOf size h@ reads a stream of arrays from file handle @h@.

143+-- The maximum size of a single array is specified by @size@. The actual size

144+-- read may be less than or equal to @size@.

145+{-# INLINABLE _readArraysOf #-}

146+_readArraysOf :: (IsStream t, MonadIO m)

147+ => Int -> Handle -> t m (Array Word8)

148+_readArraysOf size h = go

149+ where

150+ -- XXX use cons/nil instead

151+ go = mkStream $ \_ yld _ stp -> do

152+ arr <- liftIO $ readArrayUpto size h

153+ if A.length arr == 0

154+ then stp

155+ else yld arr go

156+

157+-- | @readArraysOf size handle@ reads a stream of arrays from the file

158+-- handle @handle@. The maximum size of a single array is limited to @size@.

159+-- The actual size read may be less than or equal to @size@.

160+--

161+-- @since 0.7.0

162+{-# INLINE_NORMAL readArraysOf #-}

163+readArraysOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)

164+readArraysOf size h = D.fromStreamD (D.Stream step ())

165+ where

166+ {-# INLINE_LATE step #-}

167+ step _ _ = do

168+ arr <- liftIO $ readArrayUpto size h

169+ return $

170+ case A.length arr of

171+ 0 -> D.Stop

172+ _ -> D.Yield arr ()

173+

174+-- XXX read 'Array a' instead of Word8

175+--

176+-- | @readArrays handle@ reads a stream of arrays from the specified file

177+-- handle. The maximum size of a single array is limited to

178+-- @defaultChunkSize@. The actual size read may be less than or equal to

179+-- @defaultChunkSize@.

180+--

181+-- > readArrays = readArraysOf defaultChunkSize

182+--

183+-- @since 0.7.0

184+{-# INLINE readArrays #-}

185+readArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)

186+readArrays = readArraysOf defaultChunkSize

187+

188+-------------------------------------------------------------------------------

189+-- Read File to Stream

190+-------------------------------------------------------------------------------

191+

192+-- TODO for concurrent streams implement readahead IO. We can send multiple

193+-- read requests at the same time. For serial case we can use async IO. We can

194+-- also control the read throughput in mbps or IOPS.

195+

196+-- | @readInChunksOf chunkSize handle@ reads a byte stream from a file

197+-- handle, reads are performed in chunks of up to @chunkSize@.

198+--

199+-- @since 0.7.0

200+{-# INLINE readInChunksOf #-}

201+readInChunksOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8

202+readInChunksOf chunkSize h = AS.concat $ readArraysOf chunkSize h

203+

204+-- TODO

205+-- Generate a stream of elements of the given type from a file 'Handle'.

206+-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a

207+--

208+-- > read = 'readInChunksOf' A.defaultChunkSize

209+-- | Generate a byte stream from a file 'Handle'.

210+--

211+-- @since 0.7.0

212+{-# INLINE read #-}

213+read :: (IsStream t, MonadIO m) => Handle -> t m Word8

214+read = AS.concat . readArrays

215+

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

217+-- Writing

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

219

51-------------------------------------------------------------------------------

220-------------------------------------------------------------------------------

52-- Array IO (output)

221-- Array IO (output)

@@ -113,3 +282,230 @@ writeSInChunksOf n h m = writeSArrays h $ S.arraysOf n m

113{-# INLINE writeS #-}

282{-# INLINE writeS #-}

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

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

115writeS = writeSInChunksOf defaultChunkSize

284writeS = writeSInChunksOf defaultChunkSize

285+

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

287+--

288+-- @since 0.7.0

289+{-# INLINE writeArrays #-}

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

291+writeArrays h = FL.drainBy (liftIO . writeArray h)

292+

293+-- | @writeArraysInChunksOf chunkSize handle@ writes a stream of arrays

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

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

296+-- it emitted as it is. Multiple arrays are coalesed as long as the total size

297+-- remains below the specified size.

298+--

299+-- @since 0.7.0

300+{-# INLINE writeArraysInChunksOf #-}

301+writeArraysInChunksOf :: (MonadIO m, Storable a)

302+ => Int -> Handle -> Fold m (Array a) ()

303+writeArraysInChunksOf n h = lpackArraysChunksOf n (writeArrays h)

304+

305+-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.

306+--

307+-- XXX test this

308+-- Note that if you use a chunk size less than 8K (GHC's default buffer

309+-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you

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

311+-- writes as well.

312+

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

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

315+-- collect the required input size.

316+--

317+-- @since 0.7.0

318+{-# INLINE writeInChunksOf #-}

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

320+writeInChunksOf n h = FL.lchunksOf n (writeNUnsafe n) (writeArrays h)

321+

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

323+--

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

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

326+--

327+-- @since 0.7.0

328+{-# INLINE write #-}

329+write :: MonadIO m => Handle -> Fold m Word8 ()

330+write = writeInChunksOf defaultChunkSize

331+

332+{-

333+{-# INLINE write #-}

334+write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()

335+write = toHandleWith A.defaultChunkSize

336+-}

337+

338+-- XXX mmap a file into an array. This could be useful for in-place operations

339+-- on a file. For example, we can quicksort the contents of a file by mmapping

340+-- it.

341+

342+-------------------------------------------------------------------------------

343+-- IO with encoding/decoding Unicode characters

344+-------------------------------------------------------------------------------

345+

346+{-

347+-- |

348+-- > readUtf8 = decodeUtf8 . read

349+--

350+-- Read a UTF8 encoded stream of unicode characters from a file handle.

351+--

352+-- @since 0.7.0

353+{-# INLINE readUtf8 #-}

354+readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char

355+readUtf8 = decodeUtf8 . read

356+

357+-- |

358+-- > writeUtf8 h s = write h $ encodeUtf8 s

359+--

360+-- Encode a stream of unicode characters to UTF8 and write it to the given file

361+-- handle. Default block buffering applies to the writes.

362+--

363+-- @since 0.7.0

364+{-# INLINE writeUtf8 #-}

365+writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()

366+writeUtf8 h s = write h $ encodeUtf8 s

367+

368+-- | Write a stream of unicode characters after encoding to UTF-8 in chunks

369+-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds

370+-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written

371+-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'

372+-- option.

373+--

374+-- @since 0.7.0

375+{-# INLINE writeUtf8ByLines #-}

376+writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()

377+writeUtf8ByLines = undefined

378+

379+-- | Read UTF-8 lines from a file handle and apply the specified fold to each

380+-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.

381+--

382+-- @since 0.7.0

383+{-# INLINE readLines #-}

384+readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b

385+readLines h f = foldLines (readUtf8 h) f

386+

387+-------------------------------------------------------------------------------

388+-- Framing on a sequence

389+-------------------------------------------------------------------------------

390+

391+-- | Read a stream from a file handle and split it into frames delimited by

392+-- the specified sequence of elements. The supplied fold is applied on each

393+-- frame.

394+--

395+-- @since 0.7.0

396+{-# INLINE readFrames #-}

397+readFrames :: (IsStream t, MonadIO m, Storable a)

398+ => Array a -> Handle -> Fold m a b -> t m b

399+readFrames = undefined -- foldFrames . read

400+

401+-- | Write a stream to the given file handle buffering up to frames separated

402+-- by the given sequence or up to a maximum of @defaultChunkSize@.

403+--

404+-- @since 0.7.0

405+{-# INLINE writeByFrames #-}

406+writeByFrames :: (IsStream t, MonadIO m, Storable a)

407+ => Array a -> Handle -> t m a -> m ()

408+writeByFrames = undefined

409+

410+-------------------------------------------------------------------------------

411+-- Framing by time

412+-------------------------------------------------------------------------------

413+

414+-- | Write collecting the input in sessions of n seconds or if chunkSize

415+-- gets exceeded.

416+{-# INLINE writeByChunksOrSessionsOf #-}

417+writeByChunksOrSessionsOf :: MonadIO m

418+ => Int -> Double -> Handle -> SerialT m Word8 -> m ()

419+writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined

420+

421+-- | Write collecting the input in sessions of n seconds or if defaultChunkSize

422+-- gets exceeded.

423+{-# INLINE writeBySessionsOf #-}

424+writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()

425+writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n

426+

427+-------------------------------------------------------------------------------

428+-- Random Access IO (Seek)

429+-------------------------------------------------------------------------------

430+

431+-- XXX handles could be shared, so we may not want to use the handle state at

432+-- all for these APIs. we can use pread and pwrite instead. On windows we will

433+-- need to use readFile/writeFile with an offset argument.

434+

435+-------------------------------------------------------------------------------

436+

437+-- | Read the element at the given index treating the file as an array.

438+--

439+-- @since 0.7.0

440+{-# INLINE readIndex #-}

441+readIndex :: Storable a => Handle -> Int -> Maybe a

442+readIndex arr i = undefined

443+

444+-- NOTE: To represent a range to read we have chosen (start, size) instead of

445+-- (start, end). This helps in removing the ambiguity of whether "end" is

446+-- included in the range or not.

447+--

448+-- We could avoid specifying the range to be read and instead use "take size"

449+-- on the stream, but it may end up reading more and then consume it partially.

450+

451+-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes

452+-- from @handle@ starting at the offset @pos@ from the beginning of the file.

453+--

454+-- Reads are performed in chunks of size @chunkSize@. For block devices, to

455+-- avoid reading partial blocks @chunkSize@ must align with the block size of

456+-- the underlying device. If the underlying block size is unknown, it is a good

457+-- idea to keep it a multiple 4KiB. This API ensures that the start of each

458+-- chunk is aligned with @chunkSize@ from second chunk onwards.

459+--

460+{-# INLINE readSliceWith #-}

461+readSliceWith :: (IsStream t, MonadIO m, Storable a)

462+ => Int -> Handle -> Int -> Int -> t m a

463+readSliceWith chunkSize h pos len = undefined

464+

465+-- | @readSlice h i count@ streams a slice from the file handle @h@ starting

466+-- at index @i@ and reading up to @count@ elements in the forward direction

467+-- ending at the index @i + count - 1@.

468+--

469+-- @since 0.7.0

470+{-# INLINE readSlice #-}

471+readSlice :: (IsStream t, MonadIO m, Storable a)

472+ => Handle -> Int -> Int -> t m a

473+readSlice = readSliceWith A.defaultChunkSize

474+

475+-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting

476+-- at index @i@ and reading up to @count@ elements in the reverse direction

477+-- ending at the index @i - count + 1@.

478+--

479+-- @since 0.7.0

480+{-# INLINE readSliceRev #-}

481+readSliceRev :: (IsStream t, MonadIO m, Storable a)

482+ => Handle -> Int -> Int -> t m a

483+readSliceRev h i count = undefined

484+

485+-- | Write the given element at the given index in the file.

486+--

487+-- @since 0.7.0

488+{-# INLINE writeIndex #-}

489+writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()

490+writeIndex h i a = undefined

491+

492+-- | @writeSlice h i count stream@ writes a stream to the file handle @h@

493+-- starting at index @i@ and writing up to @count@ elements in the forward

494+-- direction ending at the index @i + count - 1@.

495+--

496+-- @since 0.7.0

497+{-# INLINE writeSlice #-}

498+writeSlice :: (IsStream t, Monad m, Storable a)

499+ => Handle -> Int -> Int -> t m a -> m ()

500+writeSlice h i len s = undefined

501+

502+-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@

503+-- starting at index @i@ and writing up to @count@ elements in the reverse

504+-- direction ending at the index @i - count + 1@.

505+--

506+-- @since 0.7.0

507+{-# INLINE writeSliceRev #-}

508+writeSliceRev :: (IsStream t, Monad m, Storable a)

509+ => Handle -> Int -> Int -> t m a -> m ()

510+writeSliceRev arr i len s = undefined

511+-}