@@ -11,23 +11,66 @@
1111-- Copyright : (c) 2018 Composewell Technologies
1212--
1313-- License : BSD3
14−-- Maintainer : harendra.kumar@gmail.com
14+-- Maintainer : streamly@composewell.com
1515-- Stability : experimental
1616-- Portability : GHC
17−--
18−-- Streaming based IO write rotuines based on file handles.
1917
2018module Streamly.Internal.FileSystem.Handle
2119 (
22− writeArray
20+ -- ** Read from Handle
21+ read
22+ -- , readUtf8
23+ -- , readLines
24+ -- , readFrames
25+ , readInChunksOf
26+
27+ -- -- * Array Read
28+ -- , readArrayUpto
29+ -- , readArrayOf
30+
31+ , readArraysOf
32+ -- , readArraysOf
33+ , readArrays
2334
24− -- Byte stream write
35+ -- ** Write to Handle
36+ -- Byte stream write (Folds)
37+ , write
38+ -- , writeUtf8
39+ -- , writeUtf8ByLines
40+ -- , writeByFrames
41+ , writeInChunksOf
42+
43+ -- Byte stream write (Streams)
2544 , writeS
2645 , writeSInChunksOf
2746
47+ -- -- * Array Write
48+ , writeArray
49+ , writeArrays
50+ , writeArraysInChunksOf
51+
2852 -- -- * Array stream Write
2953 , writeSArrays
3054 , writeSArraysInChunksOf
55+
56+ -- -- * Random Access (Seek)
57+ -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or
58+ -- files that have random access or seek capability. This type of devices
59+ -- include disks, files, memory devices and exclude terminals, pipes,
60+ -- sockets and fifos.
61+ --
62+ -- , readIndex
63+ -- , readSlice
64+ -- , readSliceRev
65+ -- , readAt -- read from a given position to th end of file
66+ -- , readSliceArrayUpto
67+ -- , readSliceArrayOf
68+
69+ -- , writeIndex
70+ -- , writeSlice
71+ -- , writeSliceRev
72+ -- , writeAt -- start writing at the given position
73+ -- , writeSliceArray
3174 )
3275where
3376
@@ -35,18 +78,144 @@ import Control.Monad.IO.Class (MonadIO(..))
3578import Data.Word (Word8)
3679import Foreign.ForeignPtr (withForeignPtr)
3780import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
38−import Foreign.Ptr (minusPtr)
81+import Foreign.Ptr (minusPtr, plusPtr)
3982import Foreign.Storable (Storable(..))
40−import System.IO (Handle, hPutBuf)
83+import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
84+import System.IO (Handle, hGetBufSome, hPutBuf)
4185import Prelude hiding (read)
4286
43−import Streamly.Internal.Memory.Array.Types (Array(..))
87+import Streamly.Data.Fold (Fold)
88+import Streamly.Internal.Memory.Array.Types
89+ (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,
90+ lpackArraysChunksOf)
4491import Streamly.Streams.Serial (SerialT)
45−import Streamly.Internal.Memory.Array.Types (defaultChunkSize)
92+import Streamly.Streams.StreamK.Type (IsStream, mkStream)
93+-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
4694
47−import qualified Streamly.Memory.Array as A
95+import qualified Streamly.Data.Fold as FL
96+import qualified Streamly.Internal.Data.Fold.Types as FL
4897import qualified Streamly.Internal.Memory.ArrayStream as AS
4998import qualified Streamly.Internal.Prelude as S
99+import qualified Streamly.Memory.Array as A
100+import qualified Streamly.Streams.StreamD.Type as D
101+
102+-------------------------------------------------------------------------------
103+-- References
104+-------------------------------------------------------------------------------
105+--
106+-- The following references may be useful to build an understanding about the
107+-- file API design:
108+--
109+-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
110+-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
111+-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
112+-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
113+-- https://www.w3.org/TR/FileAPI/ for http file API.
114+
115+-------------------------------------------------------------------------------
116+-- Array IO (Input)
117+-------------------------------------------------------------------------------
118+
119+-- | Read a 'ByteArray' from a file handle. If no data is available on the
120+-- handle it blocks until some data becomes available. If data is available
121+-- then it immediately returns that data without blocking. It reads a maximum
122+-- of up to the size requested.
123+{-# INLINABLE readArrayUpto #-}
124+readArrayUpto :: Int -> Handle -> IO (Array Word8)
125+readArrayUpto size h = do
126+ ptr <- mallocPlainForeignPtrBytes size
127+ -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
128+ withForeignPtr ptr $ \p -> do
129+ n <- hGetBufSome h p size
130+ let v = Array
131+ { aStart = ptr
132+ , aEnd = p `plusPtr` n
133+ , aBound = p `plusPtr` size
134+ }
135+ -- XXX shrink only if the diff is significant
136+ shrinkToFit v
137+
138+-------------------------------------------------------------------------------
139+-- Stream of Arrays IO
140+-------------------------------------------------------------------------------
141+
142+-- | @readArraysOf size h@ reads a stream of arrays from file handle @h@.
143+-- The maximum size of a single array is specified by @size@. The actual size
144+-- read may be less than or equal to @size@.
145+{-# INLINABLE _readArraysOf #-}
146+_readArraysOf :: (IsStream t, MonadIO m)
147+ => Int -> Handle -> t m (Array Word8)
148+_readArraysOf size h = go
149+ where
150+ -- XXX use cons/nil instead
151+ go = mkStream $ \_ yld _ stp -> do
152+ arr <- liftIO $ readArrayUpto size h
153+ if A.length arr == 0
154+ then stp
155+ else yld arr go
156+
157+-- | @readArraysOf size handle@ reads a stream of arrays from the file
158+-- handle @handle@. The maximum size of a single array is limited to @size@.
159+-- The actual size read may be less than or equal to @size@.
160+--
161+-- @since 0.7.0
162+{-# INLINE_NORMAL readArraysOf #-}
163+readArraysOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
164+readArraysOf size h = D.fromStreamD (D.Stream step ())
165+ where
166+ {-# INLINE_LATE step #-}
167+ step _ _ = do
168+ arr <- liftIO $ readArrayUpto size h
169+ return $
170+ case A.length arr of
171+ 0 -> D.Stop
172+ _ -> D.Yield arr ()
173+
174+-- XXX read 'Array a' instead of Word8
175+--
176+-- | @readArrays handle@ reads a stream of arrays from the specified file
177+-- handle. The maximum size of a single array is limited to
178+-- @defaultChunkSize@. The actual size read may be less than or equal to
179+-- @defaultChunkSize@.
180+--
181+-- > readArrays = readArraysOf defaultChunkSize
182+--
183+-- @since 0.7.0
184+{-# INLINE readArrays #-}
185+readArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
186+readArrays = readArraysOf defaultChunkSize
187+
188+-------------------------------------------------------------------------------
189+-- Read File to Stream
190+-------------------------------------------------------------------------------
191+
192+-- TODO for concurrent streams implement readahead IO. We can send multiple
193+-- read requests at the same time. For serial case we can use async IO. We can
194+-- also control the read throughput in mbps or IOPS.
195+
196+-- | @readInChunksOf chunkSize handle@ reads a byte stream from a file
197+-- handle, reads are performed in chunks of up to @chunkSize@.
198+--
199+-- @since 0.7.0
200+{-# INLINE readInChunksOf #-}
201+readInChunksOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
202+readInChunksOf chunkSize h = AS.concat $ readArraysOf chunkSize h
203+
204+-- TODO
205+-- Generate a stream of elements of the given type from a file 'Handle'.
206+-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
207+--
208+-- > read = 'readInChunksOf' A.defaultChunkSize
209+-- | Generate a byte stream from a file 'Handle'.
210+--
211+-- @since 0.7.0
212+{-# INLINE read #-}
213+read :: (IsStream t, MonadIO m) => Handle -> t m Word8
214+read = AS.concat . readArrays
215+
216+-------------------------------------------------------------------------------
217+-- Writing
218+-------------------------------------------------------------------------------
50219
51220-------------------------------------------------------------------------------
52221-- Array IO (output)
@@ -113,3 +282,230 @@ writeSInChunksOf n h m = writeSArrays h $ S.arraysOf n m
113282{-# INLINE writeS #-}
114283writeS :: MonadIO m => Handle -> SerialT m Word8 -> m ()
115284writeS = writeSInChunksOf defaultChunkSize
285+
286+-- | Write a stream of arrays to a handle.
287+--
288+-- @since 0.7.0
289+{-# INLINE writeArrays #-}
290+writeArrays :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
291+writeArrays h = FL.drainBy (liftIO . writeArray h)
292+
293+-- | @writeArraysInChunksOf chunkSize handle@ writes a stream of arrays
294+-- to @handle@ after coalescing the adjacent arrays in chunks of @chunkSize@.
295+-- We never split an array, if a single array is bigger than the specified size
296+-- it emitted as it is. Multiple arrays are coalesed as long as the total size
297+-- remains below the specified size.
298+--
299+-- @since 0.7.0
300+{-# INLINE writeArraysInChunksOf #-}
301+writeArraysInChunksOf :: (MonadIO m, Storable a)
302+ => Int -> Handle -> Fold m (Array a) ()
303+writeArraysInChunksOf n h = lpackArraysChunksOf n (writeArrays h)
304+
305+-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
306+--
307+-- XXX test this
308+-- Note that if you use a chunk size less than 8K (GHC's default buffer
309+-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
310+-- do not want buffering to occur at GHC level as well. Same thing applies to
311+-- writes as well.
312+
313+-- | @writeInChunksOf chunkSize handle@ writes the input stream to @handle@ in
314+-- chunks of @chunkSize@. A write is performed to the IO device as soon as we
315+-- collect the required input size.
316+--
317+-- @since 0.7.0
318+{-# INLINE writeInChunksOf #-}
319+writeInChunksOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
320+writeInChunksOf n h = FL.lchunksOf n (writeNUnsafe n) (writeArrays h)
321+
322+-- > write = 'writeInChunksOf' A.defaultChunkSize
323+--
324+-- | Write a byte stream to a file handle. Accumulates the input in chunks of
325+-- up to 'A.defaultChunkSize' before writing.
326+--
327+-- @since 0.7.0
328+{-# INLINE write #-}
329+write :: MonadIO m => Handle -> Fold m Word8 ()
330+write = writeInChunksOf defaultChunkSize
331+
332+{-
333+{-# INLINE write #-}
334+write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
335+write = toHandleWith A.defaultChunkSize
336+-}
337+
338+-- XXX mmap a file into an array. This could be useful for in-place operations
339+-- on a file. For example, we can quicksort the contents of a file by mmapping
340+-- it.
341+
342+-------------------------------------------------------------------------------
343+-- IO with encoding/decoding Unicode characters
344+-------------------------------------------------------------------------------
345+
346+{-
347+-- |
348+-- > readUtf8 = decodeUtf8 . read
349+--
350+-- Read a UTF8 encoded stream of unicode characters from a file handle.
351+--
352+-- @since 0.7.0
353+{-# INLINE readUtf8 #-}
354+readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
355+readUtf8 = decodeUtf8 . read
356+
357+-- |
358+-- > writeUtf8 h s = write h $ encodeUtf8 s
359+--
360+-- Encode a stream of unicode characters to UTF8 and write it to the given file
361+-- handle. Default block buffering applies to the writes.
362+--
363+-- @since 0.7.0
364+{-# INLINE writeUtf8 #-}
365+writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
366+writeUtf8 h s = write h $ encodeUtf8 s
367+
368+-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
369+-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
370+-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
371+-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
372+-- option.
373+--
374+-- @since 0.7.0
375+{-# INLINE writeUtf8ByLines #-}
376+writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
377+writeUtf8ByLines = undefined
378+
379+-- | Read UTF-8 lines from a file handle and apply the specified fold to each
380+-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
381+--
382+-- @since 0.7.0
383+{-# INLINE readLines #-}
384+readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
385+readLines h f = foldLines (readUtf8 h) f
386+
387+-------------------------------------------------------------------------------
388+-- Framing on a sequence
389+-------------------------------------------------------------------------------
390+
391+-- | Read a stream from a file handle and split it into frames delimited by
392+-- the specified sequence of elements. The supplied fold is applied on each
393+-- frame.
394+--
395+-- @since 0.7.0
396+{-# INLINE readFrames #-}
397+readFrames :: (IsStream t, MonadIO m, Storable a)
398+ => Array a -> Handle -> Fold m a b -> t m b
399+readFrames = undefined -- foldFrames . read
400+
401+-- | Write a stream to the given file handle buffering up to frames separated
402+-- by the given sequence or up to a maximum of @defaultChunkSize@.
403+--
404+-- @since 0.7.0
405+{-# INLINE writeByFrames #-}
406+writeByFrames :: (IsStream t, MonadIO m, Storable a)
407+ => Array a -> Handle -> t m a -> m ()
408+writeByFrames = undefined
409+
410+-------------------------------------------------------------------------------
411+-- Framing by time
412+-------------------------------------------------------------------------------
413+
414+-- | Write collecting the input in sessions of n seconds or if chunkSize
415+-- gets exceeded.
416+{-# INLINE writeByChunksOrSessionsOf #-}
417+writeByChunksOrSessionsOf :: MonadIO m
418+ => Int -> Double -> Handle -> SerialT m Word8 -> m ()
419+writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined
420+
421+-- | Write collecting the input in sessions of n seconds or if defaultChunkSize
422+-- gets exceeded.
423+{-# INLINE writeBySessionsOf #-}
424+writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()
425+writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n
426+
427+-------------------------------------------------------------------------------
428+-- Random Access IO (Seek)
429+-------------------------------------------------------------------------------
430+
431+-- XXX handles could be shared, so we may not want to use the handle state at
432+-- all for these APIs. we can use pread and pwrite instead. On windows we will
433+-- need to use readFile/writeFile with an offset argument.
434+
435+-------------------------------------------------------------------------------
436+
437+-- | Read the element at the given index treating the file as an array.
438+--
439+-- @since 0.7.0
440+{-# INLINE readIndex #-}
441+readIndex :: Storable a => Handle -> Int -> Maybe a
442+readIndex arr i = undefined
443+
444+-- NOTE: To represent a range to read we have chosen (start, size) instead of
445+-- (start, end). This helps in removing the ambiguity of whether "end" is
446+-- included in the range or not.
447+--
448+-- We could avoid specifying the range to be read and instead use "take size"
449+-- on the stream, but it may end up reading more and then consume it partially.
450+
451+-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
452+-- from @handle@ starting at the offset @pos@ from the beginning of the file.
453+--
454+-- Reads are performed in chunks of size @chunkSize@. For block devices, to
455+-- avoid reading partial blocks @chunkSize@ must align with the block size of
456+-- the underlying device. If the underlying block size is unknown, it is a good
457+-- idea to keep it a multiple 4KiB. This API ensures that the start of each
458+-- chunk is aligned with @chunkSize@ from second chunk onwards.
459+--
460+{-# INLINE readSliceWith #-}
461+readSliceWith :: (IsStream t, MonadIO m, Storable a)
462+ => Int -> Handle -> Int -> Int -> t m a
463+readSliceWith chunkSize h pos len = undefined
464+
465+-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
466+-- at index @i@ and reading up to @count@ elements in the forward direction
467+-- ending at the index @i + count - 1@.
468+--
469+-- @since 0.7.0
470+{-# INLINE readSlice #-}
471+readSlice :: (IsStream t, MonadIO m, Storable a)
472+ => Handle -> Int -> Int -> t m a
473+readSlice = readSliceWith A.defaultChunkSize
474+
475+-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
476+-- at index @i@ and reading up to @count@ elements in the reverse direction
477+-- ending at the index @i - count + 1@.
478+--
479+-- @since 0.7.0
480+{-# INLINE readSliceRev #-}
481+readSliceRev :: (IsStream t, MonadIO m, Storable a)
482+ => Handle -> Int -> Int -> t m a
483+readSliceRev h i count = undefined
484+
485+-- | Write the given element at the given index in the file.
486+--
487+-- @since 0.7.0
488+{-# INLINE writeIndex #-}
489+writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
490+writeIndex h i a = undefined
491+
492+-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
493+-- starting at index @i@ and writing up to @count@ elements in the forward
494+-- direction ending at the index @i + count - 1@.
495+--
496+-- @since 0.7.0
497+{-# INLINE writeSlice #-}
498+writeSlice :: (IsStream t, Monad m, Storable a)
499+ => Handle -> Int -> Int -> t m a -> m ()
500+writeSlice h i len s = undefined
501+
502+-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
503+-- starting at index @i@ and writing up to @count@ elements in the reverse
504+-- direction ending at the index @i - count + 1@.
505+--
506+-- @since 0.7.0
507+{-# INLINE writeSliceRev #-}
508+writeSliceRev :: (IsStream t, Monad m, Storable a)
509+ => Handle -> Int -> Int -> t m a -> m ()
510+writeSliceRev arr i len s = undefined
511+-}