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