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