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