4-- Module : Streamly.FileSystem.FD
5-- Copyright : (c) 2019 Composewell Technologies
8-- Maintainer : streamly@composewell.com
9-- Stability : experimental
12-- This module is a an experimental replacement for
13-- "Streamly.FileSystem.Handle". The former module provides IO facilities based
14-- on the GHC Handle type. The APIs in this module avoid the GHC handle layer
15-- and provide more explicit control over buffering.
17-- Read and write data as streams and arrays to and from files.
19-- This module provides read and write APIs based on handles. Before reading or
20-- writing, a file must be opened first using 'openFile'. The 'Handle' returned
21-- by 'openFile' is then used to access the file. A 'Handle' is backed by an
22-- operating system file descriptor. When the 'Handle' is garbage collected the
23-- underlying file descriptor is automatically closed. A handle can be
24-- explicitly closed using 'closeFile'.
26-- Reading and writing APIs are divided into two categories, sequential
27-- streaming APIs and random or seekable access APIs. File IO APIs are quite
28-- similar to "Streamly.Data.Array.Storable.Foreign" read write APIs. In that regard, arrays can
29-- be considered as in-memory files or files can be considered as on-disk
32-- > import qualified Streamly.FileSystem.FD as FD
35module Streamly.FileSystem.FD
44 -- TODO file path based APIs
49 -- | Streaming APIs read or write data to or from a file or device
50 -- sequentially, they never perform a seek to a random location. When
51 -- reading, the stream is lazy and generated on-demand as the consumer
52 -- consumes it. Read IO requests to the IO device are performed in chunks
53 -- of 32KiB, this is referred to as @defaultChunkSize@ in the
54 -- documentation. One IO request may or may not read the full chunk. If the
55 -- whole stream is not consumed, it is possible that we may read slightly
56 -- more from the IO device than what the consumer needed. Unless specified
57 -- otherwise in the API, writes are collected into chunks of
58 -- @defaultChunkSize@ before they are written to the IO device.
60 -- Streaming APIs work for all kind of devices, seekable or non-seekable;
61 -- including disks, files, memory devices, terminals, pipes, sockets and
62 -- fifos. While random access APIs work only for files or devices that have
63 -- random access or seek capability for example disks, memory devices.
64 -- Devices like terminals, pipes, sockets and fifos do not have random
67 -- ** Read File to Stream
82 -- ** Write File from Stream
92 , writeArraysPackedUpto
94 -- XXX these are incomplete
96 -- , writevArraysPackedUpto
98 -- -- * Random Access (Seek)
99 -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or
100 -- files that have random access or seek capability. This type of devices
101 -- include disks, files, memory devices and exclude terminals, pipes,
102 -- sockets and fifos.
107 -- , readAt -- read from a given position to th end of file
108 -- , readSliceArrayUpto
109 -- , readSliceArrayOf
114 -- , writeAt -- start writing at the given position
119import Control.Monad.IO.Class (MonadIO(..))
120import Data.Word (Word8)
121import Foreign.ForeignPtr (withForeignPtr)
122-- import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
123import Foreign.Ptr (plusPtr, castPtr)
124import Foreign.Storable (Storable(..))
125import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
126-- import System.IO (Handle, hGetBufSome, hPutBuf)
127import System.IO (IOMode)
128import Prelude hiding (read)
130import qualified GHC.IO.FD as FD
131import qualified GHC.IO.Device as RawIO
133import Streamly.Internal.Data.Array.Storable.Foreign.Types (Array(..), byteLength, defaultChunkSize, unsafeFreeze)
134import Streamly.Internal.Data.Array.Storable.Foreign.Mut.Types (mutableArray)
136import Streamly.Internal.Data.Stream.Serial (SerialT)
137import Streamly.Internal.Data.Stream.StreamK.Type (IsStream, mkStream)
139#if !defined(mingw32_HOST_OS)
140import Streamly.Internal.Data.Array.Storable.Foreign.Types (groupIOVecsOf)
141import Streamly.Internal.Data.Stream.StreamD (toStreamD)
142import Streamly.Internal.Data.Stream.StreamD.Type (fromStreamD)
143import qualified Streamly.FileSystem.FDIO as RawIO hiding (write)
145-- import Streamly.Data.Fold (Fold)
146-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
148import qualified Streamly.Data.Array.Storable.Foreign as A
149import qualified Streamly.Internal.Memory.ArrayStream as AS
150import qualified Streamly.Prelude as S
151import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
153-------------------------------------------------------------------------------
155-------------------------------------------------------------------------------
157-- The following references may be useful to build an understanding about the
160-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
161-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
162-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
163-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
164-- https://www.w3.org/TR/FileAPI/ for http file API.
166-------------------------------------------------------------------------------
168-------------------------------------------------------------------------------
170-- XXX attach a finalizer
171-- | A 'Handle' is returned by 'openFile' and is subsequently used to perform
172-- read and write operations on a file.
174newtype Handle = Handle FD.FD
176-- | File handle for standard input
178stdin = Handle FD.stdin
180-- | File handle for standard output
182stdout = Handle FD.stdout
184-- | File handle for standard error
186stderr = Handle FD.stderr
188-- XXX we can support all the flags that the "open" system call supports.
189-- Instead of using RTS locking mechanism can we use system provided locking
192-- | Open a file that is not a directory and return a file handle.
193-- 'openFile' enforces a multiple-reader single-writer locking on files. That
194-- is, there may either be many handles on the same file which manage input, or
195-- just one handle on the file which manages output. If any open handle is
196-- managing a file for output, no new handle can be allocated for that file. If
197-- any open handle is managing a file for input, new handles can only be
198-- allocated if they do not manage output. Whether two files are the same is
199-- implementation-dependent, but they should normally be the same if they have
200-- the same absolute path name and neither has been renamed, for example.
202openFile :: FilePath -> IOMode -> IO Handle
203openFile path mode = Handle . fst <$> FD.openFile path mode True
205-------------------------------------------------------------------------------
207-------------------------------------------------------------------------------
209-- | Read a 'ByteArray' from a file handle. If no data is available on the
210-- handle it blocks until some data becomes available. If data is available
211-- then it immediately returns that data without blocking. It reads a maximum
212-- of up to the size requested.
213{-# INLINABLE readArrayUpto #-}
214readArrayUpto :: Int -> Handle -> IO (Array Word8)
215readArrayUpto size (Handle fd) = do
216 ptr <- mallocPlainForeignPtrBytes size
217 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
218 withForeignPtr ptr $ \p -> do
219 -- n <- hGetBufSome h p size
220 n <- RawIO.read fd p size
221 -- XXX shrink only if the diff is significant
222 -- Use unsafeFreezeWithShrink
224 unsafeFreeze $ mutableArray ptr (p `plusPtr` n) (p `plusPtr` size)
226-------------------------------------------------------------------------------
228-------------------------------------------------------------------------------
230-- | Write an 'Array' to a file handle.
233{-# INLINABLE writeArray #-}
234writeArray :: Storable a => Handle -> Array a -> IO ()
235writeArray _ arr | A.length arr == 0 = return ()
236writeArray (Handle fd) arr = withForeignPtr (aStart arr) $ \p ->
237 -- RawIO.writeAll fd (castPtr p) aLen
238 RawIO.write fd (castPtr p) aLen
240 -- Experiment to compare "writev" based IO with "write" based IO.
242 let iov' = iov {aEnd = aBound iov}
243 A.writeIndex iov' 0 (RawIO.IOVec (castPtr p) (fromIntegral aLen))
244 RawIO.writevAll fd (unsafeForeignPtrToPtr (aStart iov')) 1
247 aLen = byteLength arr
249#if !defined(mingw32_HOST_OS)
250-- | Write an array of 'IOVec' to a file handle.
253{-# INLINABLE writeIOVec #-}
254writeIOVec :: Handle -> Array RawIO.IOVec -> IO ()
255writeIOVec _ iov | A.length iov == 0 = return ()
256writeIOVec (Handle fd) iov =
257 withForeignPtr (aStart iov) $ \p ->
258 RawIO.writevAll fd p (A.length iov)
261-------------------------------------------------------------------------------
262-- Stream of Arrays IO
263-------------------------------------------------------------------------------
265-- | @readArraysOfUpto size h@ reads a stream of arrays from file handle @h@.
266-- The maximum size of a single array is specified by @size@. The actual size
267-- read may be less than or equal to @size@.
268{-# INLINABLE _readArraysOfUpto #-}
269_readArraysOfUpto :: (IsStream t, MonadIO m)
270 => Int -> Handle -> t m (Array Word8)
271_readArraysOfUpto size h = go
273 -- XXX use cons/nil instead
274 go = mkStream $ \_ yld _ stp -> do
275 arr <- liftIO $ readArrayUpto size h
280{-# INLINE_NORMAL readArraysOfUpto #-}
281readArraysOfUpto :: (IsStream t, MonadIO m)
282 => Int -> Handle -> t m (Array Word8)
283readArraysOfUpto size h = D.fromStreamD (D.Stream step ())
285 {-# INLINE_LATE step #-}
287 arr <- liftIO $ readArrayUpto size h
293-- XXX read 'Array a' instead of Word8
295-- | @readArrays h@ reads a stream of arrays from file handle @h@.
296-- The maximum size of a single array is limited to @defaultChunkSize@.
297-- 'readArrays' ignores the prevailing 'TextEncoding' and 'NewlineMode'
300-- > readArrays = readArraysOfUpto defaultChunkSize
303{-# INLINE readArrays #-}
304readArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
305readArrays = readArraysOfUpto defaultChunkSize
307-------------------------------------------------------------------------------
308-- Read File to Stream
309-------------------------------------------------------------------------------
311-- TODO for concurrent streams implement readahead IO. We can send multiple
312-- read requests at the same time. For serial case we can use async IO. We can
313-- also control the read throughput in mbps or IOPS.
315-- | @readInChunksOf chunkSize handle@ reads a byte stream from a file handle,
316-- reads are performed in chunks of up to @chunkSize@. The stream ends as soon
317-- as EOF is encountered.
319{-# INLINE readInChunksOf #-}
320readInChunksOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
321readInChunksOf chunkSize h = AS.concat $ readArraysOfUpto chunkSize h
324-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
326-- > read = 'readByChunks' A.defaultChunkSize
327-- | Generate a stream of elements of the given type from a file 'Handle'. The
328-- stream ends when EOF is encountered.
332read :: (IsStream t, MonadIO m) => Handle -> t m Word8
333read = AS.concat . readArrays
335-------------------------------------------------------------------------------
337-------------------------------------------------------------------------------
339-- | Write a stream of arrays to a handle.
342{-# INLINE writeArrays #-}
343writeArrays :: (MonadIO m, Storable a) => Handle -> SerialT m (Array a) -> m ()
344writeArrays h = S.mapM_ (liftIO . writeArray h)
346-- | Write a stream of arrays to a handle after coalescing them in chunks of
347-- specified size. The chunk size is only a maximum and the actual writes could
348-- be smaller than that as we do not split the arrays to fit them to the
352{-# INLINE writeArraysPackedUpto #-}
353writeArraysPackedUpto :: (MonadIO m, Storable a)
354 => Int -> Handle -> SerialT m (Array a) -> m ()
355writeArraysPackedUpto n h xs = writeArrays h $ AS.compact n xs
357#if !defined(mingw32_HOST_OS)
358-- XXX this is incomplete
359-- | Write a stream of 'IOVec' arrays to a handle.
363writev :: MonadIO m => Handle -> SerialT m (Array RawIO.IOVec) -> m ()
364writev h = S.mapM_ (liftIO . writeIOVec h)
366-- XXX this is incomplete
367-- | Write a stream of arrays to a handle after grouping them in 'IOVec' arrays
368-- of up to a maximum total size. Writes are performed using gather IO via
369-- @writev@ system call. The maximum number of entries in each 'IOVec' group
373{-# INLINE _writevArraysPackedUpto #-}
374_writevArraysPackedUpto :: MonadIO m
375 => Int -> Handle -> SerialT m (Array a) -> m ()
376_writevArraysPackedUpto n h xs =
377 writev h $ fromStreamD $ groupIOVecsOf n 512 (toStreamD xs)
380-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
383-- Note that if you use a chunk size less than 8K (GHC's default buffer
384-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
385-- do not want buffering to occur at GHC level as well. Same thing applies to
388-- | Like 'write' but provides control over the write buffer. Output will
389-- be written to the IO device as soon as we collect the specified number of
393{-# INLINE writeInChunksOf #-}
394writeInChunksOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
395writeInChunksOf n h m = writeArrays h $ AS.arraysOf n m
397-- > write = 'writeInChunksOf' A.defaultChunkSize
399-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
400-- up to 'A.defaultChunkSize' before writing. Note that the write behavior
401-- depends on the 'IOMode' and the current seek position of the handle.
405write :: MonadIO m => Handle -> SerialT m Word8 -> m ()
406write = writeInChunksOf defaultChunkSize
410write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
411write = toHandleWith A.defaultChunkSize
414-------------------------------------------------------------------------------
415-- IO with encoding/decoding Unicode characters
416-------------------------------------------------------------------------------
420-- > readUtf8 = decodeUtf8 . read
422-- Read a UTF8 encoded stream of unicode characters from a file handle.
425{-# INLINE readUtf8 #-}
426readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
427readUtf8 = decodeUtf8 . read
430-- > writeUtf8 h s = write h $ encodeUtf8 s
432-- Encode a stream of unicode characters to UTF8 and write it to the given file
433-- handle. Default block buffering applies to the writes.
436{-# INLINE writeUtf8 #-}
437writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
438writeUtf8 h s = write h $ encodeUtf8 s
440-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
441-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
442-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
443-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
447{-# INLINE writeUtf8ByLines #-}
448writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
449writeUtf8ByLines = undefined
451-- | Read UTF-8 lines from a file handle and apply the specified fold to each
452-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
455{-# INLINE readLines #-}
456readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
457readLines h f = foldLines (readUtf8 h) f
459-------------------------------------------------------------------------------
460-- Framing on a sequence
461-------------------------------------------------------------------------------
463-- | Read a stream from a file handle and split it into frames delimited by
464-- the specified sequence of elements. The supplied fold is applied on each
468{-# INLINE readFrames #-}
469readFrames :: (IsStream t, MonadIO m, Storable a)
470 => Array a -> Handle -> Fold m a b -> t m b
471readFrames = undefined -- foldFrames . read
473-- | Write a stream to the given file handle buffering up to frames separated
474-- by the given sequence or up to a maximum of @defaultChunkSize@.
477{-# INLINE writeByFrames #-}
478writeByFrames :: (IsStream t, MonadIO m, Storable a)
479 => Array a -> Handle -> t m a -> m ()
480writeByFrames = undefined
482-------------------------------------------------------------------------------
483-- Random Access IO (Seek)
484-------------------------------------------------------------------------------
486-- XXX handles could be shared, so we may not want to use the handle state at
487-- all for these APIs. we can use pread and pwrite instead. On windows we will
488-- need to use readFile/writeFile with an offset argument.
490-------------------------------------------------------------------------------
492-- | Read the element at the given index treating the file as an array.
495{-# INLINE readIndex #-}
496readIndex :: Storable a => Handle -> Int -> Maybe a
497readIndex arr i = undefined
499-- NOTE: To represent a range to read we have chosen (start, size) instead of
500-- (start, end). This helps in removing the ambiguity of whether "end" is
501-- included in the range or not.
503-- We could avoid specifying the range to be read and instead use "take size"
504-- on the stream, but it may end up reading more and then consume it partially.
506-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
507-- from @handle@ starting at the offset @pos@ from the beginning of the file.
509-- Reads are performed in chunks of size @chunkSize@. For block devices, to
510-- avoid reading partial blocks @chunkSize@ must align with the block size of
511-- the underlying device. If the underlying block size is unknown, it is a good
512-- idea to keep it a multiple 4KiB. This API ensures that the start of each
513-- chunk is aligned with @chunkSize@ from second chunk onwards.
515{-# INLINE readSliceWith #-}
516readSliceWith :: (IsStream t, MonadIO m, Storable a)
517 => Int -> Handle -> Int -> Int -> t m a
518readSliceWith chunkSize h pos len = undefined
520-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
521-- at index @i@ and reading up to @count@ elements in the forward direction
522-- ending at the index @i + count - 1@.
525{-# INLINE readSlice #-}
526readSlice :: (IsStream t, MonadIO m, Storable a)
527 => Handle -> Int -> Int -> t m a
528readSlice = readSliceWith defaultChunkSize
530-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
531-- at index @i@ and reading up to @count@ elements in the reverse direction
532-- ending at the index @i - count + 1@.
535{-# INLINE readSliceRev #-}
536readSliceRev :: (IsStream t, MonadIO m, Storable a)
537 => Handle -> Int -> Int -> t m a
538readSliceRev h i count = undefined
540-- | Write the given element at the given index in the file.
543{-# INLINE writeIndex #-}
544writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
545writeIndex h i a = undefined
547-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
548-- starting at index @i@ and writing up to @count@ elements in the forward
549-- direction ending at the index @i + count - 1@.
552{-# INLINE writeSlice #-}
553writeSlice :: (IsStream t, Monad m, Storable a)
554 => Handle -> Int -> Int -> t m a -> m ()
555writeSlice h i len s = undefined
557-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
558-- starting at index @i@ and writing up to @count@ elements in the reverse
559-- direction ending at the index @i - count + 1@.
562{-# INLINE writeSliceRev #-}
563writeSliceRev :: (IsStream t, Monad m, Storable a)
564 => Handle -> Int -> Int -> t m a -> m ()
565writeSliceRev 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