git.iko.soy/streamly
on 745265e help sign in
streamly/src/Streamly/Internal/FileSystem/Dir.hs
414 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 : experimental
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)
55-- import Data.Word (Word8)
56-- import Foreign.ForeignPtr (withForeignPtr)
57-- import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
58-- import Foreign.Ptr (minusPtr, plusPtr)
59-- import Foreign.Storable (Storable(..))
60-- import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
61-- import System.IO (Handle, hGetBufSome, hPutBuf)
62import Prelude hiding (read)
64-- import Streamly.Data.Fold (Fold)
65import Streamly.Internal.Data.Unfold.Types (Unfold(..))
66-- import Streamly.Internal.Memory.Array.Types
67-- (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,
68-- lpackArraysChunksOf)
69-- import Streamly.Internal.Data.Stream.Serial (SerialT)
70import Streamly.Internal.Data.Stream.StreamK.Type (IsStream)
71-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
73-- import qualified Streamly.Data.Fold as FL
74-- import qualified Streamly.Internal.Data.Fold.Types as FL
75import qualified Streamly.Internal.Data.Unfold as UF
76-- import qualified Streamly.Internal.Memory.ArrayStream as AS
77import qualified Streamly.Internal.Prelude as S
78-- import qualified Streamly.Memory.Array as A
79-- import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
80import qualified System.Directory as Dir
82#if MIN_VERSION_base(4,10,0)
83import Data.Either (fromRight, fromLeft)
84#else
85fromLeft :: a -> Either a b -> a
86fromLeft _ (Left a) = a
87fromLeft a _ = a
89fromRight :: b -> Either a b -> b
90fromRight _ (Right b) = b
91fromRight b _ = b
92#endif
94{-
95{-# INLINABLE readArrayUpto #-}
96readArrayUpto :: Int -> Handle -> IO (Array Word8)
97readArrayUpto size h = do
98 ptr <- mallocPlainForeignPtrBytes size
99 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
100 withForeignPtr ptr $ \p -> do
101 n <- hGetBufSome h p size
102 let v = Array
103 { aStart = ptr
104 , aEnd = p `plusPtr` n
105 , aBound = p `plusPtr` size
106 }
107 -- XXX shrink only if the diff is significant
108 shrinkToFit v
110-------------------------------------------------------------------------------
111-- Stream of Arrays IO
112-------------------------------------------------------------------------------
114-- | @toChunksWithBufferOf size h@ reads a stream of arrays from file handle @h@.
115-- The maximum size of a single array is specified by @size@. The actual size
116-- read may be less than or equal to @size@.
117{-# INLINABLE _toChunksWithBufferOf #-}
118_toChunksWithBufferOf :: (IsStream t, MonadIO m)
119 => Int -> Handle -> t m (Array Word8)
120_toChunksWithBufferOf size h = go
121 where
122 -- XXX use cons/nil instead
123 go = mkStream $ \_ yld _ stp -> do
124 arr <- liftIO $ readArrayUpto size h
125 if A.length arr == 0
126 then stp
127 else yld arr go
129-- | @toChunksWithBufferOf size handle@ reads a stream of arrays from the file
130-- handle @handle@. The maximum size of a single array is limited to @size@.
131-- The actual size read may be less than or equal to @size@.
132--
133-- @since 0.7.0
134{-# INLINE_NORMAL toChunksWithBufferOf #-}
135toChunksWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
136toChunksWithBufferOf size h = D.fromStreamD (D.Stream step ())
137 where
138 {-# INLINE_LATE step #-}
139 step _ _ = do
140 arr <- liftIO $ readArrayUpto size h
141 return $
142 case A.length arr of
143 0 -> D.Stop
144 _ -> D.Yield arr ()
146-- | Unfold the tuple @(bufsize, handle)@ into a stream of 'Word8' arrays.
147-- Read requests to the IO device are performed using a buffer of size
148-- @bufsize@. The size of an array in the resulting stream is always less than
149-- or equal to @bufsize@.
150--
151-- @since 0.7.0
152{-# INLINE_NORMAL readChunksWithBufferOf #-}
153readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Handle) (Array Word8)
154readChunksWithBufferOf = Unfold step return
155 where
156 {-# INLINE_LATE step #-}
157 step (size, h) = do
158 arr <- liftIO $ readArrayUpto size h
159 return $
160 case A.length arr of
161 0 -> D.Stop
162 _ -> D.Yield arr (size, h)
164-- XXX read 'Array a' instead of Word8
165--
166-- | @toChunks handle@ reads a stream of arrays from the specified file
167-- handle. The maximum size of a single array is limited to
168-- @defaultChunkSize@. The actual size read may be less than or equal to
169-- @defaultChunkSize@.
170--
171-- > toChunks = toChunksWithBufferOf defaultChunkSize
172--
173-- @since 0.7.0
174{-# INLINE toChunks #-}
175toChunks :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
176toChunks = toChunksWithBufferOf defaultChunkSize
178-- | Unfolds a handle into a stream of 'Word8' arrays. Requests to the IO
179-- device are performed using a buffer of size
180-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'. The
181-- size of arrays in the resulting stream are therefore less than or equal to
182-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
183--
184-- @since 0.7.0
185{-# INLINE readChunks #-}
186readChunks :: MonadIO m => Unfold m Handle (Array Word8)
187readChunks = UF.supplyFirst readChunksWithBufferOf defaultChunkSize
189-------------------------------------------------------------------------------
190-- Read File to Stream
191-------------------------------------------------------------------------------
193-- TODO for concurrent streams implement readahead IO. We can send multiple
194-- read requests at the same time. For serial case we can use async IO. We can
195-- also control the read throughput in mbps or IOPS.
197-- | Unfolds the tuple @(bufsize, handle)@ into a byte stream, read requests
198-- to the IO device are performed using buffers of @bufsize@.
199--
200-- @since 0.7.0
201{-# INLINE readWithBufferOf #-}
202readWithBufferOf :: MonadIO m => Unfold m (Int, Handle) Word8
203readWithBufferOf = UF.concat readChunksWithBufferOf A.read
205-- | @toStreamWithBufferOf bufsize handle@ reads a byte stream from a file
206-- handle, reads are performed in chunks of up to @bufsize@.
207--
208-- /Internal/
209{-# INLINE toStreamWithBufferOf #-}
210toStreamWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
211toStreamWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h
212-}
214-- XXX exception handling
215-- | Raw read of a directory
216--
217-- /Internal/
218--
219{-# INLINE read #-}
220read :: MonadIO m => Unfold m String String
221read =
222 -- XXX use proper streaming read of the dir
223 UF.lmapM (liftIO . Dir.getDirectoryContents) UF.fromList
225-- XXX We can use a more general mechanism to filter the contents of a
226-- directory. We can just stat each child and pass on the stat information. We
227-- can then use that info to do a general filtering. "find" like filters can be
228-- created.
230-- | Read directories as Left and files as Right. Filter out "." and ".."
231-- entries.
232--
233-- /Internal/
234--
235{-# INLINE readEither #-}
236readEither :: MonadIO m => Unfold m String (Either String String)
237readEither =
238 UF.mapMWithInput classify
239 $ UF.filter (\x -> x /= "." && x /= "..")
240 -- XXX use proper streaming read of the dir
241 $ UF.lmapM (liftIO . Dir.getDirectoryContents) UF.fromList
242 where
243 classify dir x = do
244 r <- liftIO $ Dir.doesDirectoryExist (dir ++ "/" ++ x)
245 return $ if r then Left x else Right x
247--
248-- | Read files only.
249--
250-- /Internal/
251--
252{-# INLINE readFiles #-}
253readFiles :: MonadIO m => Unfold m String String
254readFiles = UF.map (fromRight undefined) $ UF.filter isRight readEither
256-- | Read directories only. Filter out "." and ".." entries.
257--
258-- /Internal/
259--
260{-# INLINE readDirs #-}
261readDirs :: MonadIO m => Unfold m String String
262readDirs = UF.map (fromLeft undefined) $ UF.filter isLeft readEither
264-- | Raw read of a directory.
265--
266-- /Internal/
267{-# INLINE toStream #-}
268toStream :: (IsStream t, MonadIO m) => String -> t m String
269toStream = S.unfold read
271-- | Read directories as Left and files as Right. Filter out "." and ".."
272-- entries.
273--
274-- /Internal/
275{-# INLINE toEither #-}
276toEither :: (IsStream t, MonadIO m)
277 => String -> t m (Either String String)
278toEither = S.unfold readEither
280-- | Read files only.
281--
282-- /Internal/
283--
284{-# INLINE toFiles #-}
285toFiles :: (IsStream t, MonadIO m) => String -> t m String
286toFiles = S.unfold readFiles
288-- | Read directories only.
289--
290-- /Internal/
291--
292{-# INLINE toDirs #-}
293toDirs :: (IsStream t, MonadIO m) => String -> t m String
294toDirs = S.unfold readDirs
296{-
297-------------------------------------------------------------------------------
298-- Writing
299-------------------------------------------------------------------------------
301-------------------------------------------------------------------------------
302-- Array IO (output)
303-------------------------------------------------------------------------------
305-- | Write an 'Array' to a file handle.
306--
307-- @since 0.7.0
308{-# INLINABLE writeArray #-}
309writeArray :: Storable a => Handle -> Array a -> IO ()
310writeArray _ arr | A.length arr == 0 = return ()
311writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen
312 where
313 aLen =
314 let p = unsafeForeignPtrToPtr aStart
315 in aEnd `minusPtr` p
317-------------------------------------------------------------------------------
318-- Stream of Arrays IO
319-------------------------------------------------------------------------------
321-------------------------------------------------------------------------------
322-- Writing
323-------------------------------------------------------------------------------
325-- | Write a stream of arrays to a handle.
326--
327-- @since 0.7.0
328{-# INLINE fromChunks #-}
329fromChunks :: (MonadIO m, Storable a)
330 => Handle -> SerialT m (Array a) -> m ()
331fromChunks h m = S.mapM_ (liftIO . writeArray h) m
333-- | @fromChunksWithBufferOf bufsize handle stream@ writes a stream of arrays
334-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
335-- The chunk size is only a maximum and the actual writes could be smaller as
336-- we do not split the arrays to fit exactly to the specified size.
337--
338-- @since 0.7.0
339{-# INLINE fromChunksWithBufferOf #-}
340fromChunksWithBufferOf :: (MonadIO m, Storable a)
341 => Int -> Handle -> SerialT m (Array a) -> m ()
342fromChunksWithBufferOf n h xs = fromChunks h $ AS.compact n xs
344-- | @fromStreamWithBufferOf bufsize handle stream@ writes @stream@ to @handle@
345-- in chunks of @bufsize@. A write is performed to the IO device as soon as we
346-- collect the required input size.
347--
348-- @since 0.7.0
349{-# INLINE fromStreamWithBufferOf #-}
350fromStreamWithBufferOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
351fromStreamWithBufferOf n h m = fromChunks h $ S.arraysOf n m
352-- fromStreamWithBufferOf n h m = fromChunks h $ AS.arraysOf n m
354-- > write = 'writeWithBufferOf' A.defaultChunkSize
355--
356-- | Write a byte stream to a file handle. Accumulates the input in chunks of
357-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing.
358--
359-- NOTE: This may perform better than the 'write' fold, you can try this if you
360-- need some extra perf boost.
361--
362-- @since 0.7.0
363{-# INLINE fromStream #-}
364fromStream :: MonadIO m => Handle -> SerialT m Word8 -> m ()
365fromStream = fromStreamWithBufferOf defaultChunkSize
367-- | Write a stream of arrays to a handle. Each array in the stream is written
368-- to the device as a separate IO request.
369--
370-- @since 0.7.0
371{-# INLINE writeChunks #-}
372writeChunks :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
373writeChunks h = FL.drainBy (liftIO . writeArray h)
375-- | @writeChunksWithBufferOf bufsize handle@ writes a stream of arrays
376-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
377-- We never split an array, if a single array is bigger than the specified size
378-- it emitted as it is. Multiple arrays are coalesed as long as the total size
379-- remains below the specified size.
380--
381-- @since 0.7.0
382{-# INLINE writeChunksWithBufferOf #-}
383writeChunksWithBufferOf :: (MonadIO m, Storable a)
384 => Int -> Handle -> Fold m (Array a) ()
385writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)
387-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
388--
389-- XXX test this
390-- Note that if you use a chunk size less than 8K (GHC's default buffer
391-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
392-- do not want buffering to occur at GHC level as well. Same thing applies to
393-- writes as well.
395-- | @writeWithBufferOf reqSize handle@ writes the input stream to @handle@.
396-- Bytes in the input stream are collected into a buffer until we have a chunk
397-- of @reqSize@ and then written to the IO device.
398--
399-- @since 0.7.0
400{-# INLINE writeWithBufferOf #-}
401writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
402writeWithBufferOf n h = FL.lchunksOf n (writeNUnsafe n) (writeChunks h)
404-- > write = 'writeWithBufferOf' A.defaultChunkSize
405--
406-- | Write a byte stream to a file handle. Accumulates the input in chunks of
407-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing
408-- to the IO device.
409--
410-- @since 0.7.0
411{-# INLINE write #-}
412write :: MonadIO m => Handle -> Fold m Word8 ()
413write = writeWithBufferOf defaultChunkSize
414-}