@@ -0,0 +1,432 @@
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE BangPatterns #-}
3+{-# LANGUAGE MagicHash #-}
4+{-# LANGUAGE RecordWildCards #-}
5+{-# LANGUAGE UnboxedTuples #-}
6+
7+#include "../Streams/inline.hs"
8+
9+-- |
10+-- Module : Streamly.FileSystem.File
11+-- Copyright : (c) 2018 Harendra Kumar
12+--
13+-- License : BSD3
14+-- Maintainer : harendra.kumar@gmail.com
15+-- Stability : experimental
16+-- Portability : GHC
17+--
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.
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+ -- TODO file path based APIs
51+ -- , readFile
52+ -- , writeFile
53+
54+ -- ** Read File to Stream
55+ read
56+ -- , readUtf8
57+ -- , readLines
58+ -- , readFrames
59+ -- , readByChunks
60+
61+ -- -- * Array Read
62+ -- , readArrayUpto
63+ -- , readArrayOf
64+
65+ -- , readArraysUpto
66+ -- , readArraysOf
67+ , readArrays
68+
69+ -- ** Write File from Stream
70+ , write
71+ -- , writeUtf8
72+ -- , writeUtf8ByLines
73+ -- , writeByFrames
74+ , writeByChunks
75+
76+ -- -- * Array Write
77+ , writeArray
78+ , writeArrays
79+
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.
85+ --
86+ -- , readIndex
87+ -- , readSlice
88+ -- , readSliceRev
89+ -- , readAt -- read from a given position to th end of file
90+ -- , readSliceArrayUpto
91+ -- , readSliceArrayOf
92+
93+ -- , writeIndex
94+ -- , writeSlice
95+ -- , writeSliceRev
96+ -- , writeAt -- start writing at the given position
97+ -- , writeSliceArray
98+ )
99+where
100+
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)
110+
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)
116+
117+import qualified Streamly.Array as A
118+import qualified Streamly.Array.Types as A hiding (flattenArrays)
119+import qualified Streamly.Prelude as S
120+
121+-------------------------------------------------------------------------------
122+-- References
123+-------------------------------------------------------------------------------
124+--
125+-- The following references may be useful to build an understanding about the
126+-- file API design:
127+--
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.
133+
134+-------------------------------------------------------------------------------
135+-- Array IO (Input)
136+-------------------------------------------------------------------------------
137+
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
149+ let v = Array
150+ { aStart = ptr
151+ , aEnd = p `plusPtr` n
152+ , aBound = p `plusPtr` size
153+ }
154+ -- XXX shrink only if the diff is significant
155+ A.shrinkToFit v
156+
157+-------------------------------------------------------------------------------
158+-- Array IO (output)
159+-------------------------------------------------------------------------------
160+
161+-- | Write an Array to a file handle.
162+--
163+-- @since 0.7.0
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
168+ where
169+ aLen =
170+ let p = unsafeForeignPtrToPtr aStart
171+ in aEnd `minusPtr` p
172+
173+-------------------------------------------------------------------------------
174+-- Stream of Arrays IO
175+-------------------------------------------------------------------------------
176+
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'
180+-- on the 'Handle'.
181+{-# INLINABLE readArraysUpto #-}
182+readArraysUpto :: (IsStream t, MonadIO m)
183+ => Int -> Handle -> t m (Array Word8)
184+readArraysUpto size h = go
185+ where
186+ -- XXX use cons/nil instead
187+ go = mkStream $ \_ yld sng _ -> do
188+ arr <- liftIO $ readArrayUpto size h
189+ if A.length arr < size
190+ then sng arr
191+ else yld arr go
192+
193+-- XXX read 'Array a' instead of Word8
194+--
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'
198+-- on the 'Handle'.
199+--
200+-- @since 0.7.0
201+{-# INLINE readArrays #-}
202+readArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
203+readArrays = readArraysUpto A.defaultChunkSize
204+
205+-------------------------------------------------------------------------------
206+-- Read File to Stream
207+-------------------------------------------------------------------------------
208+
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.
212+
213+{-
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.
217+--
218+{-# INLINE readByChunksUpto #-}
219+readByChunksUpto :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
220+readByChunksUpto chunkSize h = A.flattenArrays $ readArraysUpto chunkSize h
221+-}
222+
223+-- TODO
224+-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
225+--
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.
229+--
230+-- @since 0.7.0
231+{-# INLINE read #-}
232+read :: (IsStream t, MonadIO m) => Handle -> t m Word8
233+read = A.flattenArrays . readArrays
234+
235+-------------------------------------------------------------------------------
236+-- Writing
237+-------------------------------------------------------------------------------
238+
239+-- | Write a stream of arrays to a handle.
240+--
241+-- @since 0.7.0
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
245+
246+-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
247+--
248+-- XXX test this
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
252+-- writes as well.
253+
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
256+-- input elements.
257+--
258+-- @since 0.7.0
259+{-# INLINE writeByChunks #-}
260+writeByChunks :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
261+writeByChunks n h m = writeArrays h $ A.arraysOf n m
262+
263+-- > write = 'writeByChunks' A.defaultChunkSize
264+--
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.
268+--
269+-- @since 0.7.0
270+{-# INLINE write #-}
271+write :: MonadIO m => Handle -> SerialT m Word8 -> m ()
272+write = writeByChunks A.defaultChunkSize
273+
274+{-
275+{-# INLINE write #-}
276+write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
277+write = toHandleWith A.defaultChunkSize
278+-}
279+
280+-------------------------------------------------------------------------------
281+-- IO with encoding/decoding Unicode characters
282+-------------------------------------------------------------------------------
283+
284+{-
285+-- |
286+-- > readUtf8 = decodeUtf8 . read
287+--
288+-- Read a UTF8 encoded stream of unicode characters from a file handle.
289+--
290+-- @since 0.7.0
291+{-# INLINE readUtf8 #-}
292+readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
293+readUtf8 = decodeUtf8 . read
294+
295+-- |
296+-- > writeUtf8 h s = write h $ encodeUtf8 s
297+--
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.
300+--
301+-- @since 0.7.0
302+{-# INLINE writeUtf8 #-}
303+writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
304+writeUtf8 h s = write h $ encodeUtf8 s
305+
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'
310+-- option.
311+--
312+-- @since 0.7.0
313+{-# INLINE writeUtf8ByLines #-}
314+writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
315+writeUtf8ByLines = undefined
316+
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.
319+--
320+-- @since 0.7.0
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
324+
325+-------------------------------------------------------------------------------
326+-- Framing on a sequence
327+-------------------------------------------------------------------------------
328+
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
331+-- frame.
332+--
333+-- @since 0.7.0
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
338+
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@.
341+--
342+-- @since 0.7.0
343+{-# INLINE writeByFrames #-}
344+writeByFrames :: (IsStream t, MonadIO m, Storable a)
345+ => Array a -> Handle -> t m a -> m ()
346+writeByFrames = undefined
347+
348+-------------------------------------------------------------------------------
349+-- Random Access IO (Seek)
350+-------------------------------------------------------------------------------
351+
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.
355+
356+-------------------------------------------------------------------------------
357+
358+-- | Read the element at the given index treating the file as an array.
359+--
360+-- @since 0.7.0
361+{-# INLINE readIndex #-}
362+readIndex :: Storable a => Handle -> Int -> Maybe a
363+readIndex arr i = undefined
364+
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.
368+--
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.
371+
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.
374+--
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.
380+--
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
385+
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@.
389+--
390+-- @since 0.7.0
391+{-# INLINE readSlice #-}
392+readSlice :: (IsStream t, MonadIO m, Storable a)
393+ => Handle -> Int -> Int -> t m a
394+readSlice = readSliceWith A.defaultChunkSize
395+
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@.
399+--
400+-- @since 0.7.0
401+{-# INLINE readSliceRev #-}
402+readSliceRev :: (IsStream t, MonadIO m, Storable a)
403+ => Handle -> Int -> Int -> t m a
404+readSliceRev h i count = undefined
405+
406+-- | Write the given element at the given index in the file.
407+--
408+-- @since 0.7.0
409+{-# INLINE writeIndex #-}
410+writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
411+writeIndex h i a = undefined
412+
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@.
416+--
417+-- @since 0.7.0
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
422+
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@.
426+--
427+-- @since 0.7.0
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
432+-}