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