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
35 , readChunksWithBufferOf
37 , toChunksWithBufferOf
42 -- Byte stream write (Folds)
49 -- Byte stream write (Streams)
51 , fromBytesWithBufferOf
56 , writeChunksWithBufferOf
58 -- -- * Array stream Write
59 , fromChunksWithBufferOf
64 -- -- * Random Access (Seek)
65 -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or
66 -- files that have random access or seek capability. This type of devices
67 -- include disks, files, memory devices and exclude terminals, pipes,
70 -- XXX need to decide whether to use readFromStepN style or readFromThenTo
71 -- style. The latter is consistent with list enumeration. The former may be
72 -- more unambiguous and it is easier and clearer to specify 0 elements.
74 -- We can also generate the request pattern using a funciton.
77 -- , readFrom -- read from a given position to the end of file
78 -- , readFromRev -- read from a given position the beginning of file
79 -- , readTo -- read from beginning up to the given position
80 -- , readToRev -- read from end to the given position in file
86 -- , readChunksFromToWithBufferOf
87 -- , readChunksFromThenToWithBufferOf
90 -- , writeFrom -- start writing at the given position
92 -- , writeTo -- write from beginning up to the given position
98 -- , writeChunksFromTo
99 -- , writeChunksFromToWithBufferOf
100 -- , writeChunksFromThenToWithBufferOf
104import Control.Monad.IO.Class (MonadIO(..))
105import Data.Word (Word8)
106import Foreign.ForeignPtr (withForeignPtr)
107import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
108import Foreign.Ptr (minusPtr, plusPtr)
109import Foreign.Storable (Storable(..))
110import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
111import System.IO (Handle, hGetBufSome, hPutBuf, stdin, stdout)
112import Prelude hiding (read)
114import Streamly.Data.Fold (Fold)
115import Streamly.Internal.Data.Unfold.Types (Unfold(..))
116import Streamly.Internal.Memory.Array.Types
117 (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,
119import Streamly.Streams.Serial (SerialT)
120import Streamly.Streams.StreamK.Type (IsStream, mkStream)
121-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
123import qualified Streamly.Data.Fold as FL
124import qualified Streamly.Internal.Data.Fold.Types as FL
125import qualified Streamly.Internal.Data.Unicode.Stream as U
126import qualified Streamly.Internal.Data.Unfold as UF
127import qualified Streamly.Internal.Memory.ArrayStream as AS
128import qualified Streamly.Internal.Prelude as S
129import qualified Streamly.Memory.Array as A
130import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
132-------------------------------------------------------------------------------
134-------------------------------------------------------------------------------
136-- The following references may be useful to build an understanding about the
139-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
140-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
141-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
142-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
143-- https://www.w3.org/TR/FileAPI/ for http file API.
145-------------------------------------------------------------------------------
147-------------------------------------------------------------------------------
149-- | Read a 'ByteArray' from a file handle. If no data is available on the
150-- handle it blocks until some data becomes available. If data is available
151-- then it immediately returns that data without blocking. It reads a maximum
152-- of up to the size requested.
153{-# INLINABLE readArrayUpto #-}
154readArrayUpto :: Int -> Handle -> IO (Array Word8)
155readArrayUpto size h = do
156 ptr <- mallocPlainForeignPtrBytes size
157 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
158 withForeignPtr ptr $ \p -> do
159 n <- hGetBufSome h p size
162 , aEnd = p `plusPtr` n
163 , aBound = p `plusPtr` size
165 -- XXX shrink only if the diff is significant
168-------------------------------------------------------------------------------
169-- Stream of Arrays IO
170-------------------------------------------------------------------------------
172-- | @toChunksWithBufferOf size h@ reads a stream of arrays from file handle @h@.
173-- The maximum size of a single array is specified by @size@. The actual size
174-- read may be less than or equal to @size@.
175{-# INLINABLE _toChunksWithBufferOf #-}
176_toChunksWithBufferOf :: (IsStream t, MonadIO m)
177 => Int -> Handle -> t m (Array Word8)
178_toChunksWithBufferOf size h = go
180 -- XXX use cons/nil instead
181 go = mkStream $ \_ yld _ stp -> do
182 arr <- liftIO $ readArrayUpto size h
187-- | @toChunksWithBufferOf size handle@ reads a stream of arrays from the file
188-- handle @handle@. The maximum size of a single array is limited to @size@.
189-- The actual size read may be less than or equal to @size@.
192{-# INLINE_NORMAL toChunksWithBufferOf #-}
193toChunksWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
194toChunksWithBufferOf size h = D.fromStreamD (D.Stream step ())
196 {-# INLINE_LATE step #-}
198 arr <- liftIO $ readArrayUpto size h
204-- | Unfold the tuple @(bufsize, handle)@ into a stream of 'Word8' arrays.
205-- Read requests to the IO device are performed using a buffer of size
206-- @bufsize@. The size of an array in the resulting stream is always less than
207-- or equal to @bufsize@.
210{-# INLINE_NORMAL readChunksWithBufferOf #-}
211readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Handle) (Array Word8)
212readChunksWithBufferOf = Unfold step return
214 {-# INLINE_LATE step #-}
216 arr <- liftIO $ readArrayUpto size h
220 _ -> D.Yield arr (size, h)
222-- XXX read 'Array a' instead of Word8
224-- | @toChunks handle@ reads a stream of arrays from the specified file
225-- handle. The maximum size of a single array is limited to
226-- @defaultChunkSize@. The actual size read may be less than or equal to
227-- @defaultChunkSize@.
229-- > toChunks = toChunksWithBufferOf defaultChunkSize
232{-# INLINE toChunks #-}
233toChunks :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
234toChunks = toChunksWithBufferOf defaultChunkSize
236-- | Read a stream of chunks from standard input. The maximum size of a single
237-- chunk is limited to @defaultChunkSize@. The actual size read may be less
238-- than @defaultChunkSize@.
240-- > getChunks = toChunks stdin
244{-# INLINE getChunks #-}
245getChunks :: (IsStream t, MonadIO m) => t m (Array Word8)
246getChunks = toChunks stdin
248-- | Unfolds a handle into a stream of 'Word8' arrays. Requests to the IO
249-- device are performed using a buffer of size
250-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'. The
251-- size of arrays in the resulting stream are therefore less than or equal to
252-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
255{-# INLINE readChunks #-}
256readChunks :: MonadIO m => Unfold m Handle (Array Word8)
257readChunks = UF.supplyFirst readChunksWithBufferOf defaultChunkSize
259-------------------------------------------------------------------------------
260-- Read File to Stream
261-------------------------------------------------------------------------------
263-- TODO for concurrent streams implement readahead IO. We can send multiple
264-- read requests at the same time. For serial case we can use async IO. We can
265-- also control the read throughput in mbps or IOPS.
267-- | Unfolds the tuple @(bufsize, handle)@ into a byte stream, read requests
268-- to the IO device are performed using buffers of @bufsize@.
271{-# INLINE readWithBufferOf #-}
272readWithBufferOf :: MonadIO m => Unfold m (Int, Handle) Word8
273readWithBufferOf = UF.concat 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@.
279{-# INLINE toBytesWithBufferOf #-}
280toBytesWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
281toBytesWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h
284-- Generate a stream of elements of the given type from a file 'Handle'.
285-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
287-- | Unfolds a file handle into a byte stream. IO requests to the device are
288-- performed in sizes of
289-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
293read :: MonadIO m => Unfold m Handle Word8
294read = UF.supplyFirst readWithBufferOf defaultChunkSize
296-- | Generate a byte stream from a file 'Handle'.
299{-# INLINE toBytes #-}
300toBytes :: (IsStream t, MonadIO m) => Handle -> t m Word8
301toBytes = AS.concat . toChunks
303-------------------------------------------------------------------------------
305-------------------------------------------------------------------------------
307-------------------------------------------------------------------------------
309-------------------------------------------------------------------------------
311-- | Write an 'Array' to a file handle.
314{-# INLINABLE writeArray #-}
315writeArray :: Storable a => Handle -> Array a -> IO ()
316writeArray _ arr | A.length arr == 0 = return ()
317writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen
320 let p = unsafeForeignPtrToPtr aStart
323-------------------------------------------------------------------------------
324-- Stream of Arrays IO
325-------------------------------------------------------------------------------
327-------------------------------------------------------------------------------
329-------------------------------------------------------------------------------
331-- | Write a stream of arrays to a handle.
334{-# INLINE fromChunks #-}
335fromChunks :: (MonadIO m, Storable a)
336 => Handle -> SerialT m (Array a) -> m ()
337fromChunks h m = S.mapM_ (liftIO . writeArray h) m
339-- | Write a stream of chunks to standard output.
343{-# INLINE putChunks #-}
344putChunks :: (MonadIO m, Storable a) => SerialT m (Array a) -> m ()
345putChunks = fromChunks stdout
347-- XXX this is currently buffered with 32K buffer. We need to use a timeout
348-- based scheme instead.
350-- | Write a stream of strings to standard output.
354{-# INLINE putStrings #-}
355putStrings :: MonadIO m => SerialT m String -> m ()
356putStrings = fromBytes stdout . U.encodeUtf8 . S.concatUnfold UF.fromList
358-- | @fromChunksWithBufferOf bufsize handle stream@ writes a stream of arrays
359-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
360-- The chunk size is only a maximum and the actual writes could be smaller as
361-- we do not split the arrays to fit exactly to the specified size.
364{-# INLINE fromChunksWithBufferOf #-}
365fromChunksWithBufferOf :: (MonadIO m, Storable a)
366 => Int -> Handle -> SerialT m (Array a) -> m ()
367fromChunksWithBufferOf n h xs = fromChunks h $ AS.compact n xs
369-- | @fromBytesWithBufferOf bufsize handle stream@ writes @stream@ to @handle@
370-- in chunks of @bufsize@. A write is performed to the IO device as soon as we
371-- collect the required input size.
374{-# INLINE fromBytesWithBufferOf #-}
375fromBytesWithBufferOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
376fromBytesWithBufferOf n h m = fromChunks h $ S.arraysOf n m
377-- fromBytesWithBufferOf n h m = fromChunks h $ AS.arraysOf n m
379-- > write = 'writeWithBufferOf' A.defaultChunkSize
381-- | Write a byte stream to a file handle. Accumulates the input in chunks of
382-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing.
384-- NOTE: This may perform better than the 'write' fold, you can try this if you
385-- need some extra perf boost.
388{-# INLINE fromBytes #-}
389fromBytes :: MonadIO m => Handle -> SerialT m Word8 -> m ()
390fromBytes = fromBytesWithBufferOf defaultChunkSize
392-- | Write a stream of arrays to a handle. Each array in the stream is written
393-- to the device as a separate IO request.
396{-# INLINE writeChunks #-}
397writeChunks :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
398writeChunks h = FL.drainBy (liftIO . writeArray h)
400-- | @writeChunksWithBufferOf bufsize handle@ writes a stream of arrays
401-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
402-- We never split an array, if a single array is bigger than the specified size
403-- it emitted as it is. Multiple arrays are coalesed as long as the total size
404-- remains below the specified size.
407{-# INLINE writeChunksWithBufferOf #-}
408writeChunksWithBufferOf :: (MonadIO m, Storable a)
409 => Int -> Handle -> Fold m (Array a) ()
410writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)
412-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
415-- Note that if you use a chunk size less than 8K (GHC's default buffer
416-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
417-- do not want buffering to occur at GHC level as well. Same thing applies to
420-- | @writeWithBufferOf reqSize handle@ writes the input stream to @handle@.
421-- Bytes in the input stream are collected into a buffer until we have a chunk
422-- of @reqSize@ and then written to the IO device.
425{-# INLINE writeWithBufferOf #-}
426writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
427writeWithBufferOf n h = FL.lchunksOf n (writeNUnsafe n) (writeChunks h)
429-- > write = 'writeWithBufferOf' A.defaultChunkSize
431-- | Write a byte stream to a file handle. Accumulates the input in chunks of
432-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing
437write :: MonadIO m => Handle -> Fold m Word8 ()
438write = writeWithBufferOf defaultChunkSize
442write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
443write = toHandleWith A.defaultChunkSize
446-- XXX mmap a file into an array. This could be useful for in-place operations
447-- on a file. For example, we can quicksort the contents of a file by mmapping
450-------------------------------------------------------------------------------
451-- IO with encoding/decoding Unicode characters
452-------------------------------------------------------------------------------
456-- > readUtf8 = decodeUtf8 . read
458-- Read a UTF8 encoded stream of unicode characters from a file handle.
461{-# INLINE readUtf8 #-}
462readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
463readUtf8 = decodeUtf8 . read
466-- > writeUtf8 h s = write h $ encodeUtf8 s
468-- Encode a stream of unicode characters to UTF8 and write it to the given file
469-- handle. Default block buffering applies to the writes.
472{-# INLINE writeUtf8 #-}
473writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
474writeUtf8 h s = write h $ encodeUtf8 s
476-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
477-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
478-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
479-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
483{-# INLINE writeUtf8ByLines #-}
484writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
485writeUtf8ByLines = undefined
487-- | Read UTF-8 lines from a file handle and apply the specified fold to each
488-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
491{-# INLINE readLines #-}
492readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
493readLines h f = foldLines (readUtf8 h) f
495-------------------------------------------------------------------------------
496-- Framing on a sequence
497-------------------------------------------------------------------------------
499-- | Read a stream from a file handle and split it into frames delimited by
500-- the specified sequence of elements. The supplied fold is applied on each
504{-# INLINE readFrames #-}
505readFrames :: (IsStream t, MonadIO m, Storable a)
506 => Array a -> Handle -> Fold m a b -> t m b
507readFrames = undefined -- foldFrames . read
509-- | Write a stream to the given file handle buffering up to frames separated
510-- by the given sequence or up to a maximum of @defaultChunkSize@.
513{-# INLINE writeByFrames #-}
514writeByFrames :: (IsStream t, MonadIO m, Storable a)
515 => Array a -> Handle -> t m a -> m ()
516writeByFrames = undefined
518-------------------------------------------------------------------------------
520-------------------------------------------------------------------------------
522-- | Write collecting the input in sessions of n seconds or if chunkSize
524{-# INLINE writeByChunksOrSessionsOf #-}
525writeByChunksOrSessionsOf :: MonadIO m
526 => Int -> Double -> Handle -> SerialT m Word8 -> m ()
527writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined
529-- | Write collecting the input in sessions of n seconds or if defaultChunkSize
531{-# INLINE writeBySessionsOf #-}
532writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()
533writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n
535-------------------------------------------------------------------------------
536-- Random Access IO (Seek)
537-------------------------------------------------------------------------------
539-- XXX handles could be shared, so we may not want to use the handle state at
540-- all for these APIs. we can use pread and pwrite instead. On windows we will
541-- need to use readFile/writeFile with an offset argument.
543-------------------------------------------------------------------------------
545-- | Read the element at the given index treating the file as an array.
548{-# INLINE readIndex #-}
549readIndex :: Storable a => Handle -> Int -> Maybe a
550readIndex arr i = undefined
552-- NOTE: To represent a range to read we have chosen (start, size) instead of
553-- (start, end). This helps in removing the ambiguity of whether "end" is
554-- included in the range or not.
556-- We could avoid specifying the range to be read and instead use "take size"
557-- on the stream, but it may end up reading more and then consume it partially.
559-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
560-- from @handle@ starting at the offset @pos@ from the beginning of the file.
562-- Reads are performed in chunks of size @chunkSize@. For block devices, to
563-- avoid reading partial blocks @chunkSize@ must align with the block size of
564-- the underlying device. If the underlying block size is unknown, it is a good
565-- idea to keep it a multiple 4KiB. This API ensures that the start of each
566-- chunk is aligned with @chunkSize@ from second chunk onwards.
568{-# INLINE readSliceWith #-}
569readSliceWith :: (IsStream t, MonadIO m, Storable a)
570 => Int -> Handle -> Int -> Int -> t m a
571readSliceWith chunkSize h pos len = undefined
573-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
574-- at index @i@ and reading up to @count@ elements in the forward direction
575-- ending at the index @i + count - 1@.
578{-# INLINE readSlice #-}
579readSlice :: (IsStream t, MonadIO m, Storable a)
580 => Handle -> Int -> Int -> t m a
581readSlice = readSliceWith A.defaultChunkSize
583-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
584-- at index @i@ and reading up to @count@ elements in the reverse direction
585-- ending at the index @i - count + 1@.
588{-# INLINE readSliceRev #-}
589readSliceRev :: (IsStream t, MonadIO m, Storable a)
590 => Handle -> Int -> Int -> t m a
591readSliceRev h i count = undefined
593-- | Write the given element at the given index in the file.
596{-# INLINE writeIndex #-}
597writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
598writeIndex h i a = undefined
600-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
601-- starting at index @i@ and writing up to @count@ elements in the forward
602-- direction ending at the index @i + count - 1@.
605{-# INLINE writeSlice #-}
606writeSlice :: (IsStream t, Monad m, Storable a)
607 => Handle -> Int -> Int -> t m a -> m ()
608writeSlice h i len s = undefined
610-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
611-- starting at index @i@ and writing up to @count@ elements in the reverse
612-- direction ending at the index @i - count + 1@.
615{-# INLINE writeSliceRev #-}
616writeSliceRev :: (IsStream t, Monad m, Storable a)
617 => Handle -> Int -> Int -> t m a -> m ()
618writeSliceRev 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