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
34 , readChunksRequestsOf
40 -- Byte stream write (Folds)
47 -- Byte stream write (Streams)
49 , fromStreamInChunksOf
54 , writeChunksRequestsOf
56 -- -- * Array stream Write
58 , fromStreamArraysInChunksOf
60 -- -- * Random Access (Seek)
61 -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or
62 -- files that have random access or seek capability. This type of devices
63 -- include disks, files, memory devices and exclude terminals, pipes,
66 -- XXX need to decide whether to use readFromStepN style or readFromThenTo
67 -- style. The latter is consistent with list enumeration. The former may be
68 -- more unambiguous and it is easier and clearer to specify 0 elements.
70 -- We can also generate the request pattern using a funciton.
73 -- , readFrom -- read from a given position to the end of file
74 -- , readFromRev -- read from a given position the beginning of file
75 -- , readTo -- read from beginning up to the given position
76 -- , readToRev -- read from end to the given position in file
82 -- , readChunksFromToRequestsOf
83 -- , readChunksFromThenToRequestsOf
86 -- , writeFrom -- start writing at the given position
88 -- , writeTo -- write from beginning up to the given position
94 -- , writeChunksFromTo
95 -- , writeChunksFromToRequestsOf
96 -- , writeChunksFromThenToRequestsOf
100import Control.Monad.IO.Class (MonadIO(..))
101import Data.Word (Word8)
102import Foreign.ForeignPtr (withForeignPtr)
103import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
104import Foreign.Ptr (minusPtr, plusPtr)
105import Foreign.Storable (Storable(..))
106import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
107import System.IO (Handle, hGetBufSome, hPutBuf)
108import Prelude hiding (read)
110import Streamly.Data.Fold (Fold)
111import Streamly.Internal.Data.Unfold.Types (Unfold(..))
112import Streamly.Internal.Memory.Array.Types
113 (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,
115import Streamly.Streams.Serial (SerialT)
116import Streamly.Streams.StreamK.Type (IsStream, mkStream)
117-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
119import qualified Streamly.Data.Fold as FL
120import qualified Streamly.Internal.Data.Fold.Types as FL
121import qualified Streamly.Internal.Data.Unfold as UF
122import qualified Streamly.Internal.Memory.ArrayStream as AS
123import qualified Streamly.Internal.Prelude as S
124import qualified Streamly.Memory.Array as A
125import qualified Streamly.Streams.StreamD.Type as D
127-------------------------------------------------------------------------------
129-------------------------------------------------------------------------------
131-- The following references may be useful to build an understanding about the
134-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
135-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
136-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
137-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
138-- https://www.w3.org/TR/FileAPI/ for http file API.
140-------------------------------------------------------------------------------
142-------------------------------------------------------------------------------
144-- | Read a 'ByteArray' from a file handle. If no data is available on the
145-- handle it blocks until some data becomes available. If data is available
146-- then it immediately returns that data without blocking. It reads a maximum
147-- of up to the size requested.
148{-# INLINABLE readArrayUpto #-}
149readArrayUpto :: Int -> Handle -> IO (Array Word8)
150readArrayUpto size h = do
151 ptr <- mallocPlainForeignPtrBytes size
152 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
153 withForeignPtr ptr $ \p -> do
154 n <- hGetBufSome h p size
157 , aEnd = p `plusPtr` n
158 , aBound = p `plusPtr` size
160 -- XXX shrink only if the diff is significant
163-------------------------------------------------------------------------------
164-- Stream of Arrays IO
165-------------------------------------------------------------------------------
167-- | @toChunksRequestsOf size h@ reads a stream of arrays from file handle @h@.
168-- The maximum size of a single array is specified by @size@. The actual size
169-- read may be less than or equal to @size@.
170{-# INLINABLE _toChunksRequestsOf #-}
171_toChunksRequestsOf :: (IsStream t, MonadIO m)
172 => Int -> Handle -> t m (Array Word8)
173_toChunksRequestsOf size h = go
175 -- XXX use cons/nil instead
176 go = mkStream $ \_ yld _ stp -> do
177 arr <- liftIO $ readArrayUpto size h
182-- | @toChunksRequestsOf size handle@ reads a stream of arrays from the file
183-- handle @handle@. The maximum size of a single array is limited to @size@.
184-- The actual size read may be less than or equal to @size@.
187{-# INLINE_NORMAL toChunksRequestsOf #-}
188toChunksRequestsOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
189toChunksRequestsOf size h = D.fromStreamD (D.Stream step ())
191 {-# INLINE_LATE step #-}
193 arr <- liftIO $ readArrayUpto size h
199-- | Unfold the tuple @(size, handle)@ into a stream of 'Word8' arrays. The
200-- stream consists of arrays representing chunks of data read from the handle.
201-- The size of an array in the stream is less than or equal to @size@.
204{-# INLINE_NORMAL readChunksRequestsOf #-}
205readChunksRequestsOf :: MonadIO m => Unfold m (Int, Handle) (Array Word8)
206readChunksRequestsOf = Unfold step return
208 {-# INLINE_LATE step #-}
210 arr <- liftIO $ readArrayUpto size h
214 _ -> D.Yield arr (size, h)
216-- XXX read 'Array a' instead of Word8
218-- | @toChunks handle@ reads a stream of arrays from the specified file
219-- handle. The maximum size of a single array is limited to
220-- @defaultChunkSize@. The actual size read may be less than or equal to
221-- @defaultChunkSize@.
223-- > toChunks = toChunksRequestsOf defaultChunkSize
226{-# INLINE toChunks #-}
227toChunks :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
228toChunks = toChunksRequestsOf defaultChunkSize
230-- | Unfolds a handle into a stream of 'Word8' arrays. Requests to the IO
231-- device are performed using a buffer of size
232-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'. The
233-- size of arrays in the resulting stream are therefore less than or equal to
234-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
237{-# INLINE readChunks #-}
238readChunks :: MonadIO m => Unfold m Handle (Array Word8)
239readChunks = UF.supplyFirst readChunksRequestsOf defaultChunkSize
241-------------------------------------------------------------------------------
242-- Read File to Stream
243-------------------------------------------------------------------------------
245-- TODO for concurrent streams implement readahead IO. We can send multiple
246-- read requests at the same time. For serial case we can use async IO. We can
247-- also control the read throughput in mbps or IOPS.
249-- | Unfolds the tuple @(chunksize, handle)@ into a byte stream, read requests
250-- to the IO device are performed using buffers of @chunkSize@.
253{-# INLINE readRequestsOf #-}
254readRequestsOf :: MonadIO m => Unfold m (Int, Handle) Word8
255readRequestsOf = UF.concat readChunksRequestsOf A.read
257-- | @toStreamRequestsOf chunkSize handle@ reads a byte stream from a file
258-- handle, reads are performed in chunks of up to @chunkSize@.
261{-# INLINE toStreamRequestsOf #-}
262toStreamRequestsOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
263toStreamRequestsOf chunkSize h = AS.concat $ toChunksRequestsOf chunkSize h
266-- Generate a stream of elements of the given type from a file 'Handle'.
267-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
269-- | Unfolds a file handle into a byte stream. IO requests to the device are
270-- performed in sizes of
271-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
275read :: MonadIO m => Unfold m Handle Word8
276read = UF.supplyFirst readRequestsOf defaultChunkSize
278-- | Generate a byte stream from a file 'Handle'.
281{-# INLINE toStream #-}
282toStream :: (IsStream t, MonadIO m) => Handle -> t m Word8
283toStream = AS.concat . toChunks
285-------------------------------------------------------------------------------
287-------------------------------------------------------------------------------
289-------------------------------------------------------------------------------
291-------------------------------------------------------------------------------
293-- | Write an 'Array' to a file handle.
296{-# INLINABLE writeArray #-}
297writeArray :: Storable a => Handle -> Array a -> IO ()
298writeArray _ arr | A.length arr == 0 = return ()
299writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen
302 let p = unsafeForeignPtrToPtr aStart
305-------------------------------------------------------------------------------
306-- Stream of Arrays IO
307-------------------------------------------------------------------------------
309-------------------------------------------------------------------------------
311-------------------------------------------------------------------------------
313-- | Write a stream of arrays to a handle.
316{-# INLINE fromStreamArrays #-}
317fromStreamArrays :: (MonadIO m, Storable a)
318 => Handle -> SerialT m (Array a) -> m ()
319fromStreamArrays h m = S.mapM_ (liftIO . writeArray h) m
321-- | @fromStreamArraysInChunksOf chunkSize handle stream@ writes a stream of arrays
322-- to @handle@ after coalescing the adjacent arrays in chunks of @chunkSize@.
323-- The chunk size is only a maximum and the actual writes could be smaller as
324-- we do not split the arrays to fit exactly to the specified size.
327{-# INLINE fromStreamArraysInChunksOf #-}
328fromStreamArraysInChunksOf :: (MonadIO m, Storable a)
329 => Int -> Handle -> SerialT m (Array a) -> m ()
330fromStreamArraysInChunksOf n h xs = fromStreamArrays h $ AS.compact n xs
332-- | @fromStreamInChunksOf chunkSize handle stream@ writes @stream@ to @handle@ in
333-- chunks of @chunkSize@. A write is performed to the IO device as soon as we
334-- collect the required input size.
337{-# INLINE fromStreamInChunksOf #-}
338fromStreamInChunksOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
339fromStreamInChunksOf n h m = fromStreamArrays h $ S.arraysOf n m
340-- fromStreamInChunksOf n h m = fromStreamArrays h $ AS.arraysOf n m
342-- > write = 'writeRequestsOf' A.defaultChunkSize
344-- | Write a byte stream to a file handle. Accumulates the input in chunks of
345-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing.
347-- NOTE: This may perform better than the 'write' fold, you can try this if you
348-- need some extra perf boost.
351{-# INLINE fromStream #-}
352fromStream :: MonadIO m => Handle -> SerialT m Word8 -> m ()
353fromStream = fromStreamInChunksOf defaultChunkSize
355-- | Write a stream of arrays to a handle. Each array in the stream is written
356-- to the device as a separate IO request.
359{-# INLINE writeChunks #-}
360writeChunks :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
361writeChunks h = FL.drainBy (liftIO . writeArray h)
363-- | @writeChunksRequestsOf chunkSize handle@ writes a stream of arrays
364-- to @handle@ after coalescing the adjacent arrays in chunks of @chunkSize@.
365-- We never split an array, if a single array is bigger than the specified size
366-- it emitted as it is. Multiple arrays are coalesed as long as the total size
367-- remains below the specified size.
370{-# INLINE writeChunksRequestsOf #-}
371writeChunksRequestsOf :: (MonadIO m, Storable a)
372 => Int -> Handle -> Fold m (Array a) ()
373writeChunksRequestsOf n h = lpackArraysChunksOf n (writeChunks h)
375-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
378-- Note that if you use a chunk size less than 8K (GHC's default buffer
379-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
380-- do not want buffering to occur at GHC level as well. Same thing applies to
383-- | @writeRequestsOf reqSize handle@ writes the input stream to @handle@.
384-- Bytes in the input stream are collected into a buffer until we have a chunk
385-- of @reqSize@ and then written to the IO device.
388{-# INLINE writeRequestsOf #-}
389writeRequestsOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
390writeRequestsOf n h = FL.lchunksOf n (writeNUnsafe n) (writeChunks h)
392-- > write = 'writeRequestsOf' A.defaultChunkSize
394-- | Write a byte stream to a file handle. Accumulates the input in chunks of
395-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing
400write :: MonadIO m => Handle -> Fold m Word8 ()
401write = writeRequestsOf defaultChunkSize
405write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
406write = toHandleWith A.defaultChunkSize
409-- XXX mmap a file into an array. This could be useful for in-place operations
410-- on a file. For example, we can quicksort the contents of a file by mmapping
413-------------------------------------------------------------------------------
414-- IO with encoding/decoding Unicode characters
415-------------------------------------------------------------------------------
419-- > readUtf8 = decodeUtf8 . read
421-- Read a UTF8 encoded stream of unicode characters from a file handle.
424{-# INLINE readUtf8 #-}
425readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
426readUtf8 = decodeUtf8 . read
429-- > writeUtf8 h s = write h $ encodeUtf8 s
431-- Encode a stream of unicode characters to UTF8 and write it to the given file
432-- handle. Default block buffering applies to the writes.
435{-# INLINE writeUtf8 #-}
436writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
437writeUtf8 h s = write h $ encodeUtf8 s
439-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
440-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
441-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
442-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
446{-# INLINE writeUtf8ByLines #-}
447writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
448writeUtf8ByLines = undefined
450-- | Read UTF-8 lines from a file handle and apply the specified fold to each
451-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
454{-# INLINE readLines #-}
455readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
456readLines h f = foldLines (readUtf8 h) f
458-------------------------------------------------------------------------------
459-- Framing on a sequence
460-------------------------------------------------------------------------------
462-- | Read a stream from a file handle and split it into frames delimited by
463-- the specified sequence of elements. The supplied fold is applied on each
467{-# INLINE readFrames #-}
468readFrames :: (IsStream t, MonadIO m, Storable a)
469 => Array a -> Handle -> Fold m a b -> t m b
470readFrames = undefined -- foldFrames . read
472-- | Write a stream to the given file handle buffering up to frames separated
473-- by the given sequence or up to a maximum of @defaultChunkSize@.
476{-# INLINE writeByFrames #-}
477writeByFrames :: (IsStream t, MonadIO m, Storable a)
478 => Array a -> Handle -> t m a -> m ()
479writeByFrames = undefined
481-------------------------------------------------------------------------------
483-------------------------------------------------------------------------------
485-- | Write collecting the input in sessions of n seconds or if chunkSize
487{-# INLINE writeByChunksOrSessionsOf #-}
488writeByChunksOrSessionsOf :: MonadIO m
489 => Int -> Double -> Handle -> SerialT m Word8 -> m ()
490writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined
492-- | Write collecting the input in sessions of n seconds or if defaultChunkSize
494{-# INLINE writeBySessionsOf #-}
495writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()
496writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n
498-------------------------------------------------------------------------------
499-- Random Access IO (Seek)
500-------------------------------------------------------------------------------
502-- XXX handles could be shared, so we may not want to use the handle state at
503-- all for these APIs. we can use pread and pwrite instead. On windows we will
504-- need to use readFile/writeFile with an offset argument.
506-------------------------------------------------------------------------------
508-- | Read the element at the given index treating the file as an array.
511{-# INLINE readIndex #-}
512readIndex :: Storable a => Handle -> Int -> Maybe a
513readIndex arr i = undefined
515-- NOTE: To represent a range to read we have chosen (start, size) instead of
516-- (start, end). This helps in removing the ambiguity of whether "end" is
517-- included in the range or not.
519-- We could avoid specifying the range to be read and instead use "take size"
520-- on the stream, but it may end up reading more and then consume it partially.
522-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
523-- from @handle@ starting at the offset @pos@ from the beginning of the file.
525-- Reads are performed in chunks of size @chunkSize@. For block devices, to
526-- avoid reading partial blocks @chunkSize@ must align with the block size of
527-- the underlying device. If the underlying block size is unknown, it is a good
528-- idea to keep it a multiple 4KiB. This API ensures that the start of each
529-- chunk is aligned with @chunkSize@ from second chunk onwards.
531{-# INLINE readSliceWith #-}
532readSliceWith :: (IsStream t, MonadIO m, Storable a)
533 => Int -> Handle -> Int -> Int -> t m a
534readSliceWith chunkSize h pos len = undefined
536-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
537-- at index @i@ and reading up to @count@ elements in the forward direction
538-- ending at the index @i + count - 1@.
541{-# INLINE readSlice #-}
542readSlice :: (IsStream t, MonadIO m, Storable a)
543 => Handle -> Int -> Int -> t m a
544readSlice = readSliceWith A.defaultChunkSize
546-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
547-- at index @i@ and reading up to @count@ elements in the reverse direction
548-- ending at the index @i - count + 1@.
551{-# INLINE readSliceRev #-}
552readSliceRev :: (IsStream t, MonadIO m, Storable a)
553 => Handle -> Int -> Int -> t m a
554readSliceRev h i count = undefined
556-- | Write the given element at the given index in the file.
559{-# INLINE writeIndex #-}
560writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
561writeIndex h i a = undefined
563-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
564-- starting at index @i@ and writing up to @count@ elements in the forward
565-- direction ending at the index @i + count - 1@.
568{-# INLINE writeSlice #-}
569writeSlice :: (IsStream t, Monad m, Storable a)
570 => Handle -> Int -> Int -> t m a -> m ()
571writeSlice h i len s = undefined
573-- | @writeSliceRev 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 reverse
575-- direction ending at the index @i - count + 1@.
578{-# INLINE writeSliceRev #-}
579writeSliceRev :: (IsStream t, Monad m, Storable a)
580 => Handle -> Int -> Int -> t m a -> m ()
581writeSliceRev 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