2{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE MagicHash #-}
4{-# LANGUAGE RecordWildCards #-}
5{-# LANGUAGE UnboxedTuples #-}
10-- Module : Streamly.Internal.FileSystem.Handle
11-- Copyright : (c) 2018 Composewell Technologies
14-- Maintainer : streamly@composewell.com
15-- Stability : experimental
18module Streamly.Internal.FileSystem.Handle
20 -- ** Read from Handle
36 -- Byte stream write (Folds)
43 -- Byte stream write (Streams)
45 , fromStreamInChunksOf
50 , writeArraysInChunksOf
52 -- -- * Array stream Write
54 , fromStreamArraysInChunksOf
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,
65 -- , readAt -- read from a given position to th end of file
66 -- , readSliceArrayUpto
72 -- , writeAt -- start writing at the given position
77import Control.Monad.IO.Class (MonadIO(..))
78import Data.Word (Word8)
79import Foreign.ForeignPtr (withForeignPtr)
80import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
81import Foreign.Ptr (minusPtr, plusPtr)
82import Foreign.Storable (Storable(..))
83import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
84import System.IO (Handle, hGetBufSome, hPutBuf)
85import Prelude hiding (read)
87import Streamly.Data.Fold (Fold)
88import Streamly.Internal.Memory.Array.Types
89 (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,
91import Streamly.Streams.Serial (SerialT)
92import Streamly.Streams.StreamK.Type (IsStream, mkStream)
93-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
95import qualified Streamly.Data.Fold as FL
96import qualified Streamly.Internal.Data.Fold.Types as FL
97import qualified Streamly.Internal.Memory.ArrayStream as AS
98import qualified Streamly.Internal.Prelude as S
99import qualified Streamly.Memory.Array as A
100import qualified Streamly.Streams.StreamD.Type as D
102-------------------------------------------------------------------------------
104-------------------------------------------------------------------------------
106-- The following references may be useful to build an understanding about the
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.
115-------------------------------------------------------------------------------
117-------------------------------------------------------------------------------
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 #-}
124readArrayUpto :: Int -> Handle -> IO (Array Word8)
125readArrayUpto 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
132 , aEnd = p `plusPtr` n
133 , aBound = p `plusPtr` size
135 -- XXX shrink only if the diff is significant
138-------------------------------------------------------------------------------
139-- Stream of Arrays IO
140-------------------------------------------------------------------------------
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
150 -- XXX use cons/nil instead
151 go = mkStream $ \_ yld _ stp -> do
152 arr <- liftIO $ readArrayUpto size h
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@.
162{-# INLINE_NORMAL readArraysOf #-}
163readArraysOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
164readArraysOf size h = D.fromStreamD (D.Stream step ())
166 {-# INLINE_LATE step #-}
168 arr <- liftIO $ readArrayUpto size h
174-- XXX read 'Array a' instead of Word8
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@.
181-- > readArrays = readArraysOf defaultChunkSize
184{-# INLINE readArrays #-}
185readArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
186readArrays = readArraysOf defaultChunkSize
188-------------------------------------------------------------------------------
189-- Read File to Stream
190-------------------------------------------------------------------------------
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.
196-- | @readInChunksOf chunkSize handle@ reads a byte stream from a file
197-- handle, reads are performed in chunks of up to @chunkSize@.
200{-# INLINE readInChunksOf #-}
201readInChunksOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
202readInChunksOf chunkSize h = AS.concat $ readArraysOf chunkSize h
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
208-- > read = 'readInChunksOf' A.defaultChunkSize
209-- | Generate a byte stream from a file 'Handle'.
213read :: (IsStream t, MonadIO m) => Handle -> t m Word8
214read = AS.concat . readArrays
216-------------------------------------------------------------------------------
218-------------------------------------------------------------------------------
220-------------------------------------------------------------------------------
222-------------------------------------------------------------------------------
224-- | Write an 'Array' to a file handle.
227{-# INLINABLE writeArray #-}
228writeArray :: Storable a => Handle -> Array a -> IO ()
229writeArray _ arr | A.length arr == 0 = return ()
230writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen
233 let p = unsafeForeignPtrToPtr aStart
236-------------------------------------------------------------------------------
237-- Stream of Arrays IO
238-------------------------------------------------------------------------------
240-------------------------------------------------------------------------------
242-------------------------------------------------------------------------------
244-- | Write a stream of arrays to a handle.
247{-# INLINE fromStreamArrays #-}
248fromStreamArrays :: (MonadIO m, Storable a)
249 => Handle -> SerialT m (Array a) -> m ()
250fromStreamArrays h m = S.mapM_ (liftIO . writeArray h) m
252-- | @fromStreamArraysInChunksOf chunkSize handle stream@ writes a stream of arrays
253-- to @handle@ after coalescing the adjacent arrays in chunks of @chunkSize@.
254-- The chunk size is only a maximum and the actual writes could be smaller as
255-- we do not split the arrays to fit exactly to the specified size.
258{-# INLINE fromStreamArraysInChunksOf #-}
259fromStreamArraysInChunksOf :: (MonadIO m, Storable a)
260 => Int -> Handle -> SerialT m (Array a) -> m ()
261fromStreamArraysInChunksOf n h xs = fromStreamArrays h $ AS.compact n xs
263-- | @fromStreamInChunksOf chunkSize handle stream@ writes @stream@ to @handle@ in
264-- chunks of @chunkSize@. A write is performed to the IO device as soon as we
265-- collect the required input size.
268{-# INLINE fromStreamInChunksOf #-}
269fromStreamInChunksOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
270fromStreamInChunksOf n h m = fromStreamArrays h $ S.arraysOf n m
271-- fromStreamInChunksOf n h m = fromStreamArrays h $ AS.arraysOf n m
273-- > write = 'writeInChunksOf' A.defaultChunkSize
275-- | Write a byte stream to a file handle. Accumulates the input in chunks of
276-- up to 'A.defaultChunkSize' before writing.
278-- NOTE: This may perform better than the 'write' fold, you can try this if you
279-- need some extra perf boost.
282{-# INLINE fromStream #-}
283fromStream :: MonadIO m => Handle -> SerialT m Word8 -> m ()
284fromStream = fromStreamInChunksOf defaultChunkSize
286-- | Write a stream of arrays to a handle.
289{-# INLINE writeArrays #-}
290writeArrays :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
291writeArrays h = FL.drainBy (liftIO . writeArray h)
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.
300{-# INLINE writeArraysInChunksOf #-}
301writeArraysInChunksOf :: (MonadIO m, Storable a)
302 => Int -> Handle -> Fold m (Array a) ()
303writeArraysInChunksOf n h = lpackArraysChunksOf n (writeArrays h)
305-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
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
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.
318{-# INLINE writeInChunksOf #-}
319writeInChunksOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
320writeInChunksOf n h = FL.lchunksOf n (writeNUnsafe n) (writeArrays h)
322-- > write = 'writeInChunksOf' A.defaultChunkSize
324-- | Write a byte stream to a file handle. Accumulates the input in chunks of
325-- up to 'A.defaultChunkSize' before writing.
329write :: MonadIO m => Handle -> Fold m Word8 ()
330write = writeInChunksOf defaultChunkSize
334write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
335write = toHandleWith A.defaultChunkSize
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
342-------------------------------------------------------------------------------
343-- IO with encoding/decoding Unicode characters
344-------------------------------------------------------------------------------
348-- > readUtf8 = decodeUtf8 . read
350-- Read a UTF8 encoded stream of unicode characters from a file handle.
353{-# INLINE readUtf8 #-}
354readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
355readUtf8 = decodeUtf8 . read
358-- > writeUtf8 h s = write h $ encodeUtf8 s
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.
364{-# INLINE writeUtf8 #-}
365writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
366writeUtf8 h s = write h $ encodeUtf8 s
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'
375{-# INLINE writeUtf8ByLines #-}
376writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
377writeUtf8ByLines = undefined
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.
383{-# INLINE readLines #-}
384readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
385readLines h f = foldLines (readUtf8 h) f
387-------------------------------------------------------------------------------
388-- Framing on a sequence
389-------------------------------------------------------------------------------
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
396{-# INLINE readFrames #-}
397readFrames :: (IsStream t, MonadIO m, Storable a)
398 => Array a -> Handle -> Fold m a b -> t m b
399readFrames = undefined -- foldFrames . read
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@.
405{-# INLINE writeByFrames #-}
406writeByFrames :: (IsStream t, MonadIO m, Storable a)
407 => Array a -> Handle -> t m a -> m ()
408writeByFrames = undefined
410-------------------------------------------------------------------------------
412-------------------------------------------------------------------------------
414-- | Write collecting the input in sessions of n seconds or if chunkSize
416{-# INLINE writeByChunksOrSessionsOf #-}
417writeByChunksOrSessionsOf :: MonadIO m
418 => Int -> Double -> Handle -> SerialT m Word8 -> m ()
419writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined
421-- | Write collecting the input in sessions of n seconds or if defaultChunkSize
423{-# INLINE writeBySessionsOf #-}
424writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()
425writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n
427-------------------------------------------------------------------------------
428-- Random Access IO (Seek)
429-------------------------------------------------------------------------------
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.
435-------------------------------------------------------------------------------
437-- | Read the element at the given index treating the file as an array.
440{-# INLINE readIndex #-}
441readIndex :: Storable a => Handle -> Int -> Maybe a
442readIndex arr i = undefined
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.
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.
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.
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.
460{-# INLINE readSliceWith #-}
461readSliceWith :: (IsStream t, MonadIO m, Storable a)
462 => Int -> Handle -> Int -> Int -> t m a
463readSliceWith chunkSize h pos len = undefined
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@.
470{-# INLINE readSlice #-}
471readSlice :: (IsStream t, MonadIO m, Storable a)
472 => Handle -> Int -> Int -> t m a
473readSlice = readSliceWith A.defaultChunkSize
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@.
480{-# INLINE readSliceRev #-}
481readSliceRev :: (IsStream t, MonadIO m, Storable a)
482 => Handle -> Int -> Int -> t m a
483readSliceRev h i count = undefined
485-- | Write the given element at the given index in the file.
488{-# INLINE writeIndex #-}
489writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
490writeIndex h i a = undefined
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@.
497{-# INLINE writeSlice #-}
498writeSlice :: (IsStream t, Monad m, Storable a)
499 => Handle -> Int -> Int -> t m a -> m ()
500writeSlice h i len s = undefined
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@.
507{-# INLINE writeSliceRev #-}
508writeSliceRev :: (IsStream t, Monad m, Storable a)
509 => Handle -> Int -> Int -> t m a -> m ()
510writeSliceRev 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