@@ -0,0 +1,500 @@
2+{-# LANGUAGE BangPatterns #-}
3+{-# LANGUAGE FlexibleContexts #-}
4+{-# LANGUAGE MagicHash #-}
5+{-# LANGUAGE RecordWildCards #-}
6+{-# LANGUAGE UnboxedTuples #-}
11+-- Module : Streamly.FileSystem.File
12+-- Copyright : (c) 2019 Harendra Kumar
15+-- Maintainer : harendra.kumar@gmail.com
16+-- Stability : experimental
19+-- Read and write streams and arrays to and from files specified by their paths
20+-- in the file system. Unlike the handle based APIs which can have a read/write
21+-- session consisting of multiple reads and writes to the handle, these APIs
22+-- are one shot read or write APIs. These APIs open the file handle, perform
23+-- the requested operation and close the handle. Thease are safer compared to
24+-- the handle based APIs as there is no possiblity of a file descriptor
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+ -- ** File IO Using Handle
73+ -- , writeUtf8ByLines
88+ -- -- * Random Access (Seek)
89+ -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or
90+ -- files that have random access or seek capability. This type of devices
91+ -- include disks, files, memory devices and exclude terminals, pipes,
92+ -- sockets and fifos.
97+ -- , readAt -- read from a given position to th end of file
98+ -- , readSliceArrayUpto
99+ -- , readSliceArrayOf
104+ -- , writeAt -- start writing at the given position
105+ -- , writeSliceArray
109+import Control.Monad.Catch (MonadCatch)
110+import Control.Monad.IO.Class (MonadIO(..))
111+import Data.Word (Word8)
112+import Foreign.Storable (Storable(..))
113+import System.IO (Handle, openFile, IOMode(..), hClose)
114+import Prelude hiding (read)
116+import qualified System.IO as SIO
118+import Streamly.Mem.Array.Types (Array(..))
119+import Streamly.Streams.Serial (SerialT)
120+import Streamly.Streams.StreamK.Type (IsStream)
121+import Streamly.SVar (MonadAsync)
122+-- import Streamly.Fold (Fold)
123+-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
125+import qualified Streamly.FileSystem.Handle as FH
126+import qualified Streamly.Mem.Array as A
127+import qualified Streamly.Mem.Array.Types as A hiding (flattenArrays)
128+import qualified Streamly.Prelude as S
130+-------------------------------------------------------------------------------
132+-------------------------------------------------------------------------------
134+-- The following references may be useful to build an understanding about the
137+-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
138+-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
139+-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
140+-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
141+-- https://www.w3.org/TR/FileAPI/ for http file API.
143+-------------------------------------------------------------------------------
145+-------------------------------------------------------------------------------
147+-- | @'withFile' name mode act@ opens a file using 'openFile' and passes
148+-- the resulting handle to the computation @act@. The handle will be
149+-- closed on exit from 'withFile', whether by normal termination or by
150+-- raising an exception. If closing the handle raises an exception, then
151+-- this exception will be raised by 'withFile' rather than any exception
153+withFile :: (IsStream t, MonadCatch m, MonadIO m)
154+ => FilePath -> IOMode -> (Handle -> t m a) -> t m a
155+withFile file mode = S.bracket (liftIO $ openFile file mode) (liftIO . hClose)
157+-------------------------------------------------------------------------------
159+-------------------------------------------------------------------------------
163+-------------------------------------------------------------------------------
165+-------------------------------------------------------------------------------
167+-- | Write an array to a file. Overwrites the file if it exists.
170+{-# INLINABLE writeArray #-}
171+writeArray :: Storable a => FilePath -> Array a -> IO ()
172+writeArray file arr = SIO.withFile file WriteMode (\h -> FH.writeArray h arr)
174+-- | append an array to a file.
177+{-# INLINABLE appendArray #-}
178+appendArray :: Storable a => FilePath -> Array a -> IO ()
179+appendArray file arr = SIO.withFile file AppendMode (\h -> FH.writeArray h arr)
181+-------------------------------------------------------------------------------
182+-- Stream of Arrays IO
183+-------------------------------------------------------------------------------
185+-- | @readArraysUpto size file@ reads a stream of arrays from file @file@.
186+-- The maximum size of a single array is specified by @size@. The actual size
187+-- read may be less than or equal to @size@.
188+{-# INLINABLE readArraysUpto #-}
189+readArraysUpto :: (IsStream t, MonadCatch m, MonadIO m)
190+ => Int -> FilePath -> t m (Array Word8)
191+readArraysUpto size file = withFile file ReadMode (FH.readArraysUpto size)
193+-- XXX read 'Array a' instead of Word8
195+-- | @readArrays file@ reads a stream of arrays from file @file@.
196+-- The maximum size of a single array is limited to @defaultChunkSize@.
198+-- > readArrays = readArraysUpto defaultChunkSize
201+{-# INLINE readArrays #-}
202+readArrays :: (IsStream t, MonadCatch m, MonadIO m)
203+ => FilePath -> t m (Array Word8)
204+readArrays = readArraysUpto A.defaultChunkSize
206+-------------------------------------------------------------------------------
207+-- Read File to Stream
208+-------------------------------------------------------------------------------
210+-- TODO for concurrent streams implement readahead IO. We can send multiple
211+-- read requests at the same time. For serial case we can use async IO. We can
212+-- also control the read throughput in mbps or IOPS.
215+-- | @readByChunksUpto chunkSize handle@ reads a byte stream from a file
216+-- handle, reads are performed in chunks of up to @chunkSize@. The stream ends
217+-- as soon as EOF is encountered.
219+{-# INLINE readByChunksUpto #-}
220+readByChunksUpto :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
221+readByChunksUpto chunkSize h = A.flattenArrays $ readArraysUpto chunkSize h
225+-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
227+-- > read = 'readByChunks' A.defaultChunkSize
228+-- | Generate a stream of elements of the given type from a file specified by
229+-- path. The stream ends when EOF is encountered. File is locked using multiple
230+-- reader and single writer locking mode.
234+read :: (IsStream t, MonadCatch m, MonadIO m) => FilePath -> t m Word8
235+read file = withFile file ReadMode FH.read
238+-- | Generate a stream of elements of the given type from a file 'Handle'. The
239+-- stream ends when EOF is encountered. File is not locked for exclusive reads,
240+-- writers can keep writing to the file.
243+{-# INLINE readShared #-}
244+readShared :: (IsStream t, MonadIO m) => Handle -> t m Word8
245+readShared = undefined
247+-- | Read a stream from a given file path. When end of file (EOF) is reached
248+-- this API waits for more data to be written to the file and keeps reading it
252+{-# INLINE readTailForever #-}
253+readTailForever :: (IsStream t, MonadIO m) => Handle -> t m Word8
254+readTailForever = undefined
257+-------------------------------------------------------------------------------
259+-------------------------------------------------------------------------------
261+{-# INLINE writeArraysMode #-}
262+writeArraysMode :: (MonadAsync m, MonadCatch m, Storable a)
263+ => IOMode -> FilePath -> SerialT m (Array a) -> m ()
264+writeArraysMode mode file xs = S.drain $
265+ withFile file mode (\h -> S.mapM (liftIO . FH.writeArray h) xs)
267+-- | Write a stream of arrays to a file. Overwrites the file if it exists.
270+{-# INLINE writeArrays #-}
271+writeArrays :: (MonadAsync m, MonadCatch m, Storable a)
272+ => FilePath -> SerialT m (Array a) -> m ()
273+writeArrays = writeArraysMode WriteMode
275+-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
278+-- Note that if you use a chunk size less than 8K (GHC's default buffer
279+-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
280+-- do not want buffering to occur at GHC level as well. Same thing applies to
283+-- | Like 'write' but provides control over the write buffer. Output will
284+-- be written to the IO device as soon as we collect the specified number of
288+{-# INLINE writeByChunks #-}
289+writeByChunks :: (MonadAsync m, MonadCatch m)
290+ => Int -> FilePath -> SerialT m Word8 -> m ()
291+writeByChunks n file xs = writeArrays file $ A.arraysOf n xs
293+-- > write = 'writeByChunks' A.defaultChunkSize
295+-- | Write a byte stream to a file. Combines the bytes in chunks of size
296+-- up to 'A.defaultChunkSize' before writing. If the file exists it is
297+-- truncated to zero size before writing. If the file does not exist it is
298+-- created. File is locked using single writer locking mode.
302+write :: (MonadAsync m, MonadCatch m) => FilePath -> SerialT m Word8 -> m ()
303+write = writeByChunks A.defaultChunkSize
307+write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
308+write = toHandleWith A.defaultChunkSize
311+-- | Append a stream of arrays to a file.
314+{-# INLINE appendArrays #-}
315+appendArrays :: (MonadAsync m, MonadCatch m, Storable a)
316+ => FilePath -> SerialT m (Array a) -> m ()
317+appendArrays = writeArraysMode AppendMode
319+-- | Like 'append' but provides control over the write buffer. Output will
320+-- be written to the IO device as soon as we collect the specified number of
324+{-# INLINE appendByChunks #-}
325+appendByChunks :: (MonadAsync m, MonadCatch m)
326+ => Int -> FilePath -> SerialT m Word8 -> m ()
327+appendByChunks n file xs = appendArrays file $ A.arraysOf n xs
329+-- | Append a byte stream to a file. Combines the bytes in chunks of size up to
330+-- 'A.defaultChunkSize' before writing. If the file exists then the new data
331+-- is appended to the file. If the file does not exist it is created. File is
332+-- locked using single writer locking mode.
335+{-# INLINE append #-}
336+append :: (MonadAsync m, MonadCatch m) => FilePath -> SerialT m Word8 -> m ()
337+append = appendByChunks A.defaultChunkSize
340+-- | Like 'append' but the file is not locked for exclusive writes.
343+{-# INLINE appendShared #-}
344+appendShared :: MonadIO m => Handle -> SerialT m Word8 -> m ()
345+appendShared = undefined
348+-------------------------------------------------------------------------------
349+-- IO with encoding/decoding Unicode characters
350+-------------------------------------------------------------------------------
354+-- > readUtf8 = decodeUtf8 . read
356+-- Read a UTF8 encoded stream of unicode characters from a file handle.
359+{-# INLINE readUtf8 #-}
360+readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
361+readUtf8 = decodeUtf8 . read
364+-- > writeUtf8 h s = write h $ encodeUtf8 s
366+-- Encode a stream of unicode characters to UTF8 and write it to the given file
367+-- handle. Default block buffering applies to the writes.
370+{-# INLINE writeUtf8 #-}
371+writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
372+writeUtf8 h s = write h $ encodeUtf8 s
374+-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
375+-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
376+-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
377+-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
381+{-# INLINE writeUtf8ByLines #-}
382+writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
383+writeUtf8ByLines = undefined
385+-- | Read UTF-8 lines from a file handle and apply the specified fold to each
386+-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
389+{-# INLINE readLines #-}
390+readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
391+readLines h f = foldLines (readUtf8 h) f
393+-------------------------------------------------------------------------------
394+-- Framing on a sequence
395+-------------------------------------------------------------------------------
397+-- | Read a stream from a file handle and split it into frames delimited by
398+-- the specified sequence of elements. The supplied fold is applied on each
402+{-# INLINE readFrames #-}
403+readFrames :: (IsStream t, MonadIO m, Storable a)
404+ => Array a -> Handle -> Fold m a b -> t m b
405+readFrames = undefined -- foldFrames . read
407+-- | Write a stream to the given file handle buffering up to frames separated
408+-- by the given sequence or up to a maximum of @defaultChunkSize@.
411+{-# INLINE writeByFrames #-}
412+writeByFrames :: (IsStream t, MonadIO m, Storable a)
413+ => Array a -> Handle -> t m a -> m ()
414+writeByFrames = undefined
416+-------------------------------------------------------------------------------
417+-- Random Access IO (Seek)
418+-------------------------------------------------------------------------------
420+-- XXX handles could be shared, so we may not want to use the handle state at
421+-- all for these APIs. we can use pread and pwrite instead. On windows we will
422+-- need to use readFile/writeFile with an offset argument.
424+-------------------------------------------------------------------------------
426+-- | Read the element at the given index treating the file as an array.
429+{-# INLINE readIndex #-}
430+readIndex :: Storable a => Handle -> Int -> Maybe a
431+readIndex arr i = undefined
433+-- NOTE: To represent a range to read we have chosen (start, size) instead of
434+-- (start, end). This helps in removing the ambiguity of whether "end" is
435+-- included in the range or not.
437+-- We could avoid specifying the range to be read and instead use "take size"
438+-- on the stream, but it may end up reading more and then consume it partially.
440+-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
441+-- from @handle@ starting at the offset @pos@ from the beginning of the file.
443+-- Reads are performed in chunks of size @chunkSize@. For block devices, to
444+-- avoid reading partial blocks @chunkSize@ must align with the block size of
445+-- the underlying device. If the underlying block size is unknown, it is a good
446+-- idea to keep it a multiple 4KiB. This API ensures that the start of each
447+-- chunk is aligned with @chunkSize@ from second chunk onwards.
449+{-# INLINE readSliceWith #-}
450+readSliceWith :: (IsStream t, MonadIO m, Storable a)
451+ => Int -> Handle -> Int -> Int -> t m a
452+readSliceWith chunkSize h pos len = undefined
454+-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
455+-- at index @i@ and reading up to @count@ elements in the forward direction
456+-- ending at the index @i + count - 1@.
459+{-# INLINE readSlice #-}
460+readSlice :: (IsStream t, MonadIO m, Storable a)
461+ => Handle -> Int -> Int -> t m a
462+readSlice = readSliceWith A.defaultChunkSize
464+-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
465+-- at index @i@ and reading up to @count@ elements in the reverse direction
466+-- ending at the index @i - count + 1@.
469+{-# INLINE readSliceRev #-}
470+readSliceRev :: (IsStream t, MonadIO m, Storable a)
471+ => Handle -> Int -> Int -> t m a
472+readSliceRev h i count = undefined
474+-- | Write the given element at the given index in the file.
477+{-# INLINE writeIndex #-}
478+writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
479+writeIndex h i a = undefined
481+-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
482+-- starting at index @i@ and writing up to @count@ elements in the forward
483+-- direction ending at the index @i + count - 1@.
486+{-# INLINE writeSlice #-}
487+writeSlice :: (IsStream t, Monad m, Storable a)
488+ => Handle -> Int -> Int -> t m a -> m ()
489+writeSlice h i len s = undefined
491+-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
492+-- starting at index @i@ and writing up to @count@ elements in the reverse
493+-- direction ending at the index @i - count + 1@.
496+{-# INLINE writeSliceRev #-}
497+writeSliceRev :: (IsStream t, Monad m, Storable a)
498+ => Handle -> Int -> Int -> t m a -> m ()
499+writeSliceRev arr i len s = undefined