2{-# LANGUAGE BangPatterns #-}
3{-# LANGUAGE MagicHash #-}
4{-# LANGUAGE RecordWildCards #-}
5{-# LANGUAGE UnboxedTuples #-}
10-- Module : Streamly.Internal.FileSystem.Handle
11-- Copyright : (c) 2018 Composewell Technologies
14-- Maintainer : streamly@composewell.com
15-- Stability : experimental
18module Streamly.Internal.FileSystem.Handle
20 -- ** Read from Handle
40 -- Byte stream write (Folds)
47 -- Byte stream write (Streams)
49 , fromStreamInChunksOf
54 , writeArraysInChunksOf
56 -- -- * Array stream Write
58 , fromStreamArraysInChunksOf
60 -- -- * Random Access (Seek)
61 -- -- | Unlike the streaming APIs listed above, these APIs apply to devices or
62 -- files that have random access or seek capability. This type of devices
63 -- include disks, files, memory devices and exclude terminals, pipes,
69 -- , readAt -- read from a given position to th end of file
70 -- , readSliceArrayUpto
76 -- , writeAt -- start writing at the given position
81import Control.Monad.IO.Class (MonadIO(..))
82import Data.Word (Word8)
83import Foreign.ForeignPtr (withForeignPtr)
84import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
85import Foreign.Ptr (minusPtr, plusPtr)
86import Foreign.Storable (Storable(..))
87import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
88import System.IO (Handle, hGetBufSome, hPutBuf)
89import Prelude hiding (read)
91import Streamly.Data.Fold (Fold)
92import Streamly.Internal.Data.Unfold.Types (Unfold(..))
93import Streamly.Internal.Memory.Array.Types
94 (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,
96import Streamly.Streams.Serial (SerialT)
97import Streamly.Streams.StreamK.Type (IsStream, mkStream)
98-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
100import qualified Streamly.Data.Fold as FL
101import qualified Streamly.Internal.Data.Fold.Types as FL
102import qualified Streamly.Internal.Data.Unfold as UF
103import qualified Streamly.Internal.Memory.ArrayStream as AS
104import qualified Streamly.Internal.Prelude as S
105import qualified Streamly.Memory.Array as A
106import qualified Streamly.Streams.StreamD.Type as D
108-------------------------------------------------------------------------------
110-------------------------------------------------------------------------------
112-- The following references may be useful to build an understanding about the
115-- http://www.linux-mag.com/id/308/ for blocking/non-blocking IO on linux.
116-- https://lwn.net/Articles/612483/ Non-blocking buffered file read operations
117-- https://en.wikipedia.org/wiki/C_file_input/output for C APIs.
118-- https://docs.oracle.com/javase/tutorial/essential/io/file.html for Java API.
119-- https://www.w3.org/TR/FileAPI/ for http file API.
121-------------------------------------------------------------------------------
123-------------------------------------------------------------------------------
125-- | Read a 'ByteArray' from a file handle. If no data is available on the
126-- handle it blocks until some data becomes available. If data is available
127-- then it immediately returns that data without blocking. It reads a maximum
128-- of up to the size requested.
129{-# INLINABLE readArrayUpto #-}
130readArrayUpto :: Int -> Handle -> IO (Array Word8)
131readArrayUpto size h = do
132 ptr <- mallocPlainForeignPtrBytes size
133 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
134 withForeignPtr ptr $ \p -> do
135 n <- hGetBufSome h p size
138 , aEnd = p `plusPtr` n
139 , aBound = p `plusPtr` size
141 -- XXX shrink only if the diff is significant
144-------------------------------------------------------------------------------
145-- Stream of Arrays IO
146-------------------------------------------------------------------------------
148-- | @toStreamArraysOf size h@ reads a stream of arrays from file handle @h@.
149-- The maximum size of a single array is specified by @size@. The actual size
150-- read may be less than or equal to @size@.
151{-# INLINABLE _toStreamArraysOf #-}
152_toStreamArraysOf :: (IsStream t, MonadIO m)
153 => Int -> Handle -> t m (Array Word8)
154_toStreamArraysOf size h = go
156 -- XXX use cons/nil instead
157 go = mkStream $ \_ yld _ stp -> do
158 arr <- liftIO $ readArrayUpto size h
163-- | @toStreamArraysOf size handle@ reads a stream of arrays from the file
164-- handle @handle@. The maximum size of a single array is limited to @size@.
165-- The actual size read may be less than or equal to @size@.
168{-# INLINE_NORMAL toStreamArraysOf #-}
169toStreamArraysOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
170toStreamArraysOf size h = D.fromStreamD (D.Stream step ())
172 {-# INLINE_LATE step #-}
174 arr <- liftIO $ readArrayUpto size h
180-- | Unfold the tuple @(size, handle)@ into a stream of 'Word8' arrays. The
181-- stream consists of arrays representing chunks of data read from the handle.
182-- The maximum size of a single array is limited to @size@.
185{-# INLINE_NORMAL readArraysOf #-}
186readArraysOf :: MonadIO m => Unfold m (Int, Handle) (Array Word8)
187readArraysOf = Unfold step return
189 {-# INLINE_LATE step #-}
191 arr <- liftIO $ readArrayUpto size h
195 _ -> D.Yield arr (size, h)
197-- XXX read 'Array a' instead of Word8
199-- | @toStreamArrays handle@ reads a stream of arrays from the specified file
200-- handle. The maximum size of a single array is limited to
201-- @defaultChunkSize@. The actual size read may be less than or equal to
202-- @defaultChunkSize@.
204-- > toStreamArrays = toStreamArraysOf defaultChunkSize
207{-# INLINE toStreamArrays #-}
208toStreamArrays :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
209toStreamArrays = toStreamArraysOf defaultChunkSize
211-- | Unfolds a handle into a stream of 'Word8' arrays. The maximum size of a
212-- single array is limited to @defaultChunkSize@.
215{-# INLINE readArrays #-}
216readArrays :: MonadIO m => Unfold m Handle (Array Word8)
217readArrays = UF.supplyFirst readArraysOf defaultChunkSize
219-------------------------------------------------------------------------------
220-- Read File to Stream
221-------------------------------------------------------------------------------
223-- TODO for concurrent streams implement readahead IO. We can send multiple
224-- read requests at the same time. For serial case we can use async IO. We can
225-- also control the read throughput in mbps or IOPS.
227-- | @readInChunksOf@ unfolds @(chunksize, handle)@ into a byte stream, reads
228-- are performed in chunks of up to @chunkSize@.
231{-# INLINE readInChunksOf #-}
232readInChunksOf :: MonadIO m => Unfold m (Int, Handle) Word8
233readInChunksOf = UF.concat readArraysOf A.read
235-- | @toStreamInChunksOf chunkSize handle@ reads a byte stream from a file
236-- handle, reads are performed in chunks of up to @chunkSize@.
239{-# INLINE toStreamInChunksOf #-}
240toStreamInChunksOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
241toStreamInChunksOf chunkSize h = AS.concat $ toStreamArraysOf chunkSize h
244-- Generate a stream of elements of the given type from a file 'Handle'.
245-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
247-- > read = 'readInChunksOf' A.defaultChunkSize
248-- | Unfolds a file handle into a byte stream.
252read :: MonadIO m => Unfold m Handle Word8
253read = UF.supplyFirst readInChunksOf defaultChunkSize
255-- | Generate a byte stream from a file 'Handle'.
258{-# INLINE toStream #-}
259toStream :: (IsStream t, MonadIO m) => Handle -> t m Word8
260toStream = AS.concat . toStreamArrays
262-------------------------------------------------------------------------------
264-------------------------------------------------------------------------------
266-------------------------------------------------------------------------------
268-------------------------------------------------------------------------------
270-- | Write an 'Array' to a file handle.
273{-# INLINABLE writeArray #-}
274writeArray :: Storable a => Handle -> Array a -> IO ()
275writeArray _ arr | A.length arr == 0 = return ()
276writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen
279 let p = unsafeForeignPtrToPtr aStart
282-------------------------------------------------------------------------------
283-- Stream of Arrays IO
284-------------------------------------------------------------------------------
286-------------------------------------------------------------------------------
288-------------------------------------------------------------------------------
290-- | Write a stream of arrays to a handle.
293{-# INLINE fromStreamArrays #-}
294fromStreamArrays :: (MonadIO m, Storable a)
295 => Handle -> SerialT m (Array a) -> m ()
296fromStreamArrays h m = S.mapM_ (liftIO . writeArray h) m
298-- | @fromStreamArraysInChunksOf chunkSize handle stream@ writes a stream of arrays
299-- to @handle@ after coalescing the adjacent arrays in chunks of @chunkSize@.
300-- The chunk size is only a maximum and the actual writes could be smaller as
301-- we do not split the arrays to fit exactly to the specified size.
304{-# INLINE fromStreamArraysInChunksOf #-}
305fromStreamArraysInChunksOf :: (MonadIO m, Storable a)
306 => Int -> Handle -> SerialT m (Array a) -> m ()
307fromStreamArraysInChunksOf n h xs = fromStreamArrays h $ AS.compact n xs
309-- | @fromStreamInChunksOf chunkSize handle stream@ writes @stream@ to @handle@ in
310-- chunks of @chunkSize@. A write is performed to the IO device as soon as we
311-- collect the required input size.
314{-# INLINE fromStreamInChunksOf #-}
315fromStreamInChunksOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
316fromStreamInChunksOf n h m = fromStreamArrays h $ S.arraysOf n m
317-- fromStreamInChunksOf n h m = fromStreamArrays h $ AS.arraysOf n m
319-- > write = 'writeInChunksOf' A.defaultChunkSize
321-- | Write a byte stream to a file handle. Accumulates the input in chunks of
322-- up to 'A.defaultChunkSize' before writing.
324-- NOTE: This may perform better than the 'write' fold, you can try this if you
325-- need some extra perf boost.
328{-# INLINE fromStream #-}
329fromStream :: MonadIO m => Handle -> SerialT m Word8 -> m ()
330fromStream = fromStreamInChunksOf defaultChunkSize
332-- | Write a stream of arrays to a handle.
335{-# INLINE writeArrays #-}
336writeArrays :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
337writeArrays h = FL.drainBy (liftIO . writeArray h)
339-- | @writeArraysInChunksOf chunkSize handle@ writes a stream of arrays
340-- to @handle@ after coalescing the adjacent arrays in chunks of @chunkSize@.
341-- We never split an array, if a single array is bigger than the specified size
342-- it emitted as it is. Multiple arrays are coalesed as long as the total size
343-- remains below the specified size.
346{-# INLINE writeArraysInChunksOf #-}
347writeArraysInChunksOf :: (MonadIO m, Storable a)
348 => Int -> Handle -> Fold m (Array a) ()
349writeArraysInChunksOf n h = lpackArraysChunksOf n (writeArrays h)
351-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
354-- Note that if you use a chunk size less than 8K (GHC's default buffer
355-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
356-- do not want buffering to occur at GHC level as well. Same thing applies to
359-- | @writeInChunksOf chunkSize handle@ writes the input stream to @handle@ in
360-- chunks of @chunkSize@. A write is performed to the IO device as soon as we
361-- collect the required input size.
364{-# INLINE writeInChunksOf #-}
365writeInChunksOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
366writeInChunksOf n h = FL.lchunksOf n (writeNUnsafe n) (writeArrays h)
368-- > write = 'writeInChunksOf' A.defaultChunkSize
370-- | Write a byte stream to a file handle. Accumulates the input in chunks of
371-- up to 'A.defaultChunkSize' before writing.
375write :: MonadIO m => Handle -> Fold m Word8 ()
376write = writeInChunksOf defaultChunkSize
380write :: (MonadIO m, Storable a) => Handle -> SerialT m a -> m ()
381write = toHandleWith A.defaultChunkSize
384-- XXX mmap a file into an array. This could be useful for in-place operations
385-- on a file. For example, we can quicksort the contents of a file by mmapping
388-------------------------------------------------------------------------------
389-- IO with encoding/decoding Unicode characters
390-------------------------------------------------------------------------------
394-- > readUtf8 = decodeUtf8 . read
396-- Read a UTF8 encoded stream of unicode characters from a file handle.
399{-# INLINE readUtf8 #-}
400readUtf8 :: (IsStream t, MonadIO m) => Handle -> t m Char
401readUtf8 = decodeUtf8 . read
404-- > writeUtf8 h s = write h $ encodeUtf8 s
406-- Encode a stream of unicode characters to UTF8 and write it to the given file
407-- handle. Default block buffering applies to the writes.
410{-# INLINE writeUtf8 #-}
411writeUtf8 :: MonadIO m => Handle -> SerialT m Char -> m ()
412writeUtf8 h s = write h $ encodeUtf8 s
414-- | Write a stream of unicode characters after encoding to UTF-8 in chunks
415-- separated by a linefeed character @'\n'@. If the size of the buffer exceeds
416-- @defaultChunkSize@ and a linefeed is not yet found, the buffer is written
417-- anyway. This is similar to writing to a 'Handle' with the 'LineBuffering'
421{-# INLINE writeUtf8ByLines #-}
422writeUtf8ByLines :: (IsStream t, MonadIO m) => Handle -> t m Char -> m ()
423writeUtf8ByLines = undefined
425-- | Read UTF-8 lines from a file handle and apply the specified fold to each
426-- line. This is similar to reading a 'Handle' with the 'LineBuffering' option.
429{-# INLINE readLines #-}
430readLines :: (IsStream t, MonadIO m) => Handle -> Fold m Char b -> t m b
431readLines h f = foldLines (readUtf8 h) f
433-------------------------------------------------------------------------------
434-- Framing on a sequence
435-------------------------------------------------------------------------------
437-- | Read a stream from a file handle and split it into frames delimited by
438-- the specified sequence of elements. The supplied fold is applied on each
442{-# INLINE readFrames #-}
443readFrames :: (IsStream t, MonadIO m, Storable a)
444 => Array a -> Handle -> Fold m a b -> t m b
445readFrames = undefined -- foldFrames . read
447-- | Write a stream to the given file handle buffering up to frames separated
448-- by the given sequence or up to a maximum of @defaultChunkSize@.
451{-# INLINE writeByFrames #-}
452writeByFrames :: (IsStream t, MonadIO m, Storable a)
453 => Array a -> Handle -> t m a -> m ()
454writeByFrames = undefined
456-------------------------------------------------------------------------------
458-------------------------------------------------------------------------------
460-- | Write collecting the input in sessions of n seconds or if chunkSize
462{-# INLINE writeByChunksOrSessionsOf #-}
463writeByChunksOrSessionsOf :: MonadIO m
464 => Int -> Double -> Handle -> SerialT m Word8 -> m ()
465writeByChunksOrSessionsOf chunkSize sessionSize h m = undefined
467-- | Write collecting the input in sessions of n seconds or if defaultChunkSize
469{-# INLINE writeBySessionsOf #-}
470writeBySessionsOf :: MonadIO m => Double -> Handle -> SerialT m Word8 -> m ()
471writeBySessionsOf n = writeByChunksOrSessionsOf defaultChunkSize n
473-------------------------------------------------------------------------------
474-- Random Access IO (Seek)
475-------------------------------------------------------------------------------
477-- XXX handles could be shared, so we may not want to use the handle state at
478-- all for these APIs. we can use pread and pwrite instead. On windows we will
479-- need to use readFile/writeFile with an offset argument.
481-------------------------------------------------------------------------------
483-- | Read the element at the given index treating the file as an array.
486{-# INLINE readIndex #-}
487readIndex :: Storable a => Handle -> Int -> Maybe a
488readIndex arr i = undefined
490-- NOTE: To represent a range to read we have chosen (start, size) instead of
491-- (start, end). This helps in removing the ambiguity of whether "end" is
492-- included in the range or not.
494-- We could avoid specifying the range to be read and instead use "take size"
495-- on the stream, but it may end up reading more and then consume it partially.
497-- | @readSliceWith chunkSize handle pos len@ reads up to @len@ bytes
498-- from @handle@ starting at the offset @pos@ from the beginning of the file.
500-- Reads are performed in chunks of size @chunkSize@. For block devices, to
501-- avoid reading partial blocks @chunkSize@ must align with the block size of
502-- the underlying device. If the underlying block size is unknown, it is a good
503-- idea to keep it a multiple 4KiB. This API ensures that the start of each
504-- chunk is aligned with @chunkSize@ from second chunk onwards.
506{-# INLINE readSliceWith #-}
507readSliceWith :: (IsStream t, MonadIO m, Storable a)
508 => Int -> Handle -> Int -> Int -> t m a
509readSliceWith chunkSize h pos len = undefined
511-- | @readSlice h i count@ streams a slice from the file handle @h@ starting
512-- at index @i@ and reading up to @count@ elements in the forward direction
513-- ending at the index @i + count - 1@.
516{-# INLINE readSlice #-}
517readSlice :: (IsStream t, MonadIO m, Storable a)
518 => Handle -> Int -> Int -> t m a
519readSlice = readSliceWith A.defaultChunkSize
521-- | @readSliceRev h i count@ streams a slice from the file handle @h@ starting
522-- at index @i@ and reading up to @count@ elements in the reverse direction
523-- ending at the index @i - count + 1@.
526{-# INLINE readSliceRev #-}
527readSliceRev :: (IsStream t, MonadIO m, Storable a)
528 => Handle -> Int -> Int -> t m a
529readSliceRev h i count = undefined
531-- | Write the given element at the given index in the file.
534{-# INLINE writeIndex #-}
535writeIndex :: (MonadIO m, Storable a) => Handle -> Int -> a -> m ()
536writeIndex h i a = undefined
538-- | @writeSlice h i count stream@ writes a stream to the file handle @h@
539-- starting at index @i@ and writing up to @count@ elements in the forward
540-- direction ending at the index @i + count - 1@.
543{-# INLINE writeSlice #-}
544writeSlice :: (IsStream t, Monad m, Storable a)
545 => Handle -> Int -> Int -> t m a -> m ()
546writeSlice h i len s = undefined
548-- | @writeSliceRev h i count stream@ writes a stream to the file handle @h@
549-- starting at index @i@ and writing up to @count@ elements in the reverse
550-- direction ending at the index @i - count + 1@.
553{-# INLINE writeSliceRev #-}
554writeSliceRev :: (IsStream t, Monad m, Storable a)
555 => Handle -> Int -> Int -> t m a -> m ()
556writeSliceRev arr i len s = undefined
keysesc to close
- j k
- move down, up
- enter
- open
- esc
- back, or up a level
- 1 2 3 4
- readme, files, commits, refs
- /
- filter
- y
- copy the clone command
- c
- compare the chosen commit with another (or shift-click both)
- ?
- show or hide this list