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)
53 -- Byte stream write (Streams)
55 , fromBytesWithBufferOf
60 , writeChunksWithBufferOf
62 -- -- * Array stream Write
63 , fromChunksWithBufferOf
70 -- -- * Random Access (Seek)
71 -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or
72 -- files that have random access or seek capability. This type of devices
73 -- include disks, files, memory devices and exclude terminals, pipes,
76 -- XXX need to decide whether to use readFromStepN style or readFromThenTo
77 -- style. The latter is consistent with list enumeration. The former may be
78 -- more unambiguous and it is easier and clearer to specify 0 elements.
80 -- We can also generate the request pattern using a funciton.
83 -- , readFrom -- read from a given position to the end of file
84 -- , readFromRev -- read from a given position the beginning of file
85 -- , readTo -- read from beginning up to the given position
86 -- , readToRev -- read from end to the given position in file
92 -- , readChunksFromToWithBufferOf
93 -- , readChunksFromThenToWithBufferOf
96 -- , writeFrom -- start writing at the given position
98 -- , writeTo -- write from beginning up to the given position
104 -- , writeChunksFromTo
105 -- , writeChunksFromToWithBufferOf
106 -- , writeChunksFromThenToWithBufferOf
110import Control.Monad.IO.Class (MonadIO(..))
111#if __GLASGOW_HASKELL__ < 808
112import Data.Semigroup (Semigroup(..))
114import Data.Word (Word8)
115import Foreign.ForeignPtr (withForeignPtr)
116import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
117import Foreign.Ptr (minusPtr, plusPtr)
118import Foreign.Storable (Storable(..))
119import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
120import System.IO (Handle, hGetBufSome, hPutBuf, stdin, stdout)
121import Prelude hiding (read)
123import Streamly (MonadAsync)
124import Streamly.Data.Fold (Fold)
125import Streamly.Internal.Data.Fold.Types (Fold2(..))
126import Streamly.Internal.Data.Unfold.Types (Unfold(..))
127import Streamly.Internal.Memory.Array.Types
128 (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,
130import Streamly.Streams.Serial (SerialT)
131import Streamly.Streams.StreamK.Type (IsStream, mkStream)
132-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
134import qualified Streamly.Data.Fold as FL
135import qualified Streamly.Internal.Data.Fold.Types as FL
136import qualified Streamly.Internal.Data.Unicode.Stream as U
137import qualified Streamly.Internal.Data.Unfold as UF
138import qualified Streamly.Internal.Memory.Array as IA
139import qualified Streamly.Internal.Memory.ArrayStream as AS
140import qualified Streamly.Internal.Prelude as S
141import qualified Streamly.Memory.Array as A
142import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
144-------------------------------------------------------------------------------
146-------------------------------------------------------------------------------
148-- The following references may be useful to build an understanding about the
151-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
152-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
153-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
154-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
155-- https://www.w3.org/TR/FileAPI/ for http file API.
157-------------------------------------------------------------------------------
159-------------------------------------------------------------------------------
161-- | Read a 'ByteArray' from a file handle. If no data is available on the
162-- handle it blocks until some data becomes available. If data is available
163-- then it immediately returns that data without blocking. It reads a maximum
164-- of up to the size requested.
165{-# INLINABLE readArrayUpto #-}
166readArrayUpto :: Int -> Handle -> IO (Array Word8)
167readArrayUpto size h = do
168 ptr <- mallocPlainForeignPtrBytes size
169 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
170 withForeignPtr ptr $ \p -> do
171 n <- hGetBufSome h p size
174 , aEnd = p `plusPtr` n
175 , aBound = p `plusPtr` size
177 -- XXX shrink only if the diff is significant
180-------------------------------------------------------------------------------
181-- Stream of Arrays IO
182-------------------------------------------------------------------------------
184-- | @toChunksWithBufferOf size h@ reads a stream of arrays from file handle @h@.
185-- The maximum size of a single array is specified by @size@. The actual size
186-- read may be less than or equal to @size@.
187{-# INLINABLE _toChunksWithBufferOf #-}
188_toChunksWithBufferOf :: (IsStream t, MonadIO m)
189 => Int -> Handle -> t m (Array Word8)
190_toChunksWithBufferOf size h = go
192 -- XXX use cons/nil instead
193 go = mkStream $ \_ yld _ stp -> do
194 arr <- liftIO $ readArrayUpto size h
199-- | @toChunksWithBufferOf size handle@ reads a stream of arrays from the file
200-- handle @handle@. The maximum size of a single array is limited to @size@.
201-- The actual size read may be less than or equal to @size@.
204{-# INLINE_NORMAL toChunksWithBufferOf #-}
205toChunksWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
206toChunksWithBufferOf size h = D.fromStreamD (D.Stream step ())
208 {-# INLINE_LATE step #-}
210 arr <- liftIO $ readArrayUpto size h
216-- | Unfold the tuple @(bufsize, handle)@ into a stream of 'Word8' arrays.
217-- Read requests to the IO device are performed using a buffer of size
218-- @bufsize@. The size of an array in the resulting stream is always less than
219-- or equal to @bufsize@.
222{-# INLINE_NORMAL readChunksWithBufferOf #-}
223readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Handle) (Array Word8)
224readChunksWithBufferOf = Unfold step return
226 {-# INLINE_LATE step #-}
228 arr <- liftIO $ readArrayUpto size h
232 _ -> D.Yield arr (size, h)
234-- XXX read 'Array a' instead of Word8
236-- | @toChunks handle@ reads a stream of arrays from the specified file
237-- handle. The maximum size of a single array is limited to
238-- @defaultChunkSize@. The actual size read may be less than or equal to
239-- @defaultChunkSize@.
241-- > toChunks = toChunksWithBufferOf defaultChunkSize
244{-# INLINE toChunks #-}
245toChunks :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
246toChunks = toChunksWithBufferOf defaultChunkSize
248-- | Read a stream of chunks from standard input. The maximum size of a single
249-- chunk is limited to @defaultChunkSize@. The actual size read may be less
250-- than @defaultChunkSize@.
252-- > getChunks = toChunks stdin
256{-# INLINE getChunks #-}
257getChunks :: (IsStream t, MonadIO m) => t m (Array Word8)
258getChunks = toChunks stdin
260-- | Read a stream of bytes from standard input.
262-- > getBytes = toBytes stdin
266{-# INLINE getBytes #-}
267getBytes :: (IsStream t, MonadIO m) => t m Word8
268getBytes = toBytes stdin
270-- | Unfolds a handle into a stream of 'Word8' arrays. Requests to the IO
271-- device are performed using a buffer of size
272-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'. The
273-- size of arrays in the resulting stream are therefore less than or equal to
274-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
277{-# INLINE readChunks #-}
278readChunks :: MonadIO m => Unfold m Handle (Array Word8)
279readChunks = UF.supplyFirst readChunksWithBufferOf defaultChunkSize
281-------------------------------------------------------------------------------
282-- Read File to Stream
283-------------------------------------------------------------------------------
285-- TODO for concurrent streams implement readahead IO. We can send multiple
286-- read requests at the same time. For serial case we can use async IO. We can
287-- also control the read throughput in mbps or IOPS.
289-- | Unfolds the tuple @(bufsize, handle)@ into a byte stream, read requests
290-- to the IO device are performed using buffers of @bufsize@.
293{-# INLINE readWithBufferOf #-}
294readWithBufferOf :: MonadIO m => Unfold m (Int, Handle) Word8
295readWithBufferOf = UF.concat readChunksWithBufferOf A.read
297-- | @toBytesWithBufferOf bufsize handle@ reads a byte stream from a file
298-- handle, reads are performed in chunks of up to @bufsize@.
301{-# INLINE toBytesWithBufferOf #-}
302toBytesWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
303toBytesWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h
306-- Generate a stream of elements of the given type from a file 'Handle'.
307-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
309-- | Unfolds a file handle into a byte stream. IO requests to the device are
310-- performed in sizes of
311-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
315read :: MonadIO m => Unfold m Handle Word8
316read = UF.supplyFirst readWithBufferOf defaultChunkSize
318-- | Generate a byte stream from a file 'Handle'.
321{-# INLINE toBytes #-}
322toBytes :: (IsStream t, MonadIO m) => Handle -> t m Word8
323toBytes = AS.concat . toChunks
325-------------------------------------------------------------------------------
327-------------------------------------------------------------------------------
329-------------------------------------------------------------------------------
331-------------------------------------------------------------------------------
333-- | Write an 'Array' to a file handle.
336{-# INLINABLE writeArray #-}
337writeArray :: Storable a => Handle -> Array a -> IO ()
338writeArray _ arr | A.length arr == 0 = return ()
339writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen
342 let p = unsafeForeignPtrToPtr aStart
345-------------------------------------------------------------------------------
346-- Stream of Arrays IO
347-------------------------------------------------------------------------------
349-------------------------------------------------------------------------------
351-------------------------------------------------------------------------------
353-- XXX use an unfold to fromObjects or fromUnfold so that we can put any object
354-- | Write a stream of arrays to a handle.
357{-# INLINE fromChunks #-}
358fromChunks :: (MonadIO m, Storable a)
359 => Handle -> SerialT m (Array a) -> m ()
360fromChunks h m = S.mapM_ (liftIO . writeArray h) m
362-- | Write a stream of chunks to standard output.
366{-# INLINE putChunks #-}
367putChunks :: (MonadIO m, Storable a) => SerialT m (Array a) -> m ()
368putChunks = fromChunks stdout
370-- | Write a stream of strings to standard output using Latin1 encoding.
371-- Output is flushed to the device for each string.
375{-# INLINE putStrings #-}
376putStrings :: MonadAsync m => SerialT m String -> m ()
377putStrings = putChunks . S.mapM (IA.fromStream . U.encodeLatin1 . S.fromList)
379-- XXX use an unfold so that we can put lines from any object
380-- | Write a stream of strings as separate lines to standard output using
381-- Latin1 encoding. Output is line buffered i.e. the output is written to the
382-- device as soon as a newline is encountered.
386{-# INLINE putLines #-}
387putLines :: MonadAsync m => SerialT m String -> m ()
388putLines = putChunks . S.mapM
389 (\xs -> IA.fromStream $ U.encodeLatin1 (S.fromList xs <> S.yield '\n'))
391-- | Write a stream of bytes from standard output.
393-- > putBytes = fromBytes stdout
397{-# INLINE putBytes #-}
398putBytes :: MonadIO m => SerialT m Word8 -> m ()
399putBytes = fromBytes stdout
401-- | @fromChunksWithBufferOf bufsize handle stream@ writes a stream of arrays
402-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
403-- The chunk size is only a maximum and the actual writes could be smaller as
404-- we do not split the arrays to fit exactly to the specified size.
407{-# INLINE fromChunksWithBufferOf #-}
408fromChunksWithBufferOf :: (MonadIO m, Storable a)
409 => Int -> Handle -> SerialT m (Array a) -> m ()
410fromChunksWithBufferOf n h xs = fromChunks h $ AS.compact n xs
412-- | @fromBytesWithBufferOf bufsize handle stream@ writes @stream@ to @handle@
413-- in chunks of @bufsize@. A write is performed to the IO device as soon as we
414-- collect the required input size.
417{-# INLINE fromBytesWithBufferOf #-}
418fromBytesWithBufferOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
419fromBytesWithBufferOf n h m = fromChunks h $ S.arraysOf n m
420-- fromBytesWithBufferOf n h m = fromChunks h $ AS.arraysOf n m
422-- > write = 'writeWithBufferOf' A.defaultChunkSize
424-- | Write a byte stream to a file handle. Accumulates the input in chunks of
425-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing.
427-- NOTE: This may perform better than the 'write' fold, you can try this if you
428-- need some extra perf boost.
431{-# INLINE fromBytes #-}
432fromBytes :: MonadIO m => Handle -> SerialT m Word8 -> m ()
433fromBytes = fromBytesWithBufferOf defaultChunkSize
435-- | Write a stream of arrays to a handle. Each array in the stream is written
436-- to the device as a separate IO request.
439{-# INLINE writeChunks #-}
440writeChunks :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
441writeChunks h = FL.drainBy (liftIO . writeArray h)
443{-# INLINE writeChunks2 #-}
444writeChunks2 :: (MonadIO m, Storable a) => Fold2 m Handle (Array a) ()
445writeChunks2 = Fold2 (\h arr -> liftIO $ writeArray h arr >> return h) return (\_ -> return ())
447-- | @writeChunksWithBufferOf bufsize handle@ writes a stream of arrays
448-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
449-- We never split an array, if a single array is bigger than the specified size
450-- it emitted as it is. Multiple arrays are coalesed as long as the total size
451-- remains below the specified size.
454{-# INLINE writeChunksWithBufferOf #-}
455writeChunksWithBufferOf :: (MonadIO m, Storable a)
456 => Int -> Handle -> Fold m (Array a) ()
457writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)
459-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
462-- Note that if you use a chunk size less than 8K (GHC's default buffer
463-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
464-- do not want buffering to occur at GHC level as well. Same thing applies to
467-- | @writeWithBufferOf reqSize handle@ writes the input stream to @handle@.
468-- Bytes in the input stream are collected into a buffer until we have a chunk
469-- of @reqSize@ and then written to the IO device.
472{-# INLINE writeWithBufferOf #-}
473writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
474writeWithBufferOf n h = FL.lchunksOf n (writeNUnsafe n) (writeChunks h)
476{-# INLINE writeWithBufferOf2 #-}
477writeWithBufferOf2 :: MonadIO m => Int -> Fold2 m Handle Word8 ()
478writeWithBufferOf2 n = FL.lchunksOf2 n (writeNUnsafe n) writeChunks2
480-- > write = 'writeWithBufferOf' A.defaultChunkSize
482-- | Write a byte stream to a file handle. Accumulates the input in chunks of
483-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing
488write :: MonadIO m => Handle -> Fold m Word8 ()
489write = writeWithBufferOf defaultChunkSize
492write2 :: MonadIO m => Fold2 m Handle Word8 ()
493write2 = writeWithBufferOf2 defaultChunkSize
497write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
498write = toHandleWith A.defaultChunkSize
501-- XXX mmap a file into an array. This could be useful for in-place operations
502-- on a file. For example, we can quicksort the contents of a file by mmapping
505-------------------------------------------------------------------------------
506-- IO with encoding/decoding Unicode characters
507-------------------------------------------------------------------------------
511-- > readUtf8 = decodeUtf8 . read
513-- Read a UTF8 encoded stream of unicode characters from a file handle.
516{-# INLINE readUtf8 #-}
517readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
518readUtf8 = decodeUtf8 . read
521-- > writeUtf8 h s = write h $ encodeUtf8 s
523-- Encode a stream of unicode characters to UTF8 and write it to the given file
524-- handle. Default block buffering applies to the writes.
527{-# INLINE writeUtf8 #-}
528writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
529writeUtf8 h s = write h $ encodeUtf8 s
531-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
532-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
533-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
534-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
538{-# INLINE writeUtf8ByLines #-}
539writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
540writeUtf8ByLines = undefined
542-- | Read UTF-8 lines from a file handle and apply the specified fold to each
543-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
546{-# INLINE readLines #-}
547readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
548readLines h f = foldLines (readUtf8 h) f
550-------------------------------------------------------------------------------
551-- Framing on a sequence
552-------------------------------------------------------------------------------
554-- | Read a stream from a file handle and split it into frames delimited by
555-- the specified sequence of elements. The supplied fold is applied on each
559{-# INLINE readFrames #-}
560readFrames :: (IsStream t, MonadIO m, Storable a)
561 => Array a -> Handle -> Fold m a b -> t m b
562readFrames = undefined -- foldFrames . read
564-- | Write a stream to the given file handle buffering up to frames separated
565-- by the given sequence or up to a maximum of @defaultChunkSize@.
568{-# INLINE writeByFrames #-}
569writeByFrames :: (IsStream t, MonadIO m, Storable a)
570 => Array a -> Handle -> t m a -> m ()
571writeByFrames = undefined
573-------------------------------------------------------------------------------
575-------------------------------------------------------------------------------
577-- | Write collecting the input in sessions of n seconds or if chunkSize
579{-# INLINE writeByChunksOrSessionsOf #-}
580writeByChunksOrSessionsOf :: MonadIO m
581 => Int -> Double -> Handle -> SerialT m Word8 -> m ()
582writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined
584-- | Write collecting the input in sessions of n seconds or if defaultChunkSize
586{-# INLINE writeBySessionsOf #-}
587writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()
588writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n
590-------------------------------------------------------------------------------
591-- Random Access IO (Seek)
592-------------------------------------------------------------------------------
594-- XXX handles could be shared, so we may not want to use the handle state at
595-- all for these APIs. we can use pread and pwrite instead. On windows we will
596-- need to use readFile/writeFile with an offset argument.
598-------------------------------------------------------------------------------
600-- | Read the element at the given index treating the file as an array.
603{-# INLINE readIndex #-}
604readIndex :: Storable a => Handle -> Int -> Maybe a
605readIndex arr i = undefined
607-- NOTE: To represent a range to read we have chosen (start, size) instead of
608-- (start, end). This helps in removing the ambiguity of whether "end" is
609-- included in the range or not.
611-- We could avoid specifying the range to be read and instead use "take size"
612-- on the stream, but it may end up reading more and then consume it partially.
614-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
615-- from @handle@ starting at the offset @pos@ from the beginning of the file.
617-- Reads are performed in chunks of size @chunkSize@. For block devices, to
618-- avoid reading partial blocks @chunkSize@ must align with the block size of
619-- the underlying device. If the underlying block size is unknown, it is a good
620-- idea to keep it a multiple 4KiB. This API ensures that the start of each
621-- chunk is aligned with @chunkSize@ from second chunk onwards.
623{-# INLINE readSliceWith #-}
624readSliceWith :: (IsStream t, MonadIO m, Storable a)
625 => Int -> Handle -> Int -> Int -> t m a
626readSliceWith chunkSize h pos len = undefined
628-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
629-- at index @i@ and reading up to @count@ elements in the forward direction
630-- ending at the index @i + count - 1@.
633{-# INLINE readSlice #-}
634readSlice :: (IsStream t, MonadIO m, Storable a)
635 => Handle -> Int -> Int -> t m a
636readSlice = readSliceWith A.defaultChunkSize
638-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
639-- at index @i@ and reading up to @count@ elements in the reverse direction
640-- ending at the index @i - count + 1@.
643{-# INLINE readSliceRev #-}
644readSliceRev :: (IsStream t, MonadIO m, Storable a)
645 => Handle -> Int -> Int -> t m a
646readSliceRev h i count = undefined
648-- | Write the given element at the given index in the file.
651{-# INLINE writeIndex #-}
652writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
653writeIndex h i a = undefined
655-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
656-- starting at index @i@ and writing up to @count@ elements in the forward
657-- direction ending at the index @i + count - 1@.
660{-# INLINE writeSlice #-}
661writeSlice :: (IsStream t, Monad m, Storable a)
662 => Handle -> Int -> Int -> t m a -> m ()
663writeSlice h i len s = undefined
665-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
666-- starting at index @i@ and writing up to @count@ elements in the reverse
667-- direction ending at the index @i - count + 1@.
670{-# INLINE writeSliceRev #-}
671writeSliceRev :: (IsStream t, Monad m, Storable a)
672 => Handle -> Int -> Int -> t m a -> m ()
673writeSliceRev 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