4-- Module : Streamly.Internal.FileSystem.Handle
5-- Copyright : (c) 2018 Composewell Technologies
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
12-- The fundamental singleton APIs are 'getChunk' and 'putChunk' and the
13-- fundamental stream APIs built on top of those are 'readChunksWithBufferOf'
14-- and 'writeChunks'. Rest of this module is just combinatorial programming
17module Streamly.Internal.FileSystem.Handle
19 -- ** Read from Handle
33 , readChunksWithBufferOf
35 , toChunksWithBufferOf
39 -- Byte stream write (Folds)
48 -- Byte stream write (Streams)
50 , putBytesWithBufferOf
55 , writeChunksWithBufferOf
57 -- -- * Array stream Write
58 , putChunksWithBufferOf
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 -- We can also generate the request pattern using a funciton.
70 -- , readFrom -- read from a given position to the end of file
71 -- , readFromRev -- read from a given position the beginning of file
72 -- , readTo -- read from beginning up to the given position
73 -- , readToRev -- read from end to the given position in file
79 -- , readChunksFromToWith
80 -- , readChunksFromThenToWith
83 -- , writeFrom -- start writing at the given position
85 -- , writeTo -- write from beginning up to the given position
91 -- , writeChunksFromTo
92 -- , writeChunksFromToWith
93 -- , writeChunksFromThenToWith
97import Control.Monad.IO.Class (MonadIO(..))
98import Data.Word (Word8)
99import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
100import Foreign.ForeignPtr (withForeignPtr)
101import Foreign.Ptr (minusPtr, plusPtr)
102import Foreign.Storable (Storable(..))
103import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
104import System.IO (Handle, hGetBufSome, hPutBuf)
105import Prelude hiding (read)
107import Streamly.Internal.BaseCompat
108import Streamly.Data.Fold (Fold)
109import Streamly.Internal.Data.Fold.Type (Fold2(..))
110import Streamly.Internal.Data.Unfold.Type (Unfold(..))
111import Streamly.Internal.Data.Array.Foreign.Type
112 (Array(..), writeNUnsafe, unsafeFreezeWithShrink)
113import Streamly.Internal.Data.Array.Foreign.Mut.Type (mutableArray)
114import Streamly.Internal.Data.Stream.Serial (SerialT)
115import Streamly.Internal.Data.Stream.StreamK.Type (IsStream, mkStream)
116import Streamly.Internal.Data.Array.Stream.Foreign (lpackArraysChunksOf)
117-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
118import Streamly.Internal.System.IO (defaultChunkSize)
120import qualified Streamly.Data.Fold as FL
121import qualified Streamly.Internal.Data.Fold.Type as FL
122import qualified Streamly.Internal.Data.Unfold as UF
123import qualified Streamly.Internal.Data.Array.Stream.Foreign as AS
124import qualified Streamly.Internal.Data.Stream.IsStream as S
125import qualified Streamly.Data.Array.Foreign as A
126import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
129-- >>> import qualified Streamly.Data.Array.Foreign as Array
130-- >>> import qualified Streamly.Data.Fold as Fold
131-- >>> import qualified Streamly.Data.Unfold as Unfold
132-- >>> import qualified Streamly.FileSystem.Handle as Handle
133-- >>> import qualified Streamly.Prelude as Stream
135-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
136-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold (supplyFirst)
137-- >>> import qualified Streamly.Internal.FileSystem.Handle as Handle
138-- >>> import qualified Streamly.Internal.System.IO as IO (defaultChunkSize)
140-------------------------------------------------------------------------------
142-------------------------------------------------------------------------------
144-- The following references may be useful to build an understanding about the
147-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
148-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
149-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
150-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
151-- https://www.w3.org/TR/FileAPI/ for http file API.
153-------------------------------------------------------------------------------
155-------------------------------------------------------------------------------
157-- | Read a 'ByteArray' from a file handle. If no data is available on the
158-- handle it blocks until some data becomes available. If data is available
159-- then it immediately returns that data without blocking. It reads a maximum
160-- of up to the size requested.
161{-# INLINABLE getChunk #-}
162getChunk :: Int -> Handle -> IO (Array Word8)
164 ptr <- mallocPlainForeignPtrBytes size
165 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
166 withForeignPtr ptr $ \p -> do
167 n <- hGetBufSome h p size
168 -- XXX shrink only if the diff is significant
170 unsafeFreezeWithShrink $
171 mutableArray ptr (p `plusPtr` n) (p `plusPtr` size)
173-------------------------------------------------------------------------------
174-- Stream of Arrays IO
175-------------------------------------------------------------------------------
177-- | @toChunksWithBufferOf size h@ reads a stream of arrays from file handle @h@.
178-- The maximum size of a single array is specified by @size@. The actual size
179-- read may be less than or equal to @size@.
180{-# INLINABLE _toChunksWithBufferOf #-}
181_toChunksWithBufferOf :: (IsStream t, MonadIO m)
182 => Int -> Handle -> t m (Array Word8)
183_toChunksWithBufferOf size h = go
185 -- XXX use cons/nil instead
186 go = mkStream $ \_ yld _ stp -> do
187 arr <- liftIO $ getChunk size h
192-- | @toChunksWithBufferOf size handle@ reads a stream of arrays from the file
193-- handle @handle@. The maximum size of a single array is limited to @size@.
194-- The actual size read may be less than or equal to @size@.
196-- >>> toChunksWithBufferOf size h = Stream.unfold Handle.readChunksWithBufferOf (size, h)
199{-# INLINE_NORMAL toChunksWithBufferOf #-}
200toChunksWithBufferOf :: (IsStream t, MonadIO m) =>
201 Int -> Handle -> t m (Array Word8)
202toChunksWithBufferOf size h = D.fromStreamD (D.Stream step ())
204 {-# INLINE_LATE step #-}
206 arr <- liftIO $ getChunk size h
212-- | Unfold the tuple @(bufsize, handle)@ into a stream of 'Word8' arrays.
213-- Read requests to the IO device are performed using a buffer of size
214-- @bufsize@. The size of an array in the resulting stream is always less than
215-- or equal to @bufsize@.
218{-# INLINE_NORMAL readChunksWithBufferOf #-}
219readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Handle) (Array Word8)
220readChunksWithBufferOf = Unfold step return
222 {-# INLINE_LATE step #-}
224 arr <- liftIO $ getChunk size h
228 _ -> D.Yield arr (size, h)
230-- XXX read 'Array a' instead of Word8
232-- | @toChunks handle@ reads a stream of arrays from the specified file
233-- handle. The maximum size of a single array is limited to
234-- @defaultChunkSize@. The actual size read may be less than or equal to
235-- @defaultChunkSize@.
237-- >>> toChunks = Handle.toChunksWithBufferOf IO.defaultChunkSize
240{-# INLINE toChunks #-}
241toChunks :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
242toChunks = toChunksWithBufferOf defaultChunkSize
244-- | Unfolds a handle into a stream of 'Word8' arrays. Requests to the IO
245-- device are performed using a buffer of size
246-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'. The
247-- size of arrays in the resulting stream are therefore less than or equal to
248-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'.
250-- >>> readChunks = Unfold.supplyFirst IO.defaultChunkSize Handle.readChunksWithBufferOf
253{-# INLINE readChunks #-}
254readChunks :: MonadIO m => Unfold m Handle (Array Word8)
255readChunks = UF.supplyFirst defaultChunkSize readChunksWithBufferOf
257-------------------------------------------------------------------------------
258-- Read File to Stream
259-------------------------------------------------------------------------------
261-- TODO for concurrent streams implement readahead IO. We can send multiple
262-- read requests at the same time. For serial case we can use async IO. We can
263-- also control the read throughput in mbps or IOPS.
265-- | Unfolds the tuple @(bufsize, handle)@ into a byte stream, read requests
266-- to the IO device are performed using buffers of @bufsize@.
268-- >>> readWithBufferOf = Unfold.many Handle.readChunksWithBufferOf Array.read
271{-# INLINE readWithBufferOf #-}
272readWithBufferOf :: MonadIO m => Unfold m (Int, Handle) Word8
273readWithBufferOf = UF.many readChunksWithBufferOf A.read
275-- | @toBytesWithBufferOf bufsize handle@ reads a byte stream from a file
276-- handle, reads are performed in chunks of up to @bufsize@.
278-- >>> toBytesWithBufferOf size h = Stream.unfoldMany Array.read $ Handle.toChunksWithBufferOf size h
281{-# INLINE toBytesWithBufferOf #-}
282toBytesWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
283toBytesWithBufferOf size h = AS.concat $ toChunksWithBufferOf size h
286-- Generate a stream of elements of the given type from a file 'Handle'.
287-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
289-- | Unfolds a file handle into a byte stream. IO requests to the device are
290-- performed in sizes of
291-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'.
293-- >>> read = Unfold.many Handle.readChunks Array.read
297read :: MonadIO m => Unfold m Handle Word8
298read = UF.many readChunks A.read
300-- | Generate a byte stream from a file 'Handle'.
302-- >>> toBytes h = Stream.unfoldMany Array.read $ Handle.toChunks h
305{-# INLINE toBytes #-}
306toBytes :: (IsStream t, MonadIO m) => Handle -> t m Word8
307toBytes = AS.concat . toChunks
309-------------------------------------------------------------------------------
311-------------------------------------------------------------------------------
313-------------------------------------------------------------------------------
315-------------------------------------------------------------------------------
317-- | Write an 'Array' to a file handle.
320{-# INLINABLE putChunk #-}
321putChunk :: (MonadIO m, Storable a) => Handle -> Array a -> m ()
322putChunk _ arr | A.length arr == 0 = return ()
323putChunk h Array{..} =
324 liftIO $ unsafeWithForeignPtr aStart $ \p -> hPutBuf h p aLen
329 let p = unsafeForeignPtrToPtr aStart
332-------------------------------------------------------------------------------
333-- Stream of Arrays IO
334-------------------------------------------------------------------------------
336-------------------------------------------------------------------------------
338-------------------------------------------------------------------------------
340-- XXX use an unfold to fromObjects or fromUnfold so that we can put any object
341-- | Write a stream of arrays to a handle.
343-- >>> putChunks h = Stream.mapM_ (Handle.putChunk h)
346{-# INLINE putChunks #-}
347putChunks :: (MonadIO m, Storable a)
348 => Handle -> SerialT m (Array a) -> m ()
349putChunks h = S.mapM_ (putChunk h)
351-- XXX AS.compact can be written idiomatically in terms of foldMany, just like
352-- AS.concat is written in terms of foldMany. Once that is done we can write
353-- idiomatic def in the docs.
355-- | @putChunksWithBufferOf bufsize handle stream@ writes a stream of arrays
356-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
357-- The chunk size is only a maximum and the actual writes could be smaller as
358-- we do not split the arrays to fit exactly to the specified size.
361{-# INLINE putChunksWithBufferOf #-}
362putChunksWithBufferOf :: (MonadIO m, Storable a)
363 => Int -> Handle -> SerialT m (Array a) -> m ()
364putChunksWithBufferOf n h xs = putChunks h $ AS.compact n xs
366-- | @putBytesWithBufferOf bufsize handle stream@ writes @stream@ to @handle@
367-- in chunks of @bufsize@. A write is performed to the IO device as soon as we
368-- collect the required input size.
370-- >>> putBytesWithBufferOf n h m = Handle.putChunks h $ Stream.arraysOf n m
373{-# INLINE putBytesWithBufferOf #-}
374putBytesWithBufferOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
375putBytesWithBufferOf n h m = putChunks h $ S.arraysOf n m
376-- putBytesWithBufferOf n h m = putChunks h $ AS.arraysOf n m
378-- | Write a byte stream to a file handle. Accumulates the input in chunks of
379-- up to 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize' before writing.
381-- NOTE: This may perform better than the 'write' fold, you can try this if you
382-- need some extra perf boost.
384-- >>> putBytes = Handle.putBytesWithBufferOf IO.defaultChunkSize
387{-# INLINE putBytes #-}
388putBytes :: MonadIO m => Handle -> SerialT m Word8 -> m ()
389putBytes = putBytesWithBufferOf defaultChunkSize
391-- | Write a stream of arrays to a handle. Each array in the stream is written
392-- to the device as a separate IO request.
394-- writeChunks h = Fold.drainBy (Handle.putChunk h)
397{-# INLINE writeChunks #-}
398writeChunks :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
399writeChunks h = FL.drainBy (putChunk h)
401{-# INLINE writeChunks2 #-}
402writeChunks2 :: (MonadIO m, Storable a) => Fold2 m Handle (Array a) ()
404 Fold2 (\h arr -> liftIO $ putChunk h arr >> return h)
408-- XXX lpackArraysChunksOf should be written idiomatically
410-- | @writeChunksWithBufferOf bufsize handle@ writes a stream of arrays
411-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
412-- We never split an array, if a single array is bigger than the specified size
413-- it emitted as it is. Multiple arrays are coalesed as long as the total size
414-- remains below the specified size.
417{-# INLINE writeChunksWithBufferOf #-}
418writeChunksWithBufferOf :: (MonadIO m, Storable a)
419 => Int -> Handle -> Fold m (Array a) ()
420writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)
422-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
425-- Note that if you use a chunk size less than 8K (GHC's default buffer
426-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
427-- do not want buffering to occur at GHC level as well. Same thing applies to
430-- XXX Maybe we should have a Fold.arraysOf like we have Stream.arraysOf
432-- | @writeWithBufferOf reqSize handle@ writes the input stream to @handle@.
433-- Bytes in the input stream are collected into a buffer until we have a chunk
434-- of @reqSize@ and then written to the IO device.
436-- >>> writeWithBufferOf n h = Fold.chunksOf n (Array.writeNUnsafe n) (Handle.writeChunks h)
439{-# INLINE writeWithBufferOf #-}
440writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
441writeWithBufferOf n h = FL.chunksOf n (writeNUnsafe n) (writeChunks h)
443{-# INLINE writeWithBufferOf2 #-}
444writeWithBufferOf2 :: MonadIO m => Int -> Fold2 m Handle Word8 ()
445writeWithBufferOf2 n = FL.chunksOf2 n (writeNUnsafe n) writeChunks2
447-- | Write a byte stream to a file handle. Accumulates the input in chunks of
448-- up to 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize' before writing
451-- >>> write = Handle.writeWithBufferOf IO.defaultChunkSize
455write :: MonadIO m => Handle -> Fold m Word8 ()
456write = writeWithBufferOf defaultChunkSize
459write2 :: MonadIO m => Fold2 m Handle Word8 ()
460write2 = writeWithBufferOf2 defaultChunkSize
464write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
465write = toHandleWith A.defaultChunkSize
468-- XXX mmap a file into an array. This could be useful for in-place operations
469-- on a file. For example, we can quicksort the contents of a file by mmapping
472-------------------------------------------------------------------------------
473-- IO with encoding/decoding Unicode characters
474-------------------------------------------------------------------------------
478-- > readUtf8 = decodeUtf8 . read
480-- Read a UTF8 encoded stream of unicode characters from a file handle.
483{-# INLINE readUtf8 #-}
484readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
485readUtf8 = decodeUtf8 . read
488-- > writeUtf8 h s = write h $ encodeUtf8 s
490-- Encode a stream of unicode characters to UTF8 and write it to the given file
491-- handle. Default block buffering applies to the writes.
494{-# INLINE writeUtf8 #-}
495writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
496writeUtf8 h s = write h $ encodeUtf8 s
498-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
499-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
500-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
501-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
505{-# INLINE writeUtf8ByLines #-}
506writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
507writeUtf8ByLines = undefined
509-- | Read UTF-8 lines from a file handle and apply the specified fold to each
510-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
513{-# INLINE readLines #-}
514readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
515readLines h f = foldLines (readUtf8 h) f
517-------------------------------------------------------------------------------
518-- Framing on a sequence
519-------------------------------------------------------------------------------
521-- | Read a stream from a file handle and split it into frames delimited by
522-- the specified sequence of elements. The supplied fold is applied on each
526{-# INLINE readFrames #-}
527readFrames :: (IsStream t, MonadIO m, Storable a)
528 => Array a -> Handle -> Fold m a b -> t m b
529readFrames = undefined -- foldFrames . read
531-- | Write a stream to the given file handle buffering up to frames separated
532-- by the given sequence or up to a maximum of @defaultChunkSize@.
535{-# INLINE writeByFrames #-}
536writeByFrames :: (IsStream t, MonadIO m, Storable a)
537 => Array a -> Handle -> t m a -> m ()
538writeByFrames = undefined
540-------------------------------------------------------------------------------
542-------------------------------------------------------------------------------
544-- | Write collecting the input in sessions of n seconds or if chunkSize
546{-# INLINE writeByChunksOrSessionsOf #-}
547writeByChunksOrSessionsOf :: MonadIO m
548 => Int -> Double -> Handle -> SerialT m Word8 -> m ()
549writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined
551-- | Write collecting the input in sessions of n seconds or if defaultChunkSize
553{-# INLINE writeBySessionsOf #-}
554writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()
555writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n
557-------------------------------------------------------------------------------
558-- Random Access IO (Seek)
559-------------------------------------------------------------------------------
561-- XXX handles could be shared, so we may not want to use the handle state at
562-- all for these APIs. we can use pread and pwrite instead. On windows we will
563-- need to use readFile/writeFile with an offset argument.
565-------------------------------------------------------------------------------
567-- | Read the element at the given index treating the file as an array.
570{-# INLINE readIndex #-}
571readIndex :: Storable a => Handle -> Int -> Maybe a
572readIndex arr i = undefined
574-- NOTE: To represent a range to read we have chosen (start, size) instead of
575-- (start, end). This helps in removing the ambiguity of whether "end" is
576-- included in the range or not.
578-- We could avoid specifying the range to be read and instead use "take size"
579-- on the stream, but it may end up reading more and then consume it partially.
581-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
582-- from @handle@ starting at the offset @pos@ from the beginning of the file.
584-- Reads are performed in chunks of size @chunkSize@. For block devices, to
585-- avoid reading partial blocks @chunkSize@ must align with the block size of
586-- the underlying device. If the underlying block size is unknown, it is a good
587-- idea to keep it a multiple 4KiB. This API ensures that the start of each
588-- chunk is aligned with @chunkSize@ from second chunk onwards.
590{-# INLINE readSliceWith #-}
591readSliceWith :: (IsStream t, MonadIO m, Storable a)
592 => Int -> Handle -> Int -> Int -> t m a
593readSliceWith chunkSize h pos len = undefined
595-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
596-- at index @i@ and reading up to @count@ elements in the forward direction
597-- ending at the index @i + count - 1@.
600{-# INLINE readSlice #-}
601readSlice :: (IsStream t, MonadIO m, Storable a)
602 => Handle -> Int -> Int -> t m a
603readSlice = readSliceWith A.defaultChunkSize
605-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
606-- at index @i@ and reading up to @count@ elements in the reverse direction
607-- ending at the index @i - count + 1@.
610{-# INLINE readSliceRev #-}
611readSliceRev :: (IsStream t, MonadIO m, Storable a)
612 => Handle -> Int -> Int -> t m a
613readSliceRev h i count = undefined
615-- | Write the given element at the given index in the file.
618{-# INLINE writeIndex #-}
619writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
620writeIndex h i a = undefined
622-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
623-- starting at index @i@ and writing up to @count@ elements in the forward
624-- direction ending at the index @i + count - 1@.
627{-# INLINE writeSlice #-}
628writeSlice :: (IsStream t, Monad m, Storable a)
629 => Handle -> Int -> Int -> t m a -> m ()
630writeSlice h i len s = undefined
632-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
633-- starting at index @i@ and writing up to @count@ elements in the reverse
634-- direction ending at the index @i - count + 1@.
637{-# INLINE writeSliceRev #-}
638writeSliceRev :: (IsStream t, Monad m, Storable a)
639 => Handle -> Int -> Int -> t m a -> m ()
640writeSliceRev 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