4-- Module : Streamly.Internal.FileSystem.Handle
5-- Copyright : (c) 2018 Composewell Technologies
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
12module Streamly.Internal.FileSystem.Handle
14 -- ** Read from Handle
28 , readChunksWithBufferOf
30 , toChunksWithBufferOf
34 -- Byte stream write (Folds)
43 -- Byte stream write (Streams)
45 , putBytesWithBufferOf
50 , writeChunksWithBufferOf
52 -- -- * Array stream Write
53 , putChunksWithBufferOf
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,
62 -- XXX need to decide whether to use readFromStepN style or readFromThenTo
63 -- style. The latter is consistent with list enumeration. The former may be
64 -- more unambiguous and it is easier and clearer to specify 0 elements.
66 -- We can also generate the request pattern using a funciton.
69 -- , readFrom -- read from a given position to the end of file
70 -- , readFromRev -- read from a given position the beginning of file
71 -- , readTo -- read from beginning up to the given position
72 -- , readToRev -- read from end to the given position in file
78 -- , readChunksFromToWithBufferOf
79 -- , readChunksFromThenToWithBufferOf
82 -- , writeFrom -- start writing at the given position
84 -- , writeTo -- write from beginning up to the given position
90 -- , writeChunksFromTo
91 -- , writeChunksFromToWithBufferOf
92 -- , writeChunksFromThenToWithBufferOf
96import Control.Monad.IO.Class (MonadIO(..))
97import Data.Word (Word8)
98import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
99import Foreign.ForeignPtr (withForeignPtr)
100import Foreign.Ptr (minusPtr, plusPtr)
101import Foreign.Storable (Storable(..))
102import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
103import System.IO (Handle, hGetBufSome, hPutBuf)
104import Prelude hiding (read)
106import Streamly.Internal.BaseCompat
107import Streamly.Data.Fold (Fold)
108import Streamly.Internal.Data.Fold.Type (Fold2(..))
109import Streamly.Internal.Data.Unfold.Type (Unfold(..))
110import Streamly.Internal.Data.Array.Foreign.Type
111 (Array(..), writeNUnsafe, defaultChunkSize , unsafeFreezeWithShrink)
112import Streamly.Internal.Data.Array.Foreign.Mut.Type (mutableArray)
113import Streamly.Internal.Data.Stream.Serial (SerialT)
114import Streamly.Internal.Data.Stream.StreamK.Type (IsStream, mkStream)
115import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)
116-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
118import qualified Streamly.Data.Fold as FL
119import qualified Streamly.Internal.Data.Fold.Type as FL
120import qualified Streamly.Internal.Data.Unfold as UF
121import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS
122import qualified Streamly.Internal.Data.Stream.IsStream as S
123import qualified Streamly.Data.Array.Foreign as A
124import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
126-------------------------------------------------------------------------------
128-------------------------------------------------------------------------------
130-- The following references may be useful to build an understanding about the
133-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
134-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
135-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
136-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
137-- https://www.w3.org/TR/FileAPI/ for http file API.
139-------------------------------------------------------------------------------
141-------------------------------------------------------------------------------
143-- | Read a 'ByteArray' from a file handle. If no data is available on the
144-- handle it blocks until some data becomes available. If data is available
145-- then it immediately returns that data without blocking. It reads a maximum
146-- of up to the size requested.
147{-# INLINABLE readArrayUpto #-}
148readArrayUpto :: Int -> Handle -> IO (Array Word8)
149readArrayUpto size h = do
150 ptr <- mallocPlainForeignPtrBytes size
151 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
152 withForeignPtr ptr $ \p -> do
153 n <- hGetBufSome h p size
154 -- XXX shrink only if the diff is significant
156 unsafeFreezeWithShrink $
157 mutableArray ptr (p `plusPtr` n) (p `plusPtr` size)
159-------------------------------------------------------------------------------
160-- Stream of Arrays IO
161-------------------------------------------------------------------------------
163-- | @toChunksWithBufferOf size h@ reads a stream of arrays from file handle @h@.
164-- The maximum size of a single array is specified by @size@. The actual size
165-- read may be less than or equal to @size@.
166{-# INLINABLE _toChunksWithBufferOf #-}
167_toChunksWithBufferOf :: (IsStream t, MonadIO m)
168 => Int -> Handle -> t m (Array Word8)
169_toChunksWithBufferOf size h = go
171 -- XXX use cons/nil instead
172 go = mkStream $ \_ yld _ stp -> do
173 arr <- liftIO $ readArrayUpto size h
178-- | @toChunksWithBufferOf size handle@ reads a stream of arrays from the file
179-- handle @handle@. The maximum size of a single array is limited to @size@.
180-- The actual size read may be less than or equal to @size@.
183{-# INLINE_NORMAL toChunksWithBufferOf #-}
184toChunksWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
185toChunksWithBufferOf size h = D.fromStreamD (D.Stream step ())
187 {-# INLINE_LATE step #-}
189 arr <- liftIO $ readArrayUpto size h
195-- | Unfold the tuple @(bufsize, handle)@ into a stream of 'Word8' arrays.
196-- Read requests to the IO device are performed using a buffer of size
197-- @bufsize@. The size of an array in the resulting stream is always less than
198-- or equal to @bufsize@.
201{-# INLINE_NORMAL readChunksWithBufferOf #-}
202readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Handle) (Array Word8)
203readChunksWithBufferOf = Unfold step return
205 {-# INLINE_LATE step #-}
207 arr <- liftIO $ readArrayUpto size h
211 _ -> D.Yield arr (size, h)
213-- XXX read 'Array a' instead of Word8
215-- | @toChunks handle@ reads a stream of arrays from the specified file
216-- handle. The maximum size of a single array is limited to
217-- @defaultChunkSize@. The actual size read may be less than or equal to
218-- @defaultChunkSize@.
220-- > toChunks = toChunksWithBufferOf defaultChunkSize
223{-# INLINE toChunks #-}
224toChunks :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
225toChunks = toChunksWithBufferOf defaultChunkSize
227-- | Unfolds a handle into a stream of 'Word8' arrays. Requests to the IO
228-- device are performed using a buffer of size
229-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'. The
230-- size of arrays in the resulting stream are therefore less than or equal to
231-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'.
234{-# INLINE readChunks #-}
235readChunks :: MonadIO m => Unfold m Handle (Array Word8)
236readChunks = UF.supplyFirst defaultChunkSize readChunksWithBufferOf
238-------------------------------------------------------------------------------
239-- Read File to Stream
240-------------------------------------------------------------------------------
242-- TODO for concurrent streams implement readahead IO. We can send multiple
243-- read requests at the same time. For serial case we can use async IO. We can
244-- also control the read throughput in mbps or IOPS.
246-- | Unfolds the tuple @(bufsize, handle)@ into a byte stream, read requests
247-- to the IO device are performed using buffers of @bufsize@.
250{-# INLINE readWithBufferOf #-}
251readWithBufferOf :: MonadIO m => Unfold m (Int, Handle) Word8
252readWithBufferOf = UF.many readChunksWithBufferOf A.read
254-- | @toBytesWithBufferOf bufsize handle@ reads a byte stream from a file
255-- handle, reads are performed in chunks of up to @bufsize@.
258{-# INLINE toBytesWithBufferOf #-}
259toBytesWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
260toBytesWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h
263-- Generate a stream of elements of the given type from a file 'Handle'.
264-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
266-- | Unfolds a file handle into a byte stream. IO requests to the device are
267-- performed in sizes of
268-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'.
272read :: MonadIO m => Unfold m Handle Word8
273read = UF.supplyFirst defaultChunkSize readWithBufferOf
275-- | Generate a byte stream from a file 'Handle'.
278{-# INLINE toBytes #-}
279toBytes :: (IsStream t, MonadIO m) => Handle -> t m Word8
280toBytes = AS.concat . toChunks
282-------------------------------------------------------------------------------
284-------------------------------------------------------------------------------
286-------------------------------------------------------------------------------
288-------------------------------------------------------------------------------
290-- | Write an 'Array' to a file handle.
293{-# INLINABLE writeArray #-}
294writeArray :: Storable a => Handle -> Array a -> IO ()
295writeArray _ arr | A.length arr == 0 = return ()
296writeArray h Array{..} = unsafeWithForeignPtr aStart $ \p -> hPutBuf h p aLen
299 let p = unsafeForeignPtrToPtr aStart
302-------------------------------------------------------------------------------
303-- Stream of Arrays IO
304-------------------------------------------------------------------------------
306-------------------------------------------------------------------------------
308-------------------------------------------------------------------------------
310-- XXX use an unfold to fromObjects or fromUnfold so that we can put any object
311-- | Write a stream of arrays to a handle.
314{-# INLINE putChunks #-}
315putChunks :: (MonadIO m, Storable a)
316 => Handle -> SerialT m (Array a) -> m ()
317putChunks h = S.mapM_ (liftIO . writeArray h)
319-- | @putChunksWithBufferOf bufsize handle stream@ writes a stream of arrays
320-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
321-- The chunk size is only a maximum and the actual writes could be smaller as
322-- we do not split the arrays to fit exactly to the specified size.
325{-# INLINE putChunksWithBufferOf #-}
326putChunksWithBufferOf :: (MonadIO m, Storable a)
327 => Int -> Handle -> SerialT m (Array a) -> m ()
328putChunksWithBufferOf n h xs = putChunks h $ AS.compact n xs
330-- | @putBytesWithBufferOf bufsize handle stream@ writes @stream@ to @handle@
331-- in chunks of @bufsize@. A write is performed to the IO device as soon as we
332-- collect the required input size.
335{-# INLINE putBytesWithBufferOf #-}
336putBytesWithBufferOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
337putBytesWithBufferOf n h m = putChunks h $ S.arraysOf n m
338-- putBytesWithBufferOf n h m = putChunks h $ AS.arraysOf n m
340-- > write = 'writeWithBufferOf' A.defaultChunkSize
342-- | Write a byte stream to a file handle. Accumulates the input in chunks of
343-- up to 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize' before writing.
345-- NOTE: This may perform better than the 'write' fold, you can try this if you
346-- need some extra perf boost.
349{-# INLINE putBytes #-}
350putBytes :: MonadIO m => Handle -> SerialT m Word8 -> m ()
351putBytes = putBytesWithBufferOf defaultChunkSize
353-- | Write a stream of arrays to a handle. Each array in the stream is written
354-- to the device as a separate IO request.
357{-# INLINE writeChunks #-}
358writeChunks :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
359writeChunks h = FL.drainBy (liftIO . writeArray h)
361{-# INLINE writeChunks2 #-}
362writeChunks2 :: (MonadIO m, Storable a) => Fold2 m Handle (Array a) ()
363writeChunks2 = Fold2 (\h arr -> liftIO $ writeArray h arr >> return h) return (\_ -> return ())
365-- | @writeChunksWithBufferOf bufsize handle@ writes a stream of arrays
366-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
367-- We never split an array, if a single array is bigger than the specified size
368-- it emitted as it is. Multiple arrays are coalesed as long as the total size
369-- remains below the specified size.
372{-# INLINE writeChunksWithBufferOf #-}
373writeChunksWithBufferOf :: (MonadIO m, Storable a)
374 => Int -> Handle -> Fold m (Array a) ()
375writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)
377-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
380-- Note that if you use a chunk size less than 8K (GHC's default buffer
381-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
382-- do not want buffering to occur at GHC level as well. Same thing applies to
385-- | @writeWithBufferOf reqSize handle@ writes the input stream to @handle@.
386-- Bytes in the input stream are collected into a buffer until we have a chunk
387-- of @reqSize@ and then written to the IO device.
390{-# INLINE writeWithBufferOf #-}
391writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
392writeWithBufferOf n h = FL.chunksOf n (writeNUnsafe n) (writeChunks h)
394{-# INLINE writeWithBufferOf2 #-}
395writeWithBufferOf2 :: MonadIO m => Int -> Fold2 m Handle Word8 ()
396writeWithBufferOf2 n = FL.chunksOf2 n (writeNUnsafe n) writeChunks2
398-- > write = 'writeWithBufferOf' A.defaultChunkSize
400-- | Write a byte stream to a file handle. Accumulates the input in chunks of
401-- up to 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize' before writing
406write :: MonadIO m => Handle -> Fold m Word8 ()
407write = writeWithBufferOf defaultChunkSize
410write2 :: MonadIO m => Fold2 m Handle Word8 ()
411write2 = writeWithBufferOf2 defaultChunkSize
415write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
416write = toHandleWith A.defaultChunkSize
419-- XXX mmap a file into an array. This could be useful for in-place operations
420-- on a file. For example, we can quicksort the contents of a file by mmapping
423-------------------------------------------------------------------------------
424-- IO with encoding/decoding Unicode characters
425-------------------------------------------------------------------------------
429-- > readUtf8 = decodeUtf8 . read
431-- Read a UTF8 encoded stream of unicode characters from a file handle.
434{-# INLINE readUtf8 #-}
435readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
436readUtf8 = decodeUtf8 . read
439-- > writeUtf8 h s = write h $ encodeUtf8 s
441-- Encode a stream of unicode characters to UTF8 and write it to the given file
442-- handle. Default block buffering applies to the writes.
445{-# INLINE writeUtf8 #-}
446writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
447writeUtf8 h s = write h $ encodeUtf8 s
449-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
450-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
451-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
452-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
456{-# INLINE writeUtf8ByLines #-}
457writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
458writeUtf8ByLines = undefined
460-- | Read UTF-8 lines from a file handle and apply the specified fold to each
461-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
464{-# INLINE readLines #-}
465readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
466readLines h f = foldLines (readUtf8 h) f
468-------------------------------------------------------------------------------
469-- Framing on a sequence
470-------------------------------------------------------------------------------
472-- | Read a stream from a file handle and split it into frames delimited by
473-- the specified sequence of elements. The supplied fold is applied on each
477{-# INLINE readFrames #-}
478readFrames :: (IsStream t, MonadIO m, Storable a)
479 => Array a -> Handle -> Fold m a b -> t m b
480readFrames = undefined -- foldFrames . read
482-- | Write a stream to the given file handle buffering up to frames separated
483-- by the given sequence or up to a maximum of @defaultChunkSize@.
486{-# INLINE writeByFrames #-}
487writeByFrames :: (IsStream t, MonadIO m, Storable a)
488 => Array a -> Handle -> t m a -> m ()
489writeByFrames = undefined
491-------------------------------------------------------------------------------
493-------------------------------------------------------------------------------
495-- | Write collecting the input in sessions of n seconds or if chunkSize
497{-# INLINE writeByChunksOrSessionsOf #-}
498writeByChunksOrSessionsOf :: MonadIO m
499 => Int -> Double -> Handle -> SerialT m Word8 -> m ()
500writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined
502-- | Write collecting the input in sessions of n seconds or if defaultChunkSize
504{-# INLINE writeBySessionsOf #-}
505writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()
506writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n
508-------------------------------------------------------------------------------
509-- Random Access IO (Seek)
510-------------------------------------------------------------------------------
512-- XXX handles could be shared, so we may not want to use the handle state at
513-- all for these APIs. we can use pread and pwrite instead. On windows we will
514-- need to use readFile/writeFile with an offset argument.
516-------------------------------------------------------------------------------
518-- | Read the element at the given index treating the file as an array.
521{-# INLINE readIndex #-}
522readIndex :: Storable a => Handle -> Int -> Maybe a
523readIndex arr i = undefined
525-- NOTE: To represent a range to read we have chosen (start, size) instead of
526-- (start, end). This helps in removing the ambiguity of whether "end" is
527-- included in the range or not.
529-- We could avoid specifying the range to be read and instead use "take size"
530-- on the stream, but it may end up reading more and then consume it partially.
532-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
533-- from @handle@ starting at the offset @pos@ from the beginning of the file.
535-- Reads are performed in chunks of size @chunkSize@. For block devices, to
536-- avoid reading partial blocks @chunkSize@ must align with the block size of
537-- the underlying device. If the underlying block size is unknown, it is a good
538-- idea to keep it a multiple 4KiB. This API ensures that the start of each
539-- chunk is aligned with @chunkSize@ from second chunk onwards.
541{-# INLINE readSliceWith #-}
542readSliceWith :: (IsStream t, MonadIO m, Storable a)
543 => Int -> Handle -> Int -> Int -> t m a
544readSliceWith chunkSize h pos len = undefined
546-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
547-- at index @i@ and reading up to @count@ elements in the forward direction
548-- ending at the index @i + count - 1@.
551{-# INLINE readSlice #-}
552readSlice :: (IsStream t, MonadIO m, Storable a)
553 => Handle -> Int -> Int -> t m a
554readSlice = readSliceWith A.defaultChunkSize
556-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
557-- at index @i@ and reading up to @count@ elements in the reverse direction
558-- ending at the index @i - count + 1@.
561{-# INLINE readSliceRev #-}
562readSliceRev :: (IsStream t, MonadIO m, Storable a)
563 => Handle -> Int -> Int -> t m a
564readSliceRev h i count = undefined
566-- | Write the given element at the given index in the file.
569{-# INLINE writeIndex #-}
570writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
571writeIndex h i a = undefined
573-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
574-- starting at index @i@ and writing up to @count@ elements in the forward
575-- direction ending at the index @i + count - 1@.
578{-# INLINE writeSlice #-}
579writeSlice :: (IsStream t, Monad m, Storable a)
580 => Handle -> Int -> Int -> t m a -> m ()
581writeSlice h i len s = undefined
583-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
584-- starting at index @i@ and writing up to @count@ elements in the reverse
585-- direction ending at the index @i - count + 1@.
588{-# INLINE writeSliceRev #-}
589writeSliceRev :: (IsStream t, Monad m, Storable a)
590 => Handle -> Int -> Int -> t m a -> m ()
591writeSliceRev arr i len s = undefined
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list