1{-# OPTIONS_HADDOCK hide #-}
3{-# LANGUAGE BangPatterns #-}
4{-# LANGUAGE MagicHash #-}
5{-# LANGUAGE RecordWildCards #-}
6{-# LANGUAGE UnboxedTuples #-}
11-- Module : Streamly.Internal.FileSystem.Handle
12-- Copyright : (c) 2018 Composewell Technologies
15-- Maintainer : streamly@composewell.com
16-- Stability : experimental
19module Streamly.Internal.FileSystem.Handle
21 -- ** Read from Handle
29 , toStreamWithBufferOf
35 , readChunksWithBufferOf
37 , toChunksWithBufferOf
41 -- Byte stream write (Folds)
48 -- Byte stream write (Streams)
50 , fromStreamWithBufferOf
55 , writeChunksWithBufferOf
57 -- -- * Array stream Write
59 , fromChunksWithBufferOf
61 -- -- * Random Access (Seek)
62 -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or
63 -- files that have random access or seek capability. This type of devices
64 -- include disks, files, memory devices and exclude terminals, pipes,
67 -- XXX need to decide whether to use readFromStepN style or readFromThenTo
68 -- style. The latter is consistent with list enumeration. The former may be
69 -- more unambiguous and it is easier and clearer to specify 0 elements.
71 -- We can also generate the request pattern using a funciton.
74 -- , readFrom -- read from a given position to the end of file
75 -- , readFromRev -- read from a given position the beginning of file
76 -- , readTo -- read from beginning up to the given position
77 -- , readToRev -- read from end to the given position in file
83 -- , readChunksFromToWithBufferOf
84 -- , readChunksFromThenToWithBufferOf
87 -- , writeFrom -- start writing at the given position
89 -- , writeTo -- write from beginning up to the given position
95 -- , writeChunksFromTo
96 -- , writeChunksFromToWithBufferOf
97 -- , writeChunksFromThenToWithBufferOf
101import Control.Monad.IO.Class (MonadIO(..))
102import Data.Word (Word8)
103import Foreign.ForeignPtr (withForeignPtr)
104import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
105import Foreign.Ptr (minusPtr, plusPtr)
106import Foreign.Storable (Storable(..))
107import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
108import System.IO (Handle, hGetBufSome, hPutBuf)
109import Prelude hiding (read)
111import Streamly.Data.Fold (Fold)
112import Streamly.Internal.Data.Unfold.Types (Unfold(..))
113import Streamly.Internal.Memory.Array.Types
114 (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,
116import Streamly.Streams.Serial (SerialT)
117import Streamly.Streams.StreamK.Type (IsStream, mkStream)
118-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
120import qualified Streamly.Data.Fold as FL
121import qualified Streamly.Internal.Data.Fold.Types as FL
122import qualified Streamly.Internal.Data.Unfold as UF
123import qualified Streamly.Internal.Memory.ArrayStream as AS
124import qualified Streamly.Internal.Prelude as S
125import qualified Streamly.Memory.Array as A
126import qualified Streamly.Streams.StreamD.Type as D
128-------------------------------------------------------------------------------
130-------------------------------------------------------------------------------
132-- The following references may be useful to build an understanding about the
135-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
136-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
137-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
138-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
139-- https://www.w3.org/TR/FileAPI/ for http file API.
141-------------------------------------------------------------------------------
143-------------------------------------------------------------------------------
145-- | Read a 'ByteArray' from a file handle. If no data is available on the
146-- handle it blocks until some data becomes available. If data is available
147-- then it immediately returns that data without blocking. It reads a maximum
148-- of up to the size requested.
149{-# INLINABLE readArrayUpto #-}
150readArrayUpto :: Int -> Handle -> IO (Array Word8)
151readArrayUpto size h = do
152 ptr <- mallocPlainForeignPtrBytes size
153 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
154 withForeignPtr ptr $ \p -> do
155 n <- hGetBufSome h p size
158 , aEnd = p `plusPtr` n
159 , aBound = p `plusPtr` size
161 -- XXX shrink only if the diff is significant
164-------------------------------------------------------------------------------
165-- Stream of Arrays IO
166-------------------------------------------------------------------------------
168-- | @toChunksWithBufferOf size h@ reads a stream of arrays from file handle @h@.
169-- The maximum size of a single array is specified by @size@. The actual size
170-- read may be less than or equal to @size@.
171{-# INLINABLE _toChunksWithBufferOf #-}
172_toChunksWithBufferOf :: (IsStream t, MonadIO m)
173 => Int -> Handle -> t m (Array Word8)
174_toChunksWithBufferOf size h = go
176 -- XXX use cons/nil instead
177 go = mkStream $ \_ yld _ stp -> do
178 arr <- liftIO $ readArrayUpto size h
183-- | @toChunksWithBufferOf size handle@ reads a stream of arrays from the file
184-- handle @handle@. The maximum size of a single array is limited to @size@.
185-- The actual size read may be less than or equal to @size@.
188{-# INLINE_NORMAL toChunksWithBufferOf #-}
189toChunksWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
190toChunksWithBufferOf size h = D.fromStreamD (D.Stream step ())
192 {-# INLINE_LATE step #-}
194 arr <- liftIO $ readArrayUpto size h
200-- | Unfold the tuple @(bufsize, handle)@ into a stream of 'Word8' arrays.
201-- Read requests to the IO device are performed using a buffer of size
202-- @bufsize@. The size of an array in the resulting stream is always less than
203-- or equal to @bufsize@.
206{-# INLINE_NORMAL readChunksWithBufferOf #-}
207readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Handle) (Array Word8)
208readChunksWithBufferOf = Unfold step return
210 {-# INLINE_LATE step #-}
212 arr <- liftIO $ readArrayUpto size h
216 _ -> D.Yield arr (size, h)
218-- XXX read 'Array a' instead of Word8
220-- | @toChunks handle@ reads a stream of arrays from the specified file
221-- handle. The maximum size of a single array is limited to
222-- @defaultChunkSize@. The actual size read may be less than or equal to
223-- @defaultChunkSize@.
225-- > toChunks = toChunksWithBufferOf defaultChunkSize
228{-# INLINE toChunks #-}
229toChunks :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
230toChunks = toChunksWithBufferOf defaultChunkSize
232-- | Unfolds a handle into a stream of 'Word8' arrays. Requests to the IO
233-- device are performed using a buffer of size
234-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'. The
235-- size of arrays in the resulting stream are therefore less than or equal to
236-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
239{-# INLINE readChunks #-}
240readChunks :: MonadIO m => Unfold m Handle (Array Word8)
241readChunks = UF.supplyFirst readChunksWithBufferOf defaultChunkSize
243-------------------------------------------------------------------------------
244-- Read File to Stream
245-------------------------------------------------------------------------------
247-- TODO for concurrent streams implement readahead IO. We can send multiple
248-- read requests at the same time. For serial case we can use async IO. We can
249-- also control the read throughput in mbps or IOPS.
251-- | Unfolds the tuple @(bufsize, handle)@ into a byte stream, read requests
252-- to the IO device are performed using buffers of @bufsize@.
255{-# INLINE readWithBufferOf #-}
256readWithBufferOf :: MonadIO m => Unfold m (Int, Handle) Word8
257readWithBufferOf = UF.concat readChunksWithBufferOf A.read
259-- | @toStreamWithBufferOf bufsize handle@ reads a byte stream from a file
260-- handle, reads are performed in chunks of up to @bufsize@.
263{-# INLINE toStreamWithBufferOf #-}
264toStreamWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
265toStreamWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h
268-- Generate a stream of elements of the given type from a file 'Handle'.
269-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
271-- | Unfolds a file handle into a byte stream. IO requests to the device are
272-- performed in sizes of
273-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
277read :: MonadIO m => Unfold m Handle Word8
278read = UF.supplyFirst readWithBufferOf defaultChunkSize
280-- | Generate a byte stream from a file 'Handle'.
283{-# INLINE toStream #-}
284toStream :: (IsStream t, MonadIO m) => Handle -> t m Word8
285toStream = AS.concat . toChunks
287-------------------------------------------------------------------------------
289-------------------------------------------------------------------------------
291-------------------------------------------------------------------------------
293-------------------------------------------------------------------------------
295-- | Write an 'Array' to a file handle.
298{-# INLINABLE writeArray #-}
299writeArray :: Storable a => Handle -> Array a -> IO ()
300writeArray _ arr | A.length arr == 0 = return ()
301writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen
304 let p = unsafeForeignPtrToPtr aStart
307-------------------------------------------------------------------------------
308-- Stream of Arrays IO
309-------------------------------------------------------------------------------
311-------------------------------------------------------------------------------
313-------------------------------------------------------------------------------
315-- | Write a stream of arrays to a handle.
318{-# INLINE fromChunks #-}
319fromChunks :: (MonadIO m, Storable a)
320 => Handle -> SerialT m (Array a) -> m ()
321fromChunks h m = S.mapM_ (liftIO . writeArray h) m
323-- | @fromChunksWithBufferOf bufsize handle stream@ writes a stream of arrays
324-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
325-- The chunk size is only a maximum and the actual writes could be smaller as
326-- we do not split the arrays to fit exactly to the specified size.
329{-# INLINE fromChunksWithBufferOf #-}
330fromChunksWithBufferOf :: (MonadIO m, Storable a)
331 => Int -> Handle -> SerialT m (Array a) -> m ()
332fromChunksWithBufferOf n h xs = fromChunks h $ AS.compact n xs
334-- | @fromStreamWithBufferOf bufsize handle stream@ writes @stream@ to @handle@
335-- in chunks of @bufsize@. A write is performed to the IO device as soon as we
336-- collect the required input size.
339{-# INLINE fromStreamWithBufferOf #-}
340fromStreamWithBufferOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
341fromStreamWithBufferOf n h m = fromChunks h $ S.arraysOf n m
342-- fromStreamWithBufferOf n h m = fromChunks h $ AS.arraysOf n m
344-- > write = 'writeWithBufferOf' A.defaultChunkSize
346-- | Write a byte stream to a file handle. Accumulates the input in chunks of
347-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing.
349-- NOTE: This may perform better than the 'write' fold, you can try this if you
350-- need some extra perf boost.
353{-# INLINE fromStream #-}
354fromStream :: MonadIO m => Handle -> SerialT m Word8 -> m ()
355fromStream = fromStreamWithBufferOf defaultChunkSize
357-- | Write a stream of arrays to a handle. Each array in the stream is written
358-- to the device as a separate IO request.
361{-# INLINE writeChunks #-}
362writeChunks :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
363writeChunks h = FL.drainBy (liftIO . writeArray h)
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.lchunksOf n (writeNUnsafe n) (writeChunks h)
394-- > write = 'writeWithBufferOf' A.defaultChunkSize
396-- | Write a byte stream to a file handle. Accumulates the input in chunks of
397-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing
402write :: MonadIO m => Handle -> Fold m Word8 ()
403write = writeWithBufferOf defaultChunkSize
407write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
408write = toHandleWith A.defaultChunkSize
411-- XXX mmap a file into an array. This could be useful for in-place operations
412-- on a file. For example, we can quicksort the contents of a file by mmapping
415-------------------------------------------------------------------------------
416-- IO with encoding/decoding Unicode characters
417-------------------------------------------------------------------------------
421-- > readUtf8 = decodeUtf8 . read
423-- Read a UTF8 encoded stream of unicode characters from a file handle.
426{-# INLINE readUtf8 #-}
427readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
428readUtf8 = decodeUtf8 . read
431-- > writeUtf8 h s = write h $ encodeUtf8 s
433-- Encode a stream of unicode characters to UTF8 and write it to the given file
434-- handle. Default block buffering applies to the writes.
437{-# INLINE writeUtf8 #-}
438writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
439writeUtf8 h s = write h $ encodeUtf8 s
441-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
442-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
443-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
444-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
448{-# INLINE writeUtf8ByLines #-}
449writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
450writeUtf8ByLines = undefined
452-- | Read UTF-8 lines from a file handle and apply the specified fold to each
453-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
456{-# INLINE readLines #-}
457readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
458readLines h f = foldLines (readUtf8 h) f
460-------------------------------------------------------------------------------
461-- Framing on a sequence
462-------------------------------------------------------------------------------
464-- | Read a stream from a file handle and split it into frames delimited by
465-- the specified sequence of elements. The supplied fold is applied on each
469{-# INLINE readFrames #-}
470readFrames :: (IsStream t, MonadIO m, Storable a)
471 => Array a -> Handle -> Fold m a b -> t m b
472readFrames = undefined -- foldFrames . read
474-- | Write a stream to the given file handle buffering up to frames separated
475-- by the given sequence or up to a maximum of @defaultChunkSize@.
478{-# INLINE writeByFrames #-}
479writeByFrames :: (IsStream t, MonadIO m, Storable a)
480 => Array a -> Handle -> t m a -> m ()
481writeByFrames = undefined
483-------------------------------------------------------------------------------
485-------------------------------------------------------------------------------
487-- | Write collecting the input in sessions of n seconds or if chunkSize
489{-# INLINE writeByChunksOrSessionsOf #-}
490writeByChunksOrSessionsOf :: MonadIO m
491 => Int -> Double -> Handle -> SerialT m Word8 -> m ()
492writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined
494-- | Write collecting the input in sessions of n seconds or if defaultChunkSize
496{-# INLINE writeBySessionsOf #-}
497writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()
498writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n
500-------------------------------------------------------------------------------
501-- Random Access IO (Seek)
502-------------------------------------------------------------------------------
504-- XXX handles could be shared, so we may not want to use the handle state at
505-- all for these APIs. we can use pread and pwrite instead. On windows we will
506-- need to use readFile/writeFile with an offset argument.
508-------------------------------------------------------------------------------
510-- | Read the element at the given index treating the file as an array.
513{-# INLINE readIndex #-}
514readIndex :: Storable a => Handle -> Int -> Maybe a
515readIndex arr i = undefined
517-- NOTE: To represent a range to read we have chosen (start, size) instead of
518-- (start, end). This helps in removing the ambiguity of whether "end" is
519-- included in the range or not.
521-- We could avoid specifying the range to be read and instead use "take size"
522-- on the stream, but it may end up reading more and then consume it partially.
524-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
525-- from @handle@ starting at the offset @pos@ from the beginning of the file.
527-- Reads are performed in chunks of size @chunkSize@. For block devices, to
528-- avoid reading partial blocks @chunkSize@ must align with the block size of
529-- the underlying device. If the underlying block size is unknown, it is a good
530-- idea to keep it a multiple 4KiB. This API ensures that the start of each
531-- chunk is aligned with @chunkSize@ from second chunk onwards.
533{-# INLINE readSliceWith #-}
534readSliceWith :: (IsStream t, MonadIO m, Storable a)
535 => Int -> Handle -> Int -> Int -> t m a
536readSliceWith chunkSize h pos len = undefined
538-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
539-- at index @i@ and reading up to @count@ elements in the forward direction
540-- ending at the index @i + count - 1@.
543{-# INLINE readSlice #-}
544readSlice :: (IsStream t, MonadIO m, Storable a)
545 => Handle -> Int -> Int -> t m a
546readSlice = readSliceWith A.defaultChunkSize
548-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
549-- at index @i@ and reading up to @count@ elements in the reverse direction
550-- ending at the index @i - count + 1@.
553{-# INLINE readSliceRev #-}
554readSliceRev :: (IsStream t, MonadIO m, Storable a)
555 => Handle -> Int -> Int -> t m a
556readSliceRev h i count = undefined
558-- | Write the given element at the given index in the file.
561{-# INLINE writeIndex #-}
562writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
563writeIndex h i a = undefined
565-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
566-- starting at index @i@ and writing up to @count@ elements in the forward
567-- direction ending at the index @i + count - 1@.
570{-# INLINE writeSlice #-}
571writeSlice :: (IsStream t, Monad m, Storable a)
572 => Handle -> Int -> Int -> t m a -> m ()
573writeSlice h i len s = undefined
575-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
576-- starting at index @i@ and writing up to @count@ elements in the reverse
577-- direction ending at the index @i - count + 1@.
580{-# INLINE writeSliceRev #-}
581writeSliceRev :: (IsStream t, Monad m, Storable a)
582 => Handle -> Int -> Int -> t m a -> m ()
583writeSliceRev 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