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