git.iko.soy/streamly
on 106939a help sign in
streamly/core/src/Streamly/Internal/FileSystem/Dir.hs
411 lines 14 kB history raw
1#include "inline.hs"
3-- |
4-- Module : Streamly.Internal.FileSystem.Dir
5-- Copyright : (c) 2018 Composewell Technologies
6--
7-- License : BSD3
8-- Maintainer : streamly@composewell.com
9-- Stability : pre-release
10-- Portability : GHC
12module Streamly.Internal.FileSystem.Dir
13 (
14 -- * Streams
15 read
16 , readFiles
17 , readDirs
18 , readEither
19 -- , readWithBufferOf
21 -- * Unfolds
22 -- | Use the more convenient stream APIs instead of unfolds where possible.
23 , reader
24 , fileReader
25 , dirReader
26 , eitherReader
28 {-
29 , toStreamWithBufferOf
31 , readChunks
32 , readChunksWithBufferOf
34 , toChunksWithBufferOf
35 , toChunks
37 , write
38 , writeWithBufferOf
40 -- Byte stream write (Streams)
41 , fromStream
42 , fromStreamWithBufferOf
44 -- -- * Array Write
45 , writeArray
46 , writeChunks
47 , writeChunksWithBufferOf
49 -- -- * Array stream Write
50 , fromChunks
51 , fromChunksWithBufferOf
52 -}
53 -- * Deprecated
54 , toStream
55 , toEither
56 , toFiles
57 , toDirs
58 )
59where
61import Control.Monad.IO.Class (MonadIO(..))
62import Data.Either (isRight, isLeft, fromLeft, fromRight)
63import Streamly.Data.Stream (Stream)
64import Streamly.Internal.Data.Unfold.Type (Unfold(..))
66import qualified Streamly.Data.Unfold as UF
67import qualified Streamly.Internal.Data.Unfold as UF (mapM2)
68import qualified Streamly.Data.Stream as S
69import qualified System.Directory as Dir
71import Prelude hiding (read)
73{-
74{-# INLINABLE readArrayUpto #-}
75readArrayUpto :: Int -> Handle -> IO (Array Word8)
76readArrayUpto size h = do
77 ptr <- mallocPlainForeignPtrBytes size
78 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
79 withForeignPtr ptr $ \p -> do
80 n <- hGetBufSome h p size
81 let v = Array
82 { aStart = ptr
83 , arrEnd = p `plusPtr` n
84 , arrBound = p `plusPtr` size
85 }
86 -- XXX shrink only if the diff is significant
87 shrinkToFit v
89-------------------------------------------------------------------------------
90-- Stream of Arrays IO
91-------------------------------------------------------------------------------
93-- | @toChunksWithBufferOf size h@ reads a stream of arrays from file handle @h@.
94-- The maximum size of a single array is specified by @size@. The actual size
95-- read may be less than or equal to @size@.
96{-# INLINE _toChunksWithBufferOf #-}
97_toChunksWithBufferOf :: MonadIO m => Int -> Handle -> Stream m (Array Word8)
98_toChunksWithBufferOf size h = go
99 where
100 -- XXX use cons/nil instead
101 go = mkStream $ \_ yld _ stp -> do
102 arr <- liftIO $ readArrayUpto size h
103 if A.length arr == 0
104 then stp
105 else yld arr go
107-- | @toChunksWithBufferOf size handle@ reads a stream of arrays from the file
108-- handle @handle@. The maximum size of a single array is limited to @size@.
109-- The actual size read may be less than or equal to @size@.
110--
111-- @since 0.7.0
112{-# INLINE_NORMAL toChunksWithBufferOf #-}
113toChunksWithBufferOf :: MonadIO m => Int -> Handle -> Stream m (Array Word8)
114toChunksWithBufferOf size h = D.fromStreamD (D.Stream step ())
115 where
116 {-# INLINE_LATE step #-}
117 step _ _ = do
118 arr <- liftIO $ readArrayUpto size h
119 return $
120 case A.length arr of
121 0 -> D.Stop
122 _ -> D.Yield arr ()
124-- | Unfold the tuple @(bufsize, handle)@ into a stream of 'Word8' arrays.
125-- Read requests to the IO device are performed using a buffer of size
126-- @bufsize@. The size of an array in the resulting stream is always less than
127-- or equal to @bufsize@.
128--
129-- @since 0.7.0
130{-# INLINE_NORMAL readChunksWithBufferOf #-}
131readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Handle) (Array Word8)
132readChunksWithBufferOf = Unfold step return
133 where
134 {-# INLINE_LATE step #-}
135 step (size, h) = do
136 arr <- liftIO $ readArrayUpto size h
137 return $
138 case A.length arr of
139 0 -> D.Stop
140 _ -> D.Yield arr (size, h)
142-- XXX read 'Array a' instead of Word8
143--
144-- | @toChunks handle@ reads a stream of arrays from the specified file
145-- handle. The maximum size of a single array is limited to
146-- @defaultChunkSize@. The actual size read may be less than or equal to
147-- @defaultChunkSize@.
148--
149-- > toChunks = toChunksWithBufferOf defaultChunkSize
150--
151-- @since 0.7.0
152{-# INLINE toChunks #-}
153toChunks :: MonadIO m => Handle -> Stream m (Array Word8)
154toChunks = toChunksWithBufferOf defaultChunkSize
156-- | Unfolds a handle into a stream of 'Word8' arrays. Requests to the IO
157-- device are performed using a buffer of size
158-- 'Streamly.Internal.Data.Array.Unboxed.Type.defaultChunkSize'. The
159-- size of arrays in the resulting stream are therefore less than or equal to
160-- 'Streamly.Internal.Data.Array.Unboxed.Type.defaultChunkSize'.
161--
162-- @since 0.7.0
163{-# INLINE readChunks #-}
164readChunks :: MonadIO m => Unfold m Handle (Array Word8)
165readChunks = UF.first readChunksWithBufferOf defaultChunkSize
167-------------------------------------------------------------------------------
168-- Read File to Stream
169-------------------------------------------------------------------------------
171-- TODO for concurrent streams implement readahead IO. We can send multiple
172-- read requests at the same time. For serial case we can use async IO. We can
173-- also control the read throughput in mbps or IOPS.
175-- | Unfolds the tuple @(bufsize, handle)@ into a byte stream, read requests
176-- to the IO device are performed using buffers of @bufsize@.
177--
178-- @since 0.7.0
179{-# INLINE readWithBufferOf #-}
180readWithBufferOf :: MonadIO m => Unfold m (Int, Handle) Word8
181readWithBufferOf = UF.many readChunksWithBufferOf A.read
183-- | @toStreamWithBufferOf bufsize handle@ reads a byte stream from a file
184-- handle, reads are performed in chunks of up to @bufsize@.
185--
186-- /Pre-release/
187{-# INLINE toStreamWithBufferOf #-}
188toStreamWithBufferOf :: MonadIO m => Int -> Handle -> Stream m Word8
189toStreamWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h
190-}
192-- XXX exception handling
193-- | Raw read of a directory
194--
195-- /Internal/
196--
197{-# INLINE reader #-}
198reader :: MonadIO m => Unfold m String String
199reader =
200 -- XXX use proper streaming read of the dir
201 UF.lmapM (liftIO . Dir.getDirectoryContents) UF.fromList
203-- XXX We can use a more general mechanism to filter the contents of a
204-- directory. We can just stat each child and pass on the stat information. We
205-- can then use that info to do a general filtering. "find" like filters can be
206-- created.
208-- | Read directories as Left and files as Right. Filter out "." and ".."
209-- entries.
210--
211-- /Internal/
212--
213{-# INLINE eitherReader #-}
214eitherReader :: MonadIO m => Unfold m String (Either String String)
215eitherReader =
216 UF.mapM2 classify
217 $ UF.filter (\x -> x /= "." && x /= "..")
218 -- XXX use proper streaming read of the dir
219 $ UF.lmapM (liftIO . Dir.getDirectoryContents) UF.fromList
220 where
221 classify dir x = do
222 r <- liftIO $ Dir.doesDirectoryExist (dir ++ "/" ++ x)
223 return $ if r then Left x else Right x
225--
226-- | Read files only.
227--
228-- /Internal/
229--
230{-# INLINE fileReader #-}
231fileReader :: MonadIO m => Unfold m String String
232fileReader = fmap (fromRight undefined) $ UF.filter isRight eitherReader
234-- | Read directories only. Filter out "." and ".." entries.
235--
236-- /Internal/
237--
238{-# INLINE dirReader #-}
239dirReader :: MonadIO m => Unfold m String String
240dirReader = fmap (fromLeft undefined) $ UF.filter isLeft eitherReader
242-- | Raw read of a directory.
243--
244-- /Pre-release/
245{-# INLINE read #-}
246read :: MonadIO m => String -> Stream m String
247read = S.unfold reader
249{-# DEPRECATED toStream "Please use 'read' instead" #-}
250{-# INLINE toStream #-}
251toStream :: MonadIO m => String -> Stream m String
252toStream = read
254-- | Read directories as Left and files as Right. Filter out "." and ".."
255-- entries.
256--
257-- /Pre-release/
258{-# INLINE readEither #-}
259readEither :: MonadIO m => String -> Stream m (Either String String)
260readEither = S.unfold eitherReader
262{-# DEPRECATED toEither "Please use 'readEither' instead" #-}
263{-# INLINE toEither #-}
264toEither :: MonadIO m => String -> Stream m (Either String String)
265toEither = readEither
267-- | Read files only.
268--
269-- /Internal/
270--
271{-# INLINE readFiles #-}
272readFiles :: MonadIO m => String -> Stream m String
273readFiles = S.unfold fileReader
275{-# DEPRECATED toFiles "Please use 'readFiles' instead" #-}
276{-# INLINE toFiles #-}
277toFiles :: MonadIO m => String -> Stream m String
278toFiles = readFiles
280-- | Read directories only.
281--
282-- /Internal/
283--
284{-# INLINE readDirs #-}
285readDirs :: MonadIO m => String -> Stream m String
286readDirs = S.unfold dirReader
288{-# DEPRECATED toDirs "Please use 'readDirs' instead" #-}
289{-# INLINE toDirs #-}
290toDirs :: MonadIO m => String -> Stream m String
291toDirs = readDirs
293{-
294-------------------------------------------------------------------------------
295-- Writing
296-------------------------------------------------------------------------------
298-------------------------------------------------------------------------------
299-- Array IO (output)
300-------------------------------------------------------------------------------
302-- | Write an 'Array' to a file handle.
303--
304-- @since 0.7.0
305{-# INLINABLE writeArray #-}
306writeArray :: Storable a => Handle -> Array a -> IO ()
307writeArray _ arr | A.length arr == 0 = return ()
308writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen
309 where
310 aLen =
311 let p = unsafeForeignPtrToPtr aStart
312 in arrEnd `minusPtr` p
314-------------------------------------------------------------------------------
315-- Stream of Arrays IO
316-------------------------------------------------------------------------------
318-------------------------------------------------------------------------------
319-- Writing
320-------------------------------------------------------------------------------
322-- | Write a stream of arrays to a handle.
323--
324-- @since 0.7.0
325{-# INLINE fromChunks #-}
326fromChunks :: (MonadIO m, Storable a)
327 => Handle -> Stream m (Array a) -> m ()
328fromChunks h m = S.mapM_ (liftIO . writeArray h) m
330-- | @fromChunksWithBufferOf bufsize handle stream@ writes a stream of arrays
331-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
332-- The chunk size is only a maximum and the actual writes could be smaller as
333-- we do not split the arrays to fit exactly to the specified size.
334--
335-- @since 0.7.0
336{-# INLINE fromChunksWithBufferOf #-}
337fromChunksWithBufferOf :: (MonadIO m, Storable a)
338 => Int -> Handle -> Stream m (Array a) -> m ()
339fromChunksWithBufferOf n h xs = fromChunks h $ AS.compact n xs
341-- | @fromStreamWithBufferOf bufsize handle stream@ writes @stream@ to @handle@
342-- in chunks of @bufsize@. A write is performed to the IO device as soon as we
343-- collect the required input size.
344--
345-- @since 0.7.0
346{-# INLINE fromStreamWithBufferOf #-}
347fromStreamWithBufferOf :: MonadIO m => Int -> Handle -> Stream m Word8 -> m ()
348fromStreamWithBufferOf n h m = fromChunks h $ S.arraysOf n m
349-- fromStreamWithBufferOf n h m = fromChunks h $ AS.arraysOf n m
351-- > write = 'writeWithBufferOf' A.defaultChunkSize
352--
353-- | Write a byte stream to a file handle. Accumulates the input in chunks of
354-- up to 'Streamly.Internal.Data.Array.Unboxed.Type.defaultChunkSize' before writing.
355--
356-- NOTE: This may perform better than the 'write' fold, you can try this if you
357-- need some extra perf boost.
358--
359-- @since 0.7.0
360{-# INLINE fromStream #-}
361fromStream :: MonadIO m => Handle -> Stream m Word8 -> m ()
362fromStream = fromStreamWithBufferOf defaultChunkSize
364-- | Write a stream of arrays to a handle. Each array in the stream is written
365-- to the device as a separate IO request.
366--
367-- @since 0.7.0
368{-# INLINE writeChunks #-}
369writeChunks :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
370writeChunks h = FL.drainBy (liftIO . writeArray h)
372-- | @writeChunksWithBufferOf bufsize handle@ writes a stream of arrays
373-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
374-- We never split an array, if a single array is bigger than the specified size
375-- it emitted as it is. Multiple arrays are coalesed as long as the total size
376-- remains below the specified size.
377--
378-- @since 0.7.0
379{-# INLINE writeChunksWithBufferOf #-}
380writeChunksWithBufferOf :: (MonadIO m, Storable a)
381 => Int -> Handle -> Fold m (Array a) ()
382writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)
384-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
385--
386-- XXX test this
387-- Note that if you use a chunk size less than 8K (GHC's default buffer
388-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
389-- do not want buffering to occur at GHC level as well. Same thing applies to
390-- writes as well.
392-- | @writeWithBufferOf reqSize handle@ writes the input stream to @handle@.
393-- Bytes in the input stream are collected into a buffer until we have a chunk
394-- of @reqSize@ and then written to the IO device.
395--
396-- @since 0.7.0
397{-# INLINE writeWithBufferOf #-}
398writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
399writeWithBufferOf n h = FL.chunksOf n (writeNUnsafe n) (writeChunks h)
401-- > write = 'writeWithBufferOf' A.defaultChunkSize
402--
403-- | Write a byte stream to a file handle. Accumulates the input in chunks of
404-- up to 'Streamly.Internal.Data.Array.Unboxed.Type.defaultChunkSize' before writing
405-- to the IO device.
406--
407-- @since 0.7.0
408{-# INLINE write #-}
409write :: MonadIO m => Handle -> Fold m Word8 ()
410write = writeWithBufferOf defaultChunkSize
411-}