2{-# LANGUAGE FlexibleContexts #-}
3{-# LANGUAGE RecordWildCards #-}
8-- Module : Streamly.Internal.FileSystem.Handle
9-- Copyright : (c) 2018 Composewell Technologies
12-- Maintainer : streamly@composewell.com
13-- Stability : experimental
16module Streamly.Internal.FileSystem.Handle
18 -- ** Read from Handle
33 , readChunksWithBufferOf
35 , toChunksWithBufferOf
40 -- Byte stream write (Folds)
49 -- Byte stream write (Streams)
51 , fromBytesWithBufferOf
56 , writeChunksWithBufferOf
58 -- -- * Array stream Write
59 , fromChunksWithBufferOf
66 -- -- * Random Access (Seek)
67 -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or
68 -- files that have random access or seek capability. This type of devices
69 -- include disks, files, memory devices and exclude terminals, pipes,
72 -- XXX need to decide whether to use readFromStepN style or readFromThenTo
73 -- style. The latter is consistent with list enumeration. The former may be
74 -- more unambiguous and it is easier and clearer to specify 0 elements.
76 -- We can also generate the request pattern using a funciton.
79 -- , readFrom -- read from a given position to the end of file
80 -- , readFromRev -- read from a given position the beginning of file
81 -- , readTo -- read from beginning up to the given position
82 -- , readToRev -- read from end to the given position in file
88 -- , readChunksFromToWithBufferOf
89 -- , readChunksFromThenToWithBufferOf
92 -- , writeFrom -- start writing at the given position
94 -- , writeTo -- write from beginning up to the given position
100 -- , writeChunksFromTo
101 -- , writeChunksFromToWithBufferOf
102 -- , writeChunksFromThenToWithBufferOf
106import Control.Monad.IO.Class (MonadIO(..))
107import Data.Word (Word8)
108import Foreign.ForeignPtr (withForeignPtr)
109import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
110import Foreign.Ptr (minusPtr, plusPtr)
111import Foreign.Storable (Storable(..))
112import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
113import System.IO (Handle, hGetBufSome, hPutBuf, stdin, stdout)
114import Prelude hiding (read)
116import Streamly (MonadAsync)
117import Streamly.Data.Fold (Fold)
118import Streamly.Internal.Data.Fold.Types (Fold2(..))
119import Streamly.Internal.Data.Unfold.Types (Unfold(..))
120import Streamly.Internal.Memory.Array.Types
121 (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,
123import Streamly.Internal.Data.Stream.Serial (SerialT)
124import Streamly.Internal.Data.Stream.StreamK.Type (IsStream, mkStream)
125-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
127import qualified Streamly.Data.Fold as FL
128import qualified Streamly.Internal.Data.Fold.Types as FL
129import qualified Streamly.Internal.Data.Unfold as UF
130import qualified Streamly.Internal.Memory.Array as IA
131import qualified Streamly.Internal.Memory.ArrayStream as AS
132import qualified Streamly.Internal.Prelude as S
133import qualified Streamly.Memory.Array as A
134import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
136-------------------------------------------------------------------------------
138-------------------------------------------------------------------------------
140-- The following references may be useful to build an understanding about the
143-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
144-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
145-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
146-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
147-- https://www.w3.org/TR/FileAPI/ for http file API.
149-------------------------------------------------------------------------------
151-------------------------------------------------------------------------------
153-- | Read a 'ByteArray' from a file handle. If no data is available on the
154-- handle it blocks until some data becomes available. If data is available
155-- then it immediately returns that data without blocking. It reads a maximum
156-- of up to the size requested.
157{-# INLINABLE readArrayUpto #-}
158readArrayUpto :: Int -> Handle -> IO (Array Word8)
159readArrayUpto size h = do
160 ptr <- mallocPlainForeignPtrBytes size
161 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
162 withForeignPtr ptr $ \p -> do
163 n <- hGetBufSome h p size
166 , aEnd = p `plusPtr` n
167 , aBound = p `plusPtr` size
169 -- XXX shrink only if the diff is significant
172-------------------------------------------------------------------------------
173-- Stream of Arrays IO
174-------------------------------------------------------------------------------
176-- | @toChunksWithBufferOf size h@ reads a stream of arrays from file handle @h@.
177-- The maximum size of a single array is specified by @size@. The actual size
178-- read may be less than or equal to @size@.
179{-# INLINABLE _toChunksWithBufferOf #-}
180_toChunksWithBufferOf :: (IsStream t, MonadIO m)
181 => Int -> Handle -> t m (Array Word8)
182_toChunksWithBufferOf size h = go
184 -- XXX use cons/nil instead
185 go = mkStream $ \_ yld _ stp -> do
186 arr <- liftIO $ readArrayUpto size h
191-- | @toChunksWithBufferOf size handle@ reads a stream of arrays from the file
192-- handle @handle@. The maximum size of a single array is limited to @size@.
193-- The actual size read may be less than or equal to @size@.
196{-# INLINE_NORMAL toChunksWithBufferOf #-}
197toChunksWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
198toChunksWithBufferOf size h = D.fromStreamD (D.Stream step ())
200 {-# INLINE_LATE step #-}
202 arr <- liftIO $ readArrayUpto size h
208-- | Unfold the tuple @(bufsize, handle)@ into a stream of 'Word8' arrays.
209-- Read requests to the IO device are performed using a buffer of size
210-- @bufsize@. The size of an array in the resulting stream is always less than
211-- or equal to @bufsize@.
214{-# INLINE_NORMAL readChunksWithBufferOf #-}
215readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Handle) (Array Word8)
216readChunksWithBufferOf = Unfold step return
218 {-# INLINE_LATE step #-}
220 arr <- liftIO $ readArrayUpto size h
224 _ -> D.Yield arr (size, h)
226-- XXX read 'Array a' instead of Word8
228-- | @toChunks handle@ reads a stream of arrays from the specified file
229-- handle. The maximum size of a single array is limited to
230-- @defaultChunkSize@. The actual size read may be less than or equal to
231-- @defaultChunkSize@.
233-- > toChunks = toChunksWithBufferOf defaultChunkSize
236{-# INLINE toChunks #-}
237toChunks :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
238toChunks = toChunksWithBufferOf defaultChunkSize
240-- | Read a stream of chunks from standard input. The maximum size of a single
241-- chunk is limited to @defaultChunkSize@. The actual size read may be less
242-- than @defaultChunkSize@.
244-- > getChunks = toChunks stdin
248{-# INLINE getChunks #-}
249getChunks :: (IsStream t, MonadIO m) => t m (Array Word8)
250getChunks = toChunks stdin
252-- | Read a stream of bytes from standard input.
254-- > getBytes = toBytes stdin
258{-# INLINE getBytes #-}
259getBytes :: (IsStream t, MonadIO m) => t m Word8
260getBytes = toBytes stdin
262-- | Unfolds a handle into a stream of 'Word8' arrays. Requests to the IO
263-- device are performed using a buffer of size
264-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'. The
265-- size of arrays in the resulting stream are therefore less than or equal to
266-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
269{-# INLINE readChunks #-}
270readChunks :: MonadIO m => Unfold m Handle (Array Word8)
271readChunks = UF.supplyFirst readChunksWithBufferOf defaultChunkSize
273-------------------------------------------------------------------------------
274-- Read File to Stream
275-------------------------------------------------------------------------------
277-- TODO for concurrent streams implement readahead IO. We can send multiple
278-- read requests at the same time. For serial case we can use async IO. We can
279-- also control the read throughput in mbps or IOPS.
281-- | Unfolds the tuple @(bufsize, handle)@ into a byte stream, read requests
282-- to the IO device are performed using buffers of @bufsize@.
285{-# INLINE readWithBufferOf #-}
286readWithBufferOf :: MonadIO m => Unfold m (Int, Handle) Word8
287readWithBufferOf = UF.concat readChunksWithBufferOf A.read
289-- | @toBytesWithBufferOf bufsize handle@ reads a byte stream from a file
290-- handle, reads are performed in chunks of up to @bufsize@.
293{-# INLINE toBytesWithBufferOf #-}
294toBytesWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
295toBytesWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h
298-- Generate a stream of elements of the given type from a file 'Handle'.
299-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
301-- | Unfolds a file handle into a byte stream. IO requests to the device are
302-- performed in sizes of
303-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
307read :: MonadIO m => Unfold m Handle Word8
308read = UF.supplyFirst readWithBufferOf defaultChunkSize
310-- | Generate a byte stream from a file 'Handle'.
313{-# INLINE toBytes #-}
314toBytes :: (IsStream t, MonadIO m) => Handle -> t m Word8
315toBytes = AS.concat . toChunks
317-------------------------------------------------------------------------------
319-------------------------------------------------------------------------------
321-------------------------------------------------------------------------------
323-------------------------------------------------------------------------------
325-- | Write an 'Array' to a file handle.
328{-# INLINABLE writeArray #-}
329writeArray :: Storable a => Handle -> Array a -> IO ()
330writeArray _ arr | A.length arr == 0 = return ()
331writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen
334 let p = unsafeForeignPtrToPtr aStart
337-------------------------------------------------------------------------------
338-- Stream of Arrays IO
339-------------------------------------------------------------------------------
341-------------------------------------------------------------------------------
343-------------------------------------------------------------------------------
345-- XXX use an unfold to fromObjects or fromUnfold so that we can put any object
346-- | Write a stream of arrays to a handle.
349{-# INLINE fromChunks #-}
350fromChunks :: (MonadIO m, Storable a)
351 => Handle -> SerialT m (Array a) -> m ()
352fromChunks h = S.mapM_ (liftIO . writeArray h)
354-- | Write a stream of chunks to standard output.
358{-# INLINE putChunks #-}
359putChunks :: (MonadIO m, Storable a) => SerialT m (Array a) -> m ()
360putChunks = fromChunks stdout
362-- XXX use an unfold so that we can put any type of strings.
363-- | Write a stream of strings to standard output using the supplied encoding.
364-- Output is flushed to the device for each string.
368{-# INLINE putStrings #-}
369putStrings :: MonadAsync m
370 => (SerialT m Char -> SerialT m Word8) -> SerialT m String -> m ()
371putStrings encode = putChunks . S.mapM (IA.fromStream . encode . S.fromList)
373-- XXX use an unfold so that we can put lines from any object
374-- | Write a stream of strings as separate lines to standard output using the
375-- supplied encoding. Output is line buffered i.e. the output is written to the
376-- device as soon as a newline is encountered.
380{-# INLINE putLines #-}
381putLines :: MonadAsync m
382 => (SerialT m Char -> SerialT m Word8) -> SerialT m String -> m ()
383putLines encode = putChunks . S.mapM
384 (\xs -> IA.fromStream $ encode (S.fromList (xs ++ "\n")))
386-- | Write a stream of bytes from standard output.
388-- > putBytes = fromBytes stdout
392{-# INLINE putBytes #-}
393putBytes :: MonadIO m => SerialT m Word8 -> m ()
394putBytes = fromBytes stdout
396-- | @fromChunksWithBufferOf bufsize handle stream@ writes a stream of arrays
397-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
398-- The chunk size is only a maximum and the actual writes could be smaller as
399-- we do not split the arrays to fit exactly to the specified size.
402{-# INLINE fromChunksWithBufferOf #-}
403fromChunksWithBufferOf :: (MonadIO m, Storable a)
404 => Int -> Handle -> SerialT m (Array a) -> m ()
405fromChunksWithBufferOf n h xs = fromChunks h $ AS.compact n xs
407-- | @fromBytesWithBufferOf bufsize handle stream@ writes @stream@ to @handle@
408-- in chunks of @bufsize@. A write is performed to the IO device as soon as we
409-- collect the required input size.
412{-# INLINE fromBytesWithBufferOf #-}
413fromBytesWithBufferOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
414fromBytesWithBufferOf n h m = fromChunks h $ S.arraysOf n m
415-- fromBytesWithBufferOf n h m = fromChunks h $ AS.arraysOf n m
417-- > write = 'writeWithBufferOf' A.defaultChunkSize
419-- | Write a byte stream to a file handle. Accumulates the input in chunks of
420-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing.
422-- NOTE: This may perform better than the 'write' fold, you can try this if you
423-- need some extra perf boost.
426{-# INLINE fromBytes #-}
427fromBytes :: MonadIO m => Handle -> SerialT m Word8 -> m ()
428fromBytes = fromBytesWithBufferOf defaultChunkSize
430-- | Write a stream of arrays to a handle. Each array in the stream is written
431-- to the device as a separate IO request.
434{-# INLINE writeChunks #-}
435writeChunks :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
436writeChunks h = FL.drainBy (liftIO . writeArray h)
438{-# INLINE writeChunks2 #-}
439writeChunks2 :: (MonadIO m, Storable a) => Fold2 m Handle (Array a) ()
440writeChunks2 = Fold2 (\h arr -> liftIO $ writeArray h arr >> return h) return (\_ -> return ())
442-- | @writeChunksWithBufferOf bufsize handle@ writes a stream of arrays
443-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
444-- We never split an array, if a single array is bigger than the specified size
445-- it emitted as it is. Multiple arrays are coalesed as long as the total size
446-- remains below the specified size.
449{-# INLINE writeChunksWithBufferOf #-}
450writeChunksWithBufferOf :: (MonadIO m, Storable a)
451 => Int -> Handle -> Fold m (Array a) ()
452writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)
454-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
457-- Note that if you use a chunk size less than 8K (GHC's default buffer
458-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
459-- do not want buffering to occur at GHC level as well. Same thing applies to
462-- | @writeWithBufferOf reqSize handle@ writes the input stream to @handle@.
463-- Bytes in the input stream are collected into a buffer until we have a chunk
464-- of @reqSize@ and then written to the IO device.
467{-# INLINE writeWithBufferOf #-}
468writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
469writeWithBufferOf n h = FL.lchunksOf n (writeNUnsafe n) (writeChunks h)
471{-# INLINE writeWithBufferOf2 #-}
472writeWithBufferOf2 :: MonadIO m => Int -> Fold2 m Handle Word8 ()
473writeWithBufferOf2 n = FL.lchunksOf2 n (writeNUnsafe n) writeChunks2
475-- > write = 'writeWithBufferOf' A.defaultChunkSize
477-- | Write a byte stream to a file handle. Accumulates the input in chunks of
478-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing
483write :: MonadIO m => Handle -> Fold m Word8 ()
484write = writeWithBufferOf defaultChunkSize
487write2 :: MonadIO m => Fold2 m Handle Word8 ()
488write2 = writeWithBufferOf2 defaultChunkSize
492write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
493write = toHandleWith A.defaultChunkSize
496-- XXX mmap a file into an array. This could be useful for in-place operations
497-- on a file. For example, we can quicksort the contents of a file by mmapping
500-------------------------------------------------------------------------------
501-- IO with encoding/decoding Unicode characters
502-------------------------------------------------------------------------------
506-- > readUtf8 = decodeUtf8 . read
508-- Read a UTF8 encoded stream of unicode characters from a file handle.
511{-# INLINE readUtf8 #-}
512readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
513readUtf8 = decodeUtf8 . read
516-- > writeUtf8 h s = write h $ encodeUtf8 s
518-- Encode a stream of unicode characters to UTF8 and write it to the given file
519-- handle. Default block buffering applies to the writes.
522{-# INLINE writeUtf8 #-}
523writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
524writeUtf8 h s = write h $ encodeUtf8 s
526-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
527-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
528-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
529-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
533{-# INLINE writeUtf8ByLines #-}
534writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
535writeUtf8ByLines = undefined
537-- | Read UTF-8 lines from a file handle and apply the specified fold to each
538-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
541{-# INLINE readLines #-}
542readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
543readLines h f = foldLines (readUtf8 h) f
545-------------------------------------------------------------------------------
546-- Framing on a sequence
547-------------------------------------------------------------------------------
549-- | Read a stream from a file handle and split it into frames delimited by
550-- the specified sequence of elements. The supplied fold is applied on each
554{-# INLINE readFrames #-}
555readFrames :: (IsStream t, MonadIO m, Storable a)
556 => Array a -> Handle -> Fold m a b -> t m b
557readFrames = undefined -- foldFrames . read
559-- | Write a stream to the given file handle buffering up to frames separated
560-- by the given sequence or up to a maximum of @defaultChunkSize@.
563{-# INLINE writeByFrames #-}
564writeByFrames :: (IsStream t, MonadIO m, Storable a)
565 => Array a -> Handle -> t m a -> m ()
566writeByFrames = undefined
568-------------------------------------------------------------------------------
570-------------------------------------------------------------------------------
572-- | Write collecting the input in sessions of n seconds or if chunkSize
574{-# INLINE writeByChunksOrSessionsOf #-}
575writeByChunksOrSessionsOf :: MonadIO m
576 => Int -> Double -> Handle -> SerialT m Word8 -> m ()
577writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined
579-- | Write collecting the input in sessions of n seconds or if defaultChunkSize
581{-# INLINE writeBySessionsOf #-}
582writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()
583writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n
585-------------------------------------------------------------------------------
586-- Random Access IO (Seek)
587-------------------------------------------------------------------------------
589-- XXX handles could be shared, so we may not want to use the handle state at
590-- all for these APIs. we can use pread and pwrite instead. On windows we will
591-- need to use readFile/writeFile with an offset argument.
593-------------------------------------------------------------------------------
595-- | Read the element at the given index treating the file as an array.
598{-# INLINE readIndex #-}
599readIndex :: Storable a => Handle -> Int -> Maybe a
600readIndex arr i = undefined
602-- NOTE: To represent a range to read we have chosen (start, size) instead of
603-- (start, end). This helps in removing the ambiguity of whether "end" is
604-- included in the range or not.
606-- We could avoid specifying the range to be read and instead use "take size"
607-- on the stream, but it may end up reading more and then consume it partially.
609-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
610-- from @handle@ starting at the offset @pos@ from the beginning of the file.
612-- Reads are performed in chunks of size @chunkSize@. For block devices, to
613-- avoid reading partial blocks @chunkSize@ must align with the block size of
614-- the underlying device. If the underlying block size is unknown, it is a good
615-- idea to keep it a multiple 4KiB. This API ensures that the start of each
616-- chunk is aligned with @chunkSize@ from second chunk onwards.
618{-# INLINE readSliceWith #-}
619readSliceWith :: (IsStream t, MonadIO m, Storable a)
620 => Int -> Handle -> Int -> Int -> t m a
621readSliceWith chunkSize h pos len = undefined
623-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
624-- at index @i@ and reading up to @count@ elements in the forward direction
625-- ending at the index @i + count - 1@.
628{-# INLINE readSlice #-}
629readSlice :: (IsStream t, MonadIO m, Storable a)
630 => Handle -> Int -> Int -> t m a
631readSlice = readSliceWith A.defaultChunkSize
633-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
634-- at index @i@ and reading up to @count@ elements in the reverse direction
635-- ending at the index @i - count + 1@.
638{-# INLINE readSliceRev #-}
639readSliceRev :: (IsStream t, MonadIO m, Storable a)
640 => Handle -> Int -> Int -> t m a
641readSliceRev h i count = undefined
643-- | Write the given element at the given index in the file.
646{-# INLINE writeIndex #-}
647writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
648writeIndex h i a = undefined
650-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
651-- starting at index @i@ and writing up to @count@ elements in the forward
652-- direction ending at the index @i + count - 1@.
655{-# INLINE writeSlice #-}
656writeSlice :: (IsStream t, Monad m, Storable a)
657 => Handle -> Int -> Int -> t m a -> m ()
658writeSlice h i len s = undefined
660-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
661-- starting at index @i@ and writing up to @count@ elements in the reverse
662-- direction ending at the index @i - count + 1@.
665{-# INLINE writeSliceRev #-}
666writeSliceRev :: (IsStream t, Monad m, Storable a)
667 => Handle -> Int -> Int -> t m a -> m ()
668writeSliceRev 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