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