@@ -0,0 +1,500 @@
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE BangPatterns #-}
3+{-# LANGUAGE FlexibleContexts #-}
4+{-# LANGUAGE MagicHash #-}
5+{-# LANGUAGE RecordWildCards #-}
6+{-# LANGUAGE UnboxedTuples #-}
7+
8+#include "inline.hs"
9+
10+-- |
11+-- Module : Streamly.FileSystem.File
12+-- Copyright : (c) 2019 Harendra Kumar
13+--
14+-- License : BSD3
15+-- Maintainer : harendra.kumar@gmail.com
16+-- Stability : experimental
17+-- Portability : GHC
18+--
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
25+-- leakage.
26+--
27+-- > import qualified Streamly.FileSystem.File as File
28+--
29+
30+module Streamly.FileSystem.File
31+ (
32+ -- * Streaming IO
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.
42+
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.
49+
50+ -- ** File IO Using Handle
51+ withFile
52+
53+ -- ** Read From File
54+ , read
55+ -- , readShared
56+ -- , readTailForever
57+
58+ -- , readUtf8
59+ -- , readLines
60+ -- , readFrames
61+ -- , readByChunks
62+
63+ -- -- * Array Read
64+ -- , readArrayOf
65+
66+ , readArraysUpto
67+ -- , readArraysOf
68+ , readArrays
69+
70+ -- ** Write To File
71+ , write
72+ -- , writeUtf8
73+ -- , writeUtf8ByLines
74+ -- , writeByFrames
75+ , writeByChunks
76+
77+ -- -- * Array Write
78+ , writeArray
79+ , writeArrays
80+
81+ -- ** Append To File
82+ , append
83+ , appendByChunks
84+ -- , appendShared
85+ , appendArray
86+ , appendArrays
87+
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.
93+ --
94+ -- , readIndex
95+ -- , readSlice
96+ -- , readSliceRev
97+ -- , readAt -- read from a given position to th end of file
98+ -- , readSliceArrayUpto
99+ -- , readSliceArrayOf
100+
101+ -- , writeIndex
102+ -- , writeSlice
103+ -- , writeSliceRev
104+ -- , writeAt -- start writing at the given position
105+ -- , writeSliceArray
106+ )
107+where
108+
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)
115+
116+import qualified System.IO as SIO
117+
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)
124+
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
129+
130+-------------------------------------------------------------------------------
131+-- References
132+-------------------------------------------------------------------------------
133+--
134+-- The following references may be useful to build an understanding about the
135+-- file API design:
136+--
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.
142+
143+-------------------------------------------------------------------------------
144+-- Safe file reading
145+-------------------------------------------------------------------------------
146+
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
152+-- raised by 'act'.
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)
156+
157+-------------------------------------------------------------------------------
158+-- Array IO (Input)
159+-------------------------------------------------------------------------------
160+
161+-- TODO readArrayOf
162+
163+-------------------------------------------------------------------------------
164+-- Array IO (output)
165+-------------------------------------------------------------------------------
166+
167+-- | Write an array to a file. Overwrites the file if it exists.
168+--
169+-- @since 0.7.0
170+{-# INLINABLE writeArray #-}
171+writeArray :: Storable a => FilePath -> Array a -> IO ()
172+writeArray file arr = SIO.withFile file WriteMode (\h -> FH.writeArray h arr)
173+
174+-- | append an array to a file.
175+--
176+-- @since 0.7.0
177+{-# INLINABLE appendArray #-}
178+appendArray :: Storable a => FilePath -> Array a -> IO ()
179+appendArray file arr = SIO.withFile file AppendMode (\h -> FH.writeArray h arr)
180+
181+-------------------------------------------------------------------------------
182+-- Stream of Arrays IO
183+-------------------------------------------------------------------------------
184+
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)
192+
193+-- XXX read 'Array a' instead of Word8
194+--
195+-- | @readArrays file@ reads a stream of arrays from file @file@.
196+-- The maximum size of a single array is limited to @defaultChunkSize@.
197+--
198+-- > readArrays = readArraysUpto defaultChunkSize
199+--
200+-- @since 0.7.0
201+{-# INLINE readArrays #-}
202+readArrays :: (IsStream t, MonadCatch m, MonadIO m)
203+ => FilePath -> t m (Array Word8)
204+readArrays = readArraysUpto A.defaultChunkSize
205+
206+-------------------------------------------------------------------------------
207+-- Read File to Stream
208+-------------------------------------------------------------------------------
209+
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.
213+
214+{-
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.
218+--
219+{-# INLINE readByChunksUpto #-}
220+readByChunksUpto :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
221+readByChunksUpto chunkSize h = A.flattenArrays $ readArraysUpto chunkSize h
222+-}
223+
224+-- TODO
225+-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
226+--
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.
231+--
232+-- @since 0.7.0
233+{-# INLINE read #-}
234+read :: (IsStream t, MonadCatch m, MonadIO m) => FilePath -> t m Word8
235+read file = withFile file ReadMode FH.read
236+
237+{-
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.
241+--
242+-- @since 0.7.0
243+{-# INLINE readShared #-}
244+readShared :: (IsStream t, MonadIO m) => Handle -> t m Word8
245+readShared = undefined
246+
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
249+-- as it is written.
250+--
251+-- @since 0.7.0
252+{-# INLINE readTailForever #-}
253+readTailForever :: (IsStream t, MonadIO m) => Handle -> t m Word8
254+readTailForever = undefined
255+-}
256+
257+-------------------------------------------------------------------------------
258+-- Writing
259+-------------------------------------------------------------------------------
260+
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)
266+
267+-- | Write a stream of arrays to a file. Overwrites the file if it exists.
268+--
269+-- @since 0.7.0
270+{-# INLINE writeArrays #-}
271+writeArrays :: (MonadAsync m, MonadCatch m, Storable a)
272+ => FilePath -> SerialT m (Array a) -> m ()
273+writeArrays = writeArraysMode WriteMode
274+
275+-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
276+--
277+-- XXX test this
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
281+-- writes as well.
282+
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
285+-- input elements.
286+--
287+-- @since 0.7.0
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
292+
293+-- > write = 'writeByChunks' A.defaultChunkSize
294+--
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.
299+--
300+-- @since 0.7.0
301+{-# INLINE write #-}
302+write :: (MonadAsync m, MonadCatch m) => FilePath -> SerialT m Word8 -> m ()
303+write = writeByChunks A.defaultChunkSize
304+
305+{-
306+{-# INLINE write #-}
307+write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
308+write = toHandleWith A.defaultChunkSize
309+-}
310+
311+-- | Append a stream of arrays to a file.
312+--
313+-- @since 0.7.0
314+{-# INLINE appendArrays #-}
315+appendArrays :: (MonadAsync m, MonadCatch m, Storable a)
316+ => FilePath -> SerialT m (Array a) -> m ()
317+appendArrays = writeArraysMode AppendMode
318+
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
321+-- input elements.
322+--
323+-- @since 0.7.0
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
328+
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.
333+--
334+-- @since 0.7.0
335+{-# INLINE append #-}
336+append :: (MonadAsync m, MonadCatch m) => FilePath -> SerialT m Word8 -> m ()
337+append = appendByChunks A.defaultChunkSize
338+
339+{-
340+-- | Like 'append' but the file is not locked for exclusive writes.
341+--
342+-- @since 0.7.0
343+{-# INLINE appendShared #-}
344+appendShared :: MonadIO m => Handle -> SerialT m Word8 -> m ()
345+appendShared = undefined
346+-}
347+
348+-------------------------------------------------------------------------------
349+-- IO with encoding/decoding Unicode characters
350+-------------------------------------------------------------------------------
351+
352+{-
353+-- |
354+-- > readUtf8 = decodeUtf8 . read
355+--
356+-- Read a UTF8 encoded stream of unicode characters from a file handle.
357+--
358+-- @since 0.7.0
359+{-# INLINE readUtf8 #-}
360+readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
361+readUtf8 = decodeUtf8 . read
362+
363+-- |
364+-- > writeUtf8 h s = write h $ encodeUtf8 s
365+--
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.
368+--
369+-- @since 0.7.0
370+{-# INLINE writeUtf8 #-}
371+writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
372+writeUtf8 h s = write h $ encodeUtf8 s
373+
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'
378+-- option.
379+--
380+-- @since 0.7.0
381+{-# INLINE writeUtf8ByLines #-}
382+writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
383+writeUtf8ByLines = undefined
384+
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.
387+--
388+-- @since 0.7.0
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
392+
393+-------------------------------------------------------------------------------
394+-- Framing on a sequence
395+-------------------------------------------------------------------------------
396+
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
399+-- frame.
400+--
401+-- @since 0.7.0
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
406+
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@.
409+--
410+-- @since 0.7.0
411+{-# INLINE writeByFrames #-}
412+writeByFrames :: (IsStream t, MonadIO m, Storable a)
413+ => Array a -> Handle -> t m a -> m ()
414+writeByFrames = undefined
415+
416+-------------------------------------------------------------------------------
417+-- Random Access IO (Seek)
418+-------------------------------------------------------------------------------
419+
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.
423+
424+-------------------------------------------------------------------------------
425+
426+-- | Read the element at the given index treating the file as an array.
427+--
428+-- @since 0.7.0
429+{-# INLINE readIndex #-}
430+readIndex :: Storable a => Handle -> Int -> Maybe a
431+readIndex arr i = undefined
432+
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.
436+--
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.
439+
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.
442+--
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.
448+--
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
453+
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@.
457+--
458+-- @since 0.7.0
459+{-# INLINE readSlice #-}
460+readSlice :: (IsStream t, MonadIO m, Storable a)
461+ => Handle -> Int -> Int -> t m a
462+readSlice = readSliceWith A.defaultChunkSize
463+
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@.
467+--
468+-- @since 0.7.0
469+{-# INLINE readSliceRev #-}
470+readSliceRev :: (IsStream t, MonadIO m, Storable a)
471+ => Handle -> Int -> Int -> t m a
472+readSliceRev h i count = undefined
473+
474+-- | Write the given element at the given index in the file.
475+--
476+-- @since 0.7.0
477+{-# INLINE writeIndex #-}
478+writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
479+writeIndex h i a = undefined
480+
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@.
484+--
485+-- @since 0.7.0
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
490+
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@.
494+--
495+-- @since 0.7.0
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
500+-}