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