1{-# OPTIONS_HADDOCK hide #-}
3{-# LANGUAGE BangPatterns #-}
4{-# LANGUAGE FlexibleContexts #-}
5{-# LANGUAGE MagicHash #-}
6{-# LANGUAGE RecordWildCards #-}
7{-# LANGUAGE UnboxedTuples #-}
12-- Module : Streamly.Internal.FileSystem.Handle
13-- Copyright : (c) 2018 Composewell Technologies
16-- Maintainer : streamly@composewell.com
17-- Stability : experimental
20module Streamly.Internal.FileSystem.Handle
22 -- ** Read from Handle
37 , readChunksWithBufferOf
39 , toChunksWithBufferOf
44 -- Byte stream write (Folds)
52 -- Byte stream write (Streams)
54 , fromBytesWithBufferOf
59 , writeChunksWithBufferOf
61 -- -- * Array stream Write
62 , fromChunksWithBufferOf
68 -- -- * Random Access (Seek)
69 -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or
70 -- files that have random access or seek capability. This type of devices
71 -- include disks, files, memory devices and exclude terminals, pipes,
74 -- XXX need to decide whether to use readFromStepN style or readFromThenTo
75 -- style. The latter is consistent with list enumeration. The former may be
76 -- more unambiguous and it is easier and clearer to specify 0 elements.
78 -- We can also generate the request pattern using a funciton.
81 -- , readFrom -- read from a given position to the end of file
82 -- , readFromRev -- read from a given position the beginning of file
83 -- , readTo -- read from beginning up to the given position
84 -- , readToRev -- read from end to the given position in file
90 -- , readChunksFromToWithBufferOf
91 -- , readChunksFromThenToWithBufferOf
94 -- , writeFrom -- start writing at the given position
96 -- , writeTo -- write from beginning up to the given position
102 -- , writeChunksFromTo
103 -- , writeChunksFromToWithBufferOf
104 -- , writeChunksFromThenToWithBufferOf
108import Control.Monad.IO.Class (MonadIO(..))
109import Data.Word (Word8)
110import Foreign.ForeignPtr (withForeignPtr)
111import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
112import Foreign.Ptr (minusPtr, plusPtr)
113import Foreign.Storable (Storable(..))
114import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
115import System.IO (Handle, hGetBufSome, hPutBuf, stdin, stdout)
116import Prelude hiding (read)
118import Streamly (MonadAsync)
119import Streamly.Data.Fold (Fold)
120import Streamly.Internal.Data.Unfold.Types (Unfold(..))
121import Streamly.Internal.Memory.Array.Types
122 (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,
124import Streamly.Streams.Serial (SerialT)
125import Streamly.Streams.StreamK.Type (IsStream, mkStream)
126-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
128import qualified Streamly.Data.Fold as FL
129import qualified Streamly.Internal.Data.Fold.Types as FL
130import qualified Streamly.Internal.Data.Unicode.Stream as U
131import qualified Streamly.Internal.Data.Unfold as UF
132import qualified Streamly.Internal.Memory.Array as IA
133import qualified Streamly.Internal.Memory.ArrayStream as AS
134import qualified Streamly.Internal.Prelude as S
135import qualified Streamly.Memory.Array as A
136import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
138-------------------------------------------------------------------------------
140-------------------------------------------------------------------------------
142-- The following references may be useful to build an understanding about the
145-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
146-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
147-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
148-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
149-- https://www.w3.org/TR/FileAPI/ for http file API.
151-------------------------------------------------------------------------------
153-------------------------------------------------------------------------------
155-- | Read a 'ByteArray' from a file handle. If no data is available on the
156-- handle it blocks until some data becomes available. If data is available
157-- then it immediately returns that data without blocking. It reads a maximum
158-- of up to the size requested.
159{-# INLINABLE readArrayUpto #-}
160readArrayUpto :: Int -> Handle -> IO (Array Word8)
161readArrayUpto size h = do
162 ptr <- mallocPlainForeignPtrBytes size
163 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
164 withForeignPtr ptr $ \p -> do
165 n <- hGetBufSome h p size
168 , aEnd = p `plusPtr` n
169 , aBound = p `plusPtr` size
171 -- XXX shrink only if the diff is significant
174-------------------------------------------------------------------------------
175-- Stream of Arrays IO
176-------------------------------------------------------------------------------
178-- | @toChunksWithBufferOf size h@ reads a stream of arrays from file handle @h@.
179-- The maximum size of a single array is specified by @size@. The actual size
180-- read may be less than or equal to @size@.
181{-# INLINABLE _toChunksWithBufferOf #-}
182_toChunksWithBufferOf :: (IsStream t, MonadIO m)
183 => Int -> Handle -> t m (Array Word8)
184_toChunksWithBufferOf size h = go
186 -- XXX use cons/nil instead
187 go = mkStream $ \_ yld _ stp -> do
188 arr <- liftIO $ readArrayUpto size h
193-- | @toChunksWithBufferOf size handle@ reads a stream of arrays from the file
194-- handle @handle@. The maximum size of a single array is limited to @size@.
195-- The actual size read may be less than or equal to @size@.
198{-# INLINE_NORMAL toChunksWithBufferOf #-}
199toChunksWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
200toChunksWithBufferOf size h = D.fromStreamD (D.Stream step ())
202 {-# INLINE_LATE step #-}
204 arr <- liftIO $ readArrayUpto size h
210-- | Unfold the tuple @(bufsize, handle)@ into a stream of 'Word8' arrays.
211-- Read requests to the IO device are performed using a buffer of size
212-- @bufsize@. The size of an array in the resulting stream is always less than
213-- or equal to @bufsize@.
216{-# INLINE_NORMAL readChunksWithBufferOf #-}
217readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Handle) (Array Word8)
218readChunksWithBufferOf = Unfold step return
220 {-# INLINE_LATE step #-}
222 arr <- liftIO $ readArrayUpto size h
226 _ -> D.Yield arr (size, h)
228-- XXX read 'Array a' instead of Word8
230-- | @toChunks handle@ reads a stream of arrays from the specified file
231-- handle. The maximum size of a single array is limited to
232-- @defaultChunkSize@. The actual size read may be less than or equal to
233-- @defaultChunkSize@.
235-- > toChunks = toChunksWithBufferOf defaultChunkSize
238{-# INLINE toChunks #-}
239toChunks :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
240toChunks = toChunksWithBufferOf defaultChunkSize
242-- | Read a stream of chunks from standard input. The maximum size of a single
243-- chunk is limited to @defaultChunkSize@. The actual size read may be less
244-- than @defaultChunkSize@.
246-- > getChunks = toChunks stdin
250{-# INLINE getChunks #-}
251getChunks :: (IsStream t, MonadIO m) => t m (Array Word8)
252getChunks = toChunks stdin
254-- | Read a stream of bytes from standard input.
256-- > getBytes = toBytes stdin
260{-# INLINE getBytes #-}
261getBytes :: (IsStream t, MonadIO m) => t m Word8
262getBytes = toBytes stdin
264-- | Unfolds a handle into a stream of 'Word8' arrays. Requests to the IO
265-- device are performed using a buffer of size
266-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'. The
267-- size of arrays in the resulting stream are therefore less than or equal to
268-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
271{-# INLINE readChunks #-}
272readChunks :: MonadIO m => Unfold m Handle (Array Word8)
273readChunks = UF.supplyFirst readChunksWithBufferOf defaultChunkSize
275-------------------------------------------------------------------------------
276-- Read File to Stream
277-------------------------------------------------------------------------------
279-- TODO for concurrent streams implement readahead IO. We can send multiple
280-- read requests at the same time. For serial case we can use async IO. We can
281-- also control the read throughput in mbps or IOPS.
283-- | Unfolds the tuple @(bufsize, handle)@ into a byte stream, read requests
284-- to the IO device are performed using buffers of @bufsize@.
287{-# INLINE readWithBufferOf #-}
288readWithBufferOf :: MonadIO m => Unfold m (Int, Handle) Word8
289readWithBufferOf = UF.concat readChunksWithBufferOf A.read
291-- | @toBytesWithBufferOf bufsize handle@ reads a byte stream from a file
292-- handle, reads are performed in chunks of up to @bufsize@.
295{-# INLINE toBytesWithBufferOf #-}
296toBytesWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
297toBytesWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h
300-- Generate a stream of elements of the given type from a file 'Handle'.
301-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
303-- | Unfolds a file handle into a byte stream. IO requests to the device are
304-- performed in sizes of
305-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
309read :: MonadIO m => Unfold m Handle Word8
310read = UF.supplyFirst readWithBufferOf defaultChunkSize
312-- | Generate a byte stream from a file 'Handle'.
315{-# INLINE toBytes #-}
316toBytes :: (IsStream t, MonadIO m) => Handle -> t m Word8
317toBytes = AS.concat . toChunks
319-------------------------------------------------------------------------------
321-------------------------------------------------------------------------------
323-------------------------------------------------------------------------------
325-------------------------------------------------------------------------------
327-- | Write an 'Array' to a file handle.
330{-# INLINABLE writeArray #-}
331writeArray :: Storable a => Handle -> Array a -> IO ()
332writeArray _ arr | A.length arr == 0 = return ()
333writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen
336 let p = unsafeForeignPtrToPtr aStart
339-------------------------------------------------------------------------------
340-- Stream of Arrays IO
341-------------------------------------------------------------------------------
343-------------------------------------------------------------------------------
345-------------------------------------------------------------------------------
347-- | Write a stream of arrays to a handle.
350{-# INLINE fromChunks #-}
351fromChunks :: (MonadIO m, Storable a)
352 => Handle -> SerialT m (Array a) -> m ()
353fromChunks h m = S.mapM_ (liftIO . writeArray h) m
355-- | Write a stream of chunks to standard output.
359{-# INLINE putChunks #-}
360putChunks :: (MonadIO m, Storable a) => SerialT m (Array a) -> m ()
361putChunks = fromChunks stdout
363-- | Write a stream of strings to standard output using Latin1 encoding.
367{-# INLINE putStrings #-}
368putStrings :: MonadAsync m => SerialT m String -> m ()
369putStrings = putChunks . S.mapM (IA.fromStream . U.encodeLatin1 . S.fromList)
371-- | @fromChunksWithBufferOf bufsize handle stream@ writes a stream of arrays
372-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
373-- The chunk size is only a maximum and the actual writes could be smaller as
374-- we do not split the arrays to fit exactly to the specified size.
377{-# INLINE fromChunksWithBufferOf #-}
378fromChunksWithBufferOf :: (MonadIO m, Storable a)
379 => Int -> Handle -> SerialT m (Array a) -> m ()
380fromChunksWithBufferOf n h xs = fromChunks h $ AS.compact n xs
382-- | @fromBytesWithBufferOf bufsize handle stream@ writes @stream@ to @handle@
383-- in chunks of @bufsize@. A write is performed to the IO device as soon as we
384-- collect the required input size.
387{-# INLINE fromBytesWithBufferOf #-}
388fromBytesWithBufferOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
389fromBytesWithBufferOf n h m = fromChunks h $ S.arraysOf n m
390-- fromBytesWithBufferOf n h m = fromChunks h $ AS.arraysOf n m
392-- > write = 'writeWithBufferOf' 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.
397-- NOTE: This may perform better than the 'write' fold, you can try this if you
398-- need some extra perf boost.
401{-# INLINE fromBytes #-}
402fromBytes :: MonadIO m => Handle -> SerialT m Word8 -> m ()
403fromBytes = fromBytesWithBufferOf defaultChunkSize
405-- | Write a stream of arrays to a handle. Each array in the stream is written
406-- to the device as a separate IO request.
409{-# INLINE writeChunks #-}
410writeChunks :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
411writeChunks h = FL.drainBy (liftIO . writeArray h)
413-- | @writeChunksWithBufferOf bufsize handle@ writes a stream of arrays
414-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
415-- We never split an array, if a single array is bigger than the specified size
416-- it emitted as it is. Multiple arrays are coalesed as long as the total size
417-- remains below the specified size.
420{-# INLINE writeChunksWithBufferOf #-}
421writeChunksWithBufferOf :: (MonadIO m, Storable a)
422 => Int -> Handle -> Fold m (Array a) ()
423writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)
425-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
428-- Note that if you use a chunk size less than 8K (GHC's default buffer
429-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
430-- do not want buffering to occur at GHC level as well. Same thing applies to
433-- | @writeWithBufferOf reqSize handle@ writes the input stream to @handle@.
434-- Bytes in the input stream are collected into a buffer until we have a chunk
435-- of @reqSize@ and then written to the IO device.
438{-# INLINE writeWithBufferOf #-}
439writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
440writeWithBufferOf n h = FL.lchunksOf n (writeNUnsafe n) (writeChunks h)
442-- > write = 'writeWithBufferOf' A.defaultChunkSize
444-- | Write a byte stream to a file handle. Accumulates the input in chunks of
445-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing
450write :: MonadIO m => Handle -> Fold m Word8 ()
451write = writeWithBufferOf defaultChunkSize
455write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
456write = toHandleWith A.defaultChunkSize
459-- XXX mmap a file into an array. This could be useful for in-place operations
460-- on a file. For example, we can quicksort the contents of a file by mmapping
463-------------------------------------------------------------------------------
464-- IO with encoding/decoding Unicode characters
465-------------------------------------------------------------------------------
469-- > readUtf8 = decodeUtf8 . read
471-- Read a UTF8 encoded stream of unicode characters from a file handle.
474{-# INLINE readUtf8 #-}
475readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
476readUtf8 = decodeUtf8 . read
479-- > writeUtf8 h s = write h $ encodeUtf8 s
481-- Encode a stream of unicode characters to UTF8 and write it to the given file
482-- handle. Default block buffering applies to the writes.
485{-# INLINE writeUtf8 #-}
486writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
487writeUtf8 h s = write h $ encodeUtf8 s
489-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
490-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
491-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
492-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
496{-# INLINE writeUtf8ByLines #-}
497writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
498writeUtf8ByLines = undefined
500-- | Read UTF-8 lines from a file handle and apply the specified fold to each
501-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
504{-# INLINE readLines #-}
505readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
506readLines h f = foldLines (readUtf8 h) f
508-------------------------------------------------------------------------------
509-- Framing on a sequence
510-------------------------------------------------------------------------------
512-- | Read a stream from a file handle and split it into frames delimited by
513-- the specified sequence of elements. The supplied fold is applied on each
517{-# INLINE readFrames #-}
518readFrames :: (IsStream t, MonadIO m, Storable a)
519 => Array a -> Handle -> Fold m a b -> t m b
520readFrames = undefined -- foldFrames . read
522-- | Write a stream to the given file handle buffering up to frames separated
523-- by the given sequence or up to a maximum of @defaultChunkSize@.
526{-# INLINE writeByFrames #-}
527writeByFrames :: (IsStream t, MonadIO m, Storable a)
528 => Array a -> Handle -> t m a -> m ()
529writeByFrames = undefined
531-------------------------------------------------------------------------------
533-------------------------------------------------------------------------------
535-- | Write collecting the input in sessions of n seconds or if chunkSize
537{-# INLINE writeByChunksOrSessionsOf #-}
538writeByChunksOrSessionsOf :: MonadIO m
539 => Int -> Double -> Handle -> SerialT m Word8 -> m ()
540writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined
542-- | Write collecting the input in sessions of n seconds or if defaultChunkSize
544{-# INLINE writeBySessionsOf #-}
545writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()
546writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n
548-------------------------------------------------------------------------------
549-- Random Access IO (Seek)
550-------------------------------------------------------------------------------
552-- XXX handles could be shared, so we may not want to use the handle state at
553-- all for these APIs. we can use pread and pwrite instead. On windows we will
554-- need to use readFile/writeFile with an offset argument.
556-------------------------------------------------------------------------------
558-- | Read the element at the given index treating the file as an array.
561{-# INLINE readIndex #-}
562readIndex :: Storable a => Handle -> Int -> Maybe a
563readIndex arr i = undefined
565-- NOTE: To represent a range to read we have chosen (start, size) instead of
566-- (start, end). This helps in removing the ambiguity of whether "end" is
567-- included in the range or not.
569-- We could avoid specifying the range to be read and instead use "take size"
570-- on the stream, but it may end up reading more and then consume it partially.
572-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
573-- from @handle@ starting at the offset @pos@ from the beginning of the file.
575-- Reads are performed in chunks of size @chunkSize@. For block devices, to
576-- avoid reading partial blocks @chunkSize@ must align with the block size of
577-- the underlying device. If the underlying block size is unknown, it is a good
578-- idea to keep it a multiple 4KiB. This API ensures that the start of each
579-- chunk is aligned with @chunkSize@ from second chunk onwards.
581{-# INLINE readSliceWith #-}
582readSliceWith :: (IsStream t, MonadIO m, Storable a)
583 => Int -> Handle -> Int -> Int -> t m a
584readSliceWith chunkSize h pos len = undefined
586-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
587-- at index @i@ and reading up to @count@ elements in the forward direction
588-- ending at the index @i + count - 1@.
591{-# INLINE readSlice #-}
592readSlice :: (IsStream t, MonadIO m, Storable a)
593 => Handle -> Int -> Int -> t m a
594readSlice = readSliceWith A.defaultChunkSize
596-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
597-- at index @i@ and reading up to @count@ elements in the reverse direction
598-- ending at the index @i - count + 1@.
601{-# INLINE readSliceRev #-}
602readSliceRev :: (IsStream t, MonadIO m, Storable a)
603 => Handle -> Int -> Int -> t m a
604readSliceRev h i count = undefined
606-- | Write the given element at the given index in the file.
609{-# INLINE writeIndex #-}
610writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
611writeIndex h i a = undefined
613-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
614-- starting at index @i@ and writing up to @count@ elements in the forward
615-- direction ending at the index @i + count - 1@.
618{-# INLINE writeSlice #-}
619writeSlice :: (IsStream t, Monad m, Storable a)
620 => Handle -> Int -> Int -> t m a -> m ()
621writeSlice h i len s = undefined
623-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
624-- starting at index @i@ and writing up to @count@ elements in the reverse
625-- direction ending at the index @i - count + 1@.
628{-# INLINE writeSliceRev #-}
629writeSliceRev :: (IsStream t, Monad m, Storable a)
630 => Handle -> Int -> Int -> t m a -> m ()
631writeSliceRev 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