@@ -0,0 +1,432 @@
2+{-# LANGUAGE BangPatterns #-}
3+{-# LANGUAGE MagicHash #-}
4+{-# LANGUAGE RecordWildCards #-}
5+{-# LANGUAGE UnboxedTuples #-}
7+#include "../Streams/inline.hs"
10+-- Module : Streamly.FileSystem.File
11+-- Copyright : (c) 2018 Harendra Kumar
14+-- Maintainer : harendra.kumar@gmail.com
15+-- Stability : experimental
18+-- Read and write streams and arrays to and from files. File IO APIs are quite
19+-- similar to "Streamly.Array" read write APIs. In that regard, arrays can be
20+-- considered as in-memory files or files can be considered as on-disk arrays.
21+-- IO APIs are divided into two categories, sequential streaming IO APIs and
22+-- random access IO APIs. Control over the file reading and writing behavior
23+-- in terms of buffering, encoding, decoding is in the hands of the
24+-- programmer, the 'TextEncoding', 'NewLineMode', and 'Buffering' options of
25+-- the underlying handle provided by GHC are not needed and ignored.
27+-- > import qualified Streamly.FileSystem.File as File
30+module Streamly.FileSystem.File
33+ -- | Stream data to or from a file or device sequentially. When reading,
34+ -- the stream is lazy and generated on-demand as the consumer consumes it.
35+ -- Read IO requests to the IO device are performed in chunks limited to a
36+ -- maximum size of 32KiB, this is referred to as @defaultChunkSize@ in the
37+ -- documentation. One IO request may or may not read the full
38+ -- chunk. If the whole stream is not consumed, it is possible that we may
39+ -- read slightly more from the IO device than what the consumer needed.
40+ -- Unless specified otherwise in the API, writes are collected into chunks
41+ -- of @defaultChunkSize@ before they are written to the IO device.
43+ -- Streaming APIs work for all kind of devices, seekable or non-seekable;
44+ -- including disks, files, memory devices, terminals, pipes, sockets and
45+ -- fifos. While random access APIs work only for files or devices that have
46+ -- random access or seek capability for example disks, memory devices.
47+ -- Devices like terminals, pipes, sockets and fifos do not have random
48+ -- access capability.
50+ -- TODO file path based APIs
54+ -- ** Read File to Stream
69+ -- ** Write File from Stream
72+ -- , writeUtf8ByLines
80+ -- -- * Random Access (Seek)
81+ -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or
82+ -- files that have random access or seek capability. This type of devices
83+ -- include disks, files, memory devices and exclude terminals, pipes,
84+ -- sockets and fifos.
89+ -- , readAt -- read from a given position to th end of file
90+ -- , readSliceArrayUpto
91+ -- , readSliceArrayOf
96+ -- , writeAt -- start writing at the given position
101+import Control.Monad.IO.Class (MonadIO(..))
102+import Data.Word (Word8)
103+import Foreign.ForeignPtr (withForeignPtr)
104+import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
105+import Foreign.Ptr (minusPtr, plusPtr)
106+import Foreign.Storable (Storable(..))
107+import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
108+import System.IO (Handle, hGetBufSome, hPutBuf)
109+import Prelude hiding (read)
111+import Streamly.Array.Types (Array(..))
112+import Streamly.Streams.Serial (SerialT)
113+import Streamly.Streams.StreamK.Type (IsStream, mkStream)
114+-- import Streamly.Fold (Fold)
115+-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
117+import qualified Streamly.Array as A
118+import qualified Streamly.Array.Types as A hiding (flattenArrays)
119+import qualified Streamly.Prelude as S
121+-------------------------------------------------------------------------------
123+-------------------------------------------------------------------------------
125+-- The following references may be useful to build an understanding about the
128+-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
129+-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
130+-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
131+-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
132+-- https://www.w3.org/TR/FileAPI/ for http file API.
134+-------------------------------------------------------------------------------
136+-------------------------------------------------------------------------------
138+-- | Read a 'ByteArray' from a file handle. If no data is available on the
139+-- handle it blocks until some data becomes available. If data is available
140+-- then it immediately returns that data without blocking. It reads a maximum
141+-- of up to the size requested.
142+{-# INLINABLE readArrayUpto #-}
143+readArrayUpto :: Int -> Handle -> IO (Array Word8)
144+readArrayUpto size h = do
145+ ptr <- mallocPlainForeignPtrBytes size
146+ -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
147+ withForeignPtr ptr $ \p -> do
148+ n <- hGetBufSome h p size
151+ , aEnd = p `plusPtr` n
152+ , aBound = p `plusPtr` size
154+ -- XXX shrink only if the diff is significant
157+-------------------------------------------------------------------------------
159+-------------------------------------------------------------------------------
161+-- | Write an Array to a file handle.
164+{-# INLINABLE writeArray #-}
165+writeArray :: Storable a => Handle -> Array a -> IO ()
166+writeArray _ arr | A.length arr == 0 = return ()
167+writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen
170+ let p = unsafeForeignPtrToPtr aStart
171+ in aEnd `minusPtr` p
173+-------------------------------------------------------------------------------
174+-- Stream of Arrays IO
175+-------------------------------------------------------------------------------
177+-- | @readArraysUpto size h@ reads a stream of arrays from file handle @h@.
178+-- The maximum size of a single array is limited to @size@.
179+-- 'fromHandleArraysUpto' ignores the prevailing 'TextEncoding' and 'NewlineMode'
181+{-# INLINABLE readArraysUpto #-}
182+readArraysUpto :: (IsStream t, MonadIO m)
183+ => Int -> Handle -> t m (Array Word8)
184+readArraysUpto size h = go
186+ -- XXX use cons/nil instead
187+ go = mkStream $ \_ yld sng _ -> do
188+ arr <- liftIO $ readArrayUpto size h
189+ if A.length arr < size
193+-- XXX read 'Array a' instead of Word8
195+-- | @readArrays h@ reads a stream of arrays from file handle @h@.
196+-- The maximum size of a single array is limited to @defaultChunkSize@.
197+-- 'readArrays' ignores the prevailing 'TextEncoding' and 'NewlineMode'
201+{-# INLINE readArrays #-}
202+readArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
203+readArrays = readArraysUpto A.defaultChunkSize
205+-------------------------------------------------------------------------------
206+-- Read File to Stream
207+-------------------------------------------------------------------------------
209+-- TODO for concurrent streams implement readahead IO. We can send multiple
210+-- read requests at the same time. For serial case we can use async IO. We can
211+-- also control the read throughput in mbps or IOPS.
214+-- | @readByChunksUpto chunkSize handle@ reads a byte stream from a file
215+-- handle, reads are performed in chunks of up to @chunkSize@. The stream ends
216+-- as soon as EOF is encountered.
218+{-# INLINE readByChunksUpto #-}
219+readByChunksUpto :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
220+readByChunksUpto chunkSize h = A.flattenArrays $ readArraysUpto chunkSize h
224+-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
226+-- > read = 'readByChunks' A.defaultChunkSize
227+-- | Generate a stream of elements of the given type from a file 'Handle'. The
228+-- stream ends when EOF is encountered.
232+read :: (IsStream t, MonadIO m) => Handle -> t m Word8
233+read = A.flattenArrays . readArrays
235+-------------------------------------------------------------------------------
237+-------------------------------------------------------------------------------
239+-- | Write a stream of arrays to a handle.
242+{-# INLINE writeArrays #-}
243+writeArrays :: (MonadIO m, Storable a) => Handle -> SerialT m (Array a) -> m ()
244+writeArrays h m = S.mapM_ (liftIO . writeArray h) m
246+-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
249+-- Note that if you use a chunk size less than 8K (GHC's default buffer
250+-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
251+-- do not want buffering to occur at GHC level as well. Same thing applies to
254+-- | Like 'write' but provides control over the write buffer. Output will
255+-- be written to the IO device as soon as we collect the specified number of
259+{-# INLINE writeByChunks #-}
260+writeByChunks :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
261+writeByChunks n h m = writeArrays h $ A.arraysOf n m
263+-- > write = 'writeByChunks' A.defaultChunkSize
265+-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
266+-- up to 'A.defaultChunkSize' before writing. Note that the write behavior
267+-- depends on the 'IOMode' and the current seek position of the handle.
271+write :: MonadIO m => Handle -> SerialT m Word8 -> m ()
272+write = writeByChunks A.defaultChunkSize
276+write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
277+write = toHandleWith A.defaultChunkSize
280+-------------------------------------------------------------------------------
281+-- IO with encoding/decoding Unicode characters
282+-------------------------------------------------------------------------------
286+-- > readUtf8 = decodeUtf8 . read
288+-- Read a UTF8 encoded stream of unicode characters from a file handle.
291+{-# INLINE readUtf8 #-}
292+readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
293+readUtf8 = decodeUtf8 . read
296+-- > writeUtf8 h s = write h $ encodeUtf8 s
298+-- Encode a stream of unicode characters to UTF8 and write it to the given file
299+-- handle. Default block buffering applies to the writes.
302+{-# INLINE writeUtf8 #-}
303+writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
304+writeUtf8 h s = write h $ encodeUtf8 s
306+-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
307+-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
308+-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
309+-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
313+{-# INLINE writeUtf8ByLines #-}
314+writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
315+writeUtf8ByLines = undefined
317+-- | Read UTF-8 lines from a file handle and apply the specified fold to each
318+-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
321+{-# INLINE readLines #-}
322+readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
323+readLines h f = foldLines (readUtf8 h) f
325+-------------------------------------------------------------------------------
326+-- Framing on a sequence
327+-------------------------------------------------------------------------------
329+-- | Read a stream from a file handle and split it into frames delimited by
330+-- the specified sequence of elements. The supplied fold is applied on each
334+{-# INLINE readFrames #-}
335+readFrames :: (IsStream t, MonadIO m, Storable a)
336+ => Array a -> Handle -> Fold m a b -> t m b
337+readFrames = undefined -- foldFrames . read
339+-- | Write a stream to the given file handle buffering up to frames separated
340+-- by the given sequence or up to a maximum of @defaultChunkSize@.
343+{-# INLINE writeByFrames #-}
344+writeByFrames :: (IsStream t, MonadIO m, Storable a)
345+ => Array a -> Handle -> t m a -> m ()
346+writeByFrames = undefined
348+-------------------------------------------------------------------------------
349+-- Random Access IO (Seek)
350+-------------------------------------------------------------------------------
352+-- XXX handles could be shared, so we may not want to use the handle state at
353+-- all for these APIs. we can use pread and pwrite instead. On windows we will
354+-- need to use readFile/writeFile with an offset argument.
356+-------------------------------------------------------------------------------
358+-- | Read the element at the given index treating the file as an array.
361+{-# INLINE readIndex #-}
362+readIndex :: Storable a => Handle -> Int -> Maybe a
363+readIndex arr i = undefined
365+-- NOTE: To represent a range to read we have chosen (start, size) instead of
366+-- (start, end). This helps in removing the ambiguity of whether "end" is
367+-- included in the range or not.
369+-- We could avoid specifying the range to be read and instead use "take size"
370+-- on the stream, but it may end up reading more and then consume it partially.
372+-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
373+-- from @handle@ starting at the offset @pos@ from the beginning of the file.
375+-- Reads are performed in chunks of size @chunkSize@. For block devices, to
376+-- avoid reading partial blocks @chunkSize@ must align with the block size of
377+-- the underlying device. If the underlying block size is unknown, it is a good
378+-- idea to keep it a multiple 4KiB. This API ensures that the start of each
379+-- chunk is aligned with @chunkSize@ from second chunk onwards.
381+{-# INLINE readSliceWith #-}
382+readSliceWith :: (IsStream t, MonadIO m, Storable a)
383+ => Int -> Handle -> Int -> Int -> t m a
384+readSliceWith chunkSize h pos len = undefined
386+-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
387+-- at index @i@ and reading up to @count@ elements in the forward direction
388+-- ending at the index @i + count - 1@.
391+{-# INLINE readSlice #-}
392+readSlice :: (IsStream t, MonadIO m, Storable a)
393+ => Handle -> Int -> Int -> t m a
394+readSlice = readSliceWith A.defaultChunkSize
396+-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
397+-- at index @i@ and reading up to @count@ elements in the reverse direction
398+-- ending at the index @i - count + 1@.
401+{-# INLINE readSliceRev #-}
402+readSliceRev :: (IsStream t, MonadIO m, Storable a)
403+ => Handle -> Int -> Int -> t m a
404+readSliceRev h i count = undefined
406+-- | Write the given element at the given index in the file.
409+{-# INLINE writeIndex #-}
410+writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
411+writeIndex h i a = undefined
413+-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
414+-- starting at index @i@ and writing up to @count@ elements in the forward
415+-- direction ending at the index @i + count - 1@.
418+{-# INLINE writeSlice #-}
419+writeSlice :: (IsStream t, Monad m, Storable a)
420+ => Handle -> Int -> Int -> t m a -> m ()
421+writeSlice h i len s = undefined
423+-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
424+-- starting at index @i@ and writing up to @count@ elements in the reverse
425+-- direction ending at the index @i - count + 1@.
428+{-# INLINE writeSliceRev #-}
429+writeSliceRev :: (IsStream t, Monad m, Storable a)
430+ => Handle -> Int -> Int -> t m a -> m ()
431+writeSliceRev arr i len s = undefined