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