@@ -0,0 +1,561 @@
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE BangPatterns #-}
3+{-# LANGUAGE MagicHash #-}
4+{-# LANGUAGE RecordWildCards #-}
5+{-# LANGUAGE UnboxedTuples #-}
6+
7+#include "inline.hs"
8+
9+-- |
10+-- Module : Streamly.FileSystem.FD
11+-- Copyright : (c) 2019 Composewell Technologies
12+--
13+-- License : BSD3
14+-- Maintainer : harendra.kumar@gmail.com
15+-- Stability : experimental
16+-- Portability : GHC
17+--
18+-- This module is a an experimental replacement for
19+-- "Streamly.FileSystem.Handle". The former module provides IO facilities based
20+-- on the GHC Handle type. The APIs in this module avoid the GHC handle layer
21+-- and provide more explicit control over buffering.
22+--
23+-- Read and write data as streams and arrays to and from files.
24+--
25+-- This module provides read and write APIs based on handles. Before reading or
26+-- writing, a file must be opened first using 'openFile'. The 'Handle' returned
27+-- by 'openFile' is then used to access the file. A 'Handle' is backed by an
28+-- operating system file descriptor. When the 'Handle' is garbage collected the
29+-- underlying file descriptor is automatically closed. A handle can be
30+-- explicitly closed using 'closeFile'.
31+--
32+-- Reading and writing APIs are divided into two categories, sequential
33+-- streaming APIs and random or seekable access APIs. File IO APIs are quite
34+-- similar to "Streamly.Mem.Array" read write APIs. In that regard, arrays can
35+-- be considered as in-memory files or files can be considered as on-disk
36+-- arrays.
37+--
38+-- > import qualified Streamly.FileSystem.FD as FD
39+--
40+
41+module Streamly.FileSystem.FD
42+ (
43+ -- * File Handles
44+ Handle
45+ , stdin
46+ , stdout
47+ , stderr
48+ , openFile
49+
50+ -- TODO file path based APIs
51+ -- , readFile
52+ -- , writeFile
53+
54+ -- * Streaming IO
55+ -- | Streaming APIs read or write data to or from a file or device
56+ -- sequentially, they never perform a seek to a random location. When
57+ -- reading, the stream is lazy and generated on-demand as the consumer
58+ -- consumes it. Read IO requests to the IO device are performed in chunks
59+ -- of 32KiB, this is referred to as @defaultChunkSize@ in the
60+ -- documentation. One IO request may or may not read the full chunk. If the
61+ -- whole stream is not consumed, it is possible that we may read slightly
62+ -- more from the IO device than what the consumer needed. Unless specified
63+ -- otherwise in the API, writes are collected into chunks of
64+ -- @defaultChunkSize@ before they are written to the IO device.
65+
66+ -- Streaming APIs work for all kind of devices, seekable or non-seekable;
67+ -- including disks, files, memory devices, terminals, pipes, sockets and
68+ -- fifos. While random access APIs work only for files or devices that have
69+ -- random access or seek capability for example disks, memory devices.
70+ -- Devices like terminals, pipes, sockets and fifos do not have random
71+ -- access capability.
72+
73+ -- ** Read File to Stream
74+ , read
75+ -- , readUtf8
76+ -- , readLines
77+ -- , readFrames
78+ , readByChunksUpto
79+
80+ -- -- * Array Read
81+ -- , readArrayUpto
82+ -- , readArrayOf
83+
84+ , readArrays
85+ , readArraysOfUpto
86+ -- , readArraysOf
87+
88+ -- ** Write File from Stream
89+ , write
90+ -- , writeUtf8
91+ -- , writeUtf8Lines
92+ -- , writeFrames
93+ , writeByChunksOf
94+
95+ -- -- * Array Write
96+ -- , writeArray
97+ , writeArrays
98+ , writeArraysOfUpto
99+ , writev
100+ , writevArraysOfUpto
101+
102+ -- -- * Random Access (Seek)
103+ -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or
104+ -- files that have random access or seek capability. This type of devices
105+ -- include disks, files, memory devices and exclude terminals, pipes,
106+ -- sockets and fifos.
107+ --
108+ -- , readIndex
109+ -- , readSlice
110+ -- , readSliceRev
111+ -- , readAt -- read from a given position to th end of file
112+ -- , readSliceArrayUpto
113+ -- , readSliceArrayOf
114+
115+ -- , writeIndex
116+ -- , writeSlice
117+ -- , writeSliceRev
118+ -- , writeAt -- start writing at the given position
119+ -- , writeSliceArray
120+ )
121+where
122+
123+import Control.Monad.IO.Class (MonadIO(..))
124+import Data.Word (Word8)
125+import Foreign.ForeignPtr (withForeignPtr)
126+-- import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
127+import Foreign.Ptr (plusPtr, castPtr)
128+import Foreign.Storable (Storable(..))
129+import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
130+-- import System.IO (Handle, hGetBufSome, hPutBuf)
131+import System.IO (IOMode)
132+import Prelude hiding (read)
133+
134+import qualified GHC.IO.FD as FD
135+import qualified GHC.IO.Device as RawIO hiding (write)
136+import qualified Streamly.FileSystem.FDIO as RawIO
137+
138+import Streamly.Mem.Array.Types (Array(..))
139+import Streamly.Streams.Serial (SerialT)
140+import Streamly.Streams.StreamD (toStreamD)
141+import Streamly.Streams.StreamD.Type (fromStreamD)
142+import Streamly.Streams.StreamK.Type (IsStream, mkStream)
143+import Streamly.Mem.Array.Types (byteLength, defaultChunkSize, groupIOVecsOf)
144+-- import Streamly.Fold (Fold)
145+-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
146+
147+import qualified Streamly.Mem.Array as A
148+import qualified Streamly.Prelude as S
149+import qualified Streamly.Streams.StreamD.Type as D
150+
151+-------------------------------------------------------------------------------
152+-- References
153+-------------------------------------------------------------------------------
154+--
155+-- The following references may be useful to build an understanding about the
156+-- file API design:
157+--
158+-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
159+-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
160+-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
161+-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
162+-- https://www.w3.org/TR/FileAPI/ for http file API.
163+
164+-------------------------------------------------------------------------------
165+-- Handles
166+-------------------------------------------------------------------------------
167+
168+-- XXX attach a finalizer
169+-- | A 'Handle' is returned by 'openFile' and is subsequently used to perform
170+-- read and write operations on a file.
171+--
172+newtype Handle = Handle FD.FD
173+
174+-- | File handle for standard input
175+stdin :: Handle
176+stdin = Handle FD.stdin
177+
178+-- | File handle for standard output
179+stdout :: Handle
180+stdout = Handle FD.stdout
181+
182+-- | File handle for standard error
183+stderr :: Handle
184+stderr = Handle FD.stderr
185+
186+-- XXX we can support all the flags that the "open" system call supports.
187+-- Instead of using RTS locking mechanism can we use system provided locking
188+-- instead?
189+--
190+-- | Open a file that is not a directory and return a file handle.
191+-- 'openFile' enforces a multiple-reader single-writer locking on files. That
192+-- is, there may either be many handles on the same file which manage input, or
193+-- just one handle on the file which manages output. If any open handle is
194+-- managing a file for output, no new handle can be allocated for that file. If
195+-- any open handle is managing a file for input, new handles can only be
196+-- allocated if they do not manage output. Whether two files are the same is
197+-- implementation-dependent, but they should normally be the same if they have
198+-- the same absolute path name and neither has been renamed, for example.
199+--
200+openFile :: FilePath -> IOMode -> IO Handle
201+openFile path mode = fmap (Handle . fst) $ FD.openFile path mode True
202+
203+-------------------------------------------------------------------------------
204+-- Array IO (Input)
205+-------------------------------------------------------------------------------
206+
207+-- | Read a 'ByteArray' from a file handle. If no data is available on the
208+-- handle it blocks until some data becomes available. If data is available
209+-- then it immediately returns that data without blocking. It reads a maximum
210+-- of up to the size requested.
211+{-# INLINABLE readArrayUpto #-}
212+readArrayUpto :: Int -> Handle -> IO (Array Word8)
213+readArrayUpto size (Handle fd) = do
214+ ptr <- mallocPlainForeignPtrBytes size
215+ -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
216+ withForeignPtr ptr $ \p -> do
217+ -- n <- hGetBufSome h p size
218+ n <- RawIO.read fd p size
219+ let v = Array
220+ { aStart = ptr
221+ , aEnd = p `plusPtr` n
222+ , aBound = p `plusPtr` size
223+ }
224+ -- XXX shrink only if the diff is significant
225+ -- A.shrinkToFit v
226+ return v
227+
228+-------------------------------------------------------------------------------
229+-- Array IO (output)
230+-------------------------------------------------------------------------------
231+
232+-- | Write an 'Array' to a file handle.
233+--
234+-- @since 0.7.0
235+{-# INLINABLE writeArray #-}
236+writeArray :: Storable a => Handle -> Array a -> IO ()
237+writeArray _ arr | A.length arr == 0 = return ()
238+writeArray (Handle fd) arr = withForeignPtr (aStart arr) $ \p -> do
239+ RawIO.writeAll fd (castPtr p) aLen
240+ {-
241+ -- Experiment to compare "writev" based IO with "write" based IO.
242+ iov <- A.newArray 1
243+ let iov' = iov {aEnd = aBound iov}
244+ A.writeIndex iov' 0 (RawIO.IOVec (castPtr p) (fromIntegral aLen))
245+ RawIO.writevAll fd (unsafeForeignPtrToPtr (aStart iov')) 1
246+ -}
247+ where
248+ aLen = byteLength arr
249+
250+-- | Write an array of 'IOVec' to a file handle.
251+--
252+-- @since 0.7.0
253+{-# INLINABLE writeIOVec #-}
254+writeIOVec :: Handle -> Array RawIO.IOVec -> IO ()
255+writeIOVec _ iov | A.length iov == 0 = return ()
256+writeIOVec (Handle fd) iov =
257+ withForeignPtr (aStart iov) $ \p ->
258+ RawIO.writevAll fd p (A.length iov)
259+
260+-------------------------------------------------------------------------------
261+-- Stream of Arrays IO
262+-------------------------------------------------------------------------------
263+
264+-- | @readArraysOfUpto size h@ reads a stream of arrays from file handle @h@.
265+-- The maximum size of a single array is specified by @size@. The actual size
266+-- read may be less than or equal to @size@.
267+{-# INLINABLE _readArraysOfUpto #-}
268+_readArraysOfUpto :: (IsStream t, MonadIO m)
269+ => Int -> Handle -> t m (Array Word8)
270+_readArraysOfUpto size h = go
271+ where
272+ -- XXX use cons/nil instead
273+ go = mkStream $ \_ yld _ stp -> do
274+ arr <- liftIO $ readArrayUpto size h
275+ if A.length arr == 0
276+ then stp
277+ else yld arr go
278+
279+{-# INLINE_NORMAL readArraysOfUpto #-}
280+readArraysOfUpto :: (IsStream t, MonadIO m)
281+ => Int -> Handle -> t m (Array Word8)
282+readArraysOfUpto size h = D.fromStreamD (D.Stream step ())
283+ where
284+ {-# INLINE_LATE step #-}
285+ step _ _ = do
286+ arr <- liftIO $ readArrayUpto size h
287+ return $
288+ case A.length arr of
289+ 0 -> D.Stop
290+ _ -> D.Yield arr ()
291+
292+-- XXX read 'Array a' instead of Word8
293+--
294+-- | @readArrays h@ reads a stream of arrays from file handle @h@.
295+-- The maximum size of a single array is limited to @defaultChunkSize@.
296+-- 'readArrays' ignores the prevailing 'TextEncoding' and 'NewlineMode'
297+-- on the 'Handle'.
298+--
299+-- > readArrays = readArraysOfUpto defaultChunkSize
300+--
301+-- @since 0.7.0
302+{-# INLINE readArrays #-}
303+readArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
304+readArrays = readArraysOfUpto defaultChunkSize
305+
306+-------------------------------------------------------------------------------
307+-- Read File to Stream
308+-------------------------------------------------------------------------------
309+
310+-- TODO for concurrent streams implement readahead IO. We can send multiple
311+-- read requests at the same time. For serial case we can use async IO. We can
312+-- also control the read throughput in mbps or IOPS.
313+
314+-- | @readByChunksUpto chunkSize handle@ reads a byte stream from a file handle,
315+-- reads are performed in chunks of up to @chunkSize@. The stream ends as soon
316+-- as EOF is encountered.
317+--
318+{-# INLINE readByChunksUpto #-}
319+readByChunksUpto :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
320+readByChunksUpto chunkSize h = A.flattenArrays $ readArraysOfUpto chunkSize h
321+
322+-- TODO
323+-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
324+--
325+-- > read = 'readByChunks' A.defaultChunkSize
326+-- | Generate a stream of elements of the given type from a file 'Handle'. The
327+-- stream ends when EOF is encountered.
328+--
329+-- @since 0.7.0
330+{-# INLINE read #-}
331+read :: (IsStream t, MonadIO m) => Handle -> t m Word8
332+read = A.flattenArrays . readArrays
333+
334+-------------------------------------------------------------------------------
335+-- Writing
336+-------------------------------------------------------------------------------
337+
338+-- | Write a stream of arrays to a handle.
339+--
340+-- @since 0.7.0
341+{-# INLINE writeArrays #-}
342+writeArrays :: (MonadIO m, Storable a) => Handle -> SerialT m (Array a) -> m ()
343+writeArrays h m = S.mapM_ (liftIO . writeArray h) m
344+
345+-- | Write a stream of 'IOVec' arrays to a handle.
346+--
347+-- @since 0.7.0
348+{-# INLINE writev #-}
349+writev :: MonadIO m => Handle -> SerialT m (Array RawIO.IOVec) -> m ()
350+writev h m = S.mapM_ (liftIO . writeIOVec h) m
351+
352+-- | Write a stream of arrays to a handle after coalescing them in chunks of
353+-- specified size. The chunk size is only a maximum and the actual writes could
354+-- be smaller than that as we do not split the arrays to fit them to the
355+-- specified size.
356+--
357+-- @since 0.7.0
358+{-# INLINE writeArraysOfUpto #-}
359+writeArraysOfUpto :: (MonadIO m, Storable a)
360+ => Int -> Handle -> SerialT m (Array a) -> m ()
361+writeArraysOfUpto n h xs = writeArrays h $ A.coalesceChunksOf n xs
362+
363+-- | Write a stream of arrays to a handle after grouping them in 'IOVec' arrays
364+-- of up to a maximum total size. Writes are performed using gather IO via
365+-- @writev@ system call. The maximum number of entries in each 'IOVec' group
366+-- limited to 512.
367+--
368+-- @since 0.7.0
369+{-# INLINE writevArraysOfUpto #-}
370+writevArraysOfUpto :: MonadIO m
371+ => Int -> Handle -> SerialT m (Array a) -> m ()
372+writevArraysOfUpto n h xs =
373+ writev h $ fromStreamD $ groupIOVecsOf n 512 (toStreamD xs)
374+
375+-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
376+--
377+-- XXX test this
378+-- Note that if you use a chunk size less than 8K (GHC's default buffer
379+-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
380+-- do not want buffering to occur at GHC level as well. Same thing applies to
381+-- writes as well.
382+
383+-- | Like 'write' but provides control over the write buffer. Output will
384+-- be written to the IO device as soon as we collect the specified number of
385+-- input elements.
386+--
387+-- @since 0.7.0
388+{-# INLINE writeByChunksOf #-}
389+writeByChunksOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
390+writeByChunksOf n h m = writeArrays h $ A.arraysOf n m
391+
392+-- > write = 'writeByChunks' A.defaultChunkSize
393+--
394+-- | Write a byte stream to a file handle. Combines the bytes in chunks of size
395+-- up to 'A.defaultChunkSize' before writing. Note that the write behavior
396+-- depends on the 'IOMode' and the current seek position of the handle.
397+--
398+-- @since 0.7.0
399+{-# INLINE write #-}
400+write :: MonadIO m => Handle -> SerialT m Word8 -> m ()
401+write = writeByChunksOf defaultChunkSize
402+
403+{-
404+{-# INLINE write #-}
405+write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
406+write = toHandleWith A.defaultChunkSize
407+-}
408+
409+-------------------------------------------------------------------------------
410+-- IO with encoding/decoding Unicode characters
411+-------------------------------------------------------------------------------
412+
413+{-
414+-- |
415+-- > readUtf8 = decodeUtf8 . read
416+--
417+-- Read a UTF8 encoded stream of unicode characters from a file handle.
418+--
419+-- @since 0.7.0
420+{-# INLINE readUtf8 #-}
421+readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
422+readUtf8 = decodeUtf8 . read
423+
424+-- |
425+-- > writeUtf8 h s = write h $ encodeUtf8 s
426+--
427+-- Encode a stream of unicode characters to UTF8 and write it to the given file
428+-- handle. Default block buffering applies to the writes.
429+--
430+-- @since 0.7.0
431+{-# INLINE writeUtf8 #-}
432+writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
433+writeUtf8 h s = write h $ encodeUtf8 s
434+
435+-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
436+-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
437+-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
438+-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
439+-- option.
440+--
441+-- @since 0.7.0
442+{-# INLINE writeUtf8ByLines #-}
443+writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
444+writeUtf8ByLines = undefined
445+
446+-- | Read UTF-8 lines from a file handle and apply the specified fold to each
447+-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
448+--
449+-- @since 0.7.0
450+{-# INLINE readLines #-}
451+readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
452+readLines h f = foldLines (readUtf8 h) f
453+
454+-------------------------------------------------------------------------------
455+-- Framing on a sequence
456+-------------------------------------------------------------------------------
457+
458+-- | Read a stream from a file handle and split it into frames delimited by
459+-- the specified sequence of elements. The supplied fold is applied on each
460+-- frame.
461+--
462+-- @since 0.7.0
463+{-# INLINE readFrames #-}
464+readFrames :: (IsStream t, MonadIO m, Storable a)
465+ => Array a -> Handle -> Fold m a b -> t m b
466+readFrames = undefined -- foldFrames . read
467+
468+-- | Write a stream to the given file handle buffering up to frames separated
469+-- by the given sequence or up to a maximum of @defaultChunkSize@.
470+--
471+-- @since 0.7.0
472+{-# INLINE writeByFrames #-}
473+writeByFrames :: (IsStream t, MonadIO m, Storable a)
474+ => Array a -> Handle -> t m a -> m ()
475+writeByFrames = undefined
476+
477+-------------------------------------------------------------------------------
478+-- Random Access IO (Seek)
479+-------------------------------------------------------------------------------
480+
481+-- XXX handles could be shared, so we may not want to use the handle state at
482+-- all for these APIs. we can use pread and pwrite instead. On windows we will
483+-- need to use readFile/writeFile with an offset argument.
484+
485+-------------------------------------------------------------------------------
486+
487+-- | Read the element at the given index treating the file as an array.
488+--
489+-- @since 0.7.0
490+{-# INLINE readIndex #-}
491+readIndex :: Storable a => Handle -> Int -> Maybe a
492+readIndex arr i = undefined
493+
494+-- NOTE: To represent a range to read we have chosen (start, size) instead of
495+-- (start, end). This helps in removing the ambiguity of whether "end" is
496+-- included in the range or not.
497+--
498+-- We could avoid specifying the range to be read and instead use "take size"
499+-- on the stream, but it may end up reading more and then consume it partially.
500+
501+-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
502+-- from @handle@ starting at the offset @pos@ from the beginning of the file.
503+--
504+-- Reads are performed in chunks of size @chunkSize@. For block devices, to
505+-- avoid reading partial blocks @chunkSize@ must align with the block size of
506+-- the underlying device. If the underlying block size is unknown, it is a good
507+-- idea to keep it a multiple 4KiB. This API ensures that the start of each
508+-- chunk is aligned with @chunkSize@ from second chunk onwards.
509+--
510+{-# INLINE readSliceWith #-}
511+readSliceWith :: (IsStream t, MonadIO m, Storable a)
512+ => Int -> Handle -> Int -> Int -> t m a
513+readSliceWith chunkSize h pos len = undefined
514+
515+-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
516+-- at index @i@ and reading up to @count@ elements in the forward direction
517+-- ending at the index @i + count - 1@.
518+--
519+-- @since 0.7.0
520+{-# INLINE readSlice #-}
521+readSlice :: (IsStream t, MonadIO m, Storable a)
522+ => Handle -> Int -> Int -> t m a
523+readSlice = readSliceWith A.defaultChunkSize
524+
525+-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
526+-- at index @i@ and reading up to @count@ elements in the reverse direction
527+-- ending at the index @i - count + 1@.
528+--
529+-- @since 0.7.0
530+{-# INLINE readSliceRev #-}
531+readSliceRev :: (IsStream t, MonadIO m, Storable a)
532+ => Handle -> Int -> Int -> t m a
533+readSliceRev h i count = undefined
534+
535+-- | Write the given element at the given index in the file.
536+--
537+-- @since 0.7.0
538+{-# INLINE writeIndex #-}
539+writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
540+writeIndex h i a = undefined
541+
542+-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
543+-- starting at index @i@ and writing up to @count@ elements in the forward
544+-- direction ending at the index @i + count - 1@.
545+--
546+-- @since 0.7.0
547+{-# INLINE writeSlice #-}
548+writeSlice :: (IsStream t, Monad m, Storable a)
549+ => Handle -> Int -> Int -> t m a -> m ()
550+writeSlice h i len s = undefined
551+
552+-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
553+-- starting at index @i@ and writing up to @count@ elements in the reverse
554+-- direction ending at the index @i - count + 1@.
555+--
556+-- @since 0.7.0
557+{-# INLINE writeSliceRev #-}
558+writeSliceRev :: (IsStream t, Monad m, Storable a)
559+ => Handle -> Int -> Int -> t m a -> m ()
560+writeSliceRev arr i len s = undefined
561+-}