@@ -0,0 +1,349 @@
1+{-# OPTIONS_HADDOCK hide #-}
2+{-# LANGUAGE CPP #-}
3+{-# LANGUAGE BangPatterns #-}
4+{-# LANGUAGE MagicHash #-}
5+{-# LANGUAGE RecordWildCards #-}
6+{-# LANGUAGE UnboxedTuples #-}
7+
8+#include "inline.hs"
9+
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
18+
19+module Streamly.Internal.FileSystem.Dir
20+ (
21+ -- ** Read from Directory
22+ read
23+ , readFiles
24+ {-
25+ , readWithBufferOf
26+
27+ , toStream
28+ , toStreamWithBufferOf
29+
30+ , readChunks
31+ , readChunksWithBufferOf
32+
33+ , toChunksWithBufferOf
34+ , toChunks
35+
36+ , write
37+ , writeWithBufferOf
38+
39+ -- Byte stream write (Streams)
40+ , fromStream
41+ , fromStreamWithBufferOf
42+
43+ -- -- * Array Write
44+ , writeArray
45+ , writeChunks
46+ , writeChunksWithBufferOf
47+
48+ -- -- * Array stream Write
49+ , fromChunks
50+ , fromChunksWithBufferOf
51+ -}
52+ )
53+where
54+
55+import Control.Monad.IO.Class (MonadIO(..))
56+import Data.Word (Word8)
57+import Foreign.ForeignPtr (withForeignPtr)
58+import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr)
59+import Foreign.Ptr (minusPtr, plusPtr)
60+import Foreign.Storable (Storable(..))
61+import GHC.ForeignPtr (mallocPlainForeignPtrBytes)
62+import System.IO (Handle, hGetBufSome, hPutBuf)
63+import Prelude hiding (read)
64+
65+import Streamly.Data.Fold (Fold)
66+import Streamly.Internal.Data.Unfold.Types (Unfold(..))
67+import Streamly.Internal.Memory.Array.Types
68+ (Array(..), writeNUnsafe, defaultChunkSize, shrinkToFit,
69+ lpackArraysChunksOf)
70+import Streamly.Streams.Serial (SerialT)
71+import Streamly.Streams.StreamK.Type (IsStream, mkStream)
72+-- import Streamly.String (encodeUtf8, decodeUtf8, foldLines)
73+
74+import qualified Streamly.Data.Fold as FL
75+import qualified Streamly.Internal.Data.Fold.Types as FL
76+import qualified Streamly.Internal.Data.Unfold as UF
77+import qualified Streamly.Internal.Memory.ArrayStream as AS
78+import qualified Streamly.Internal.Prelude as S
79+import qualified Streamly.Memory.Array as A
80+import qualified Streamly.Streams.StreamD.Type as D
81+import qualified System.Directory as Dir
82+
83+{-
84+{-# INLINABLE readArrayUpto #-}
85+readArrayUpto :: Int -> Handle -> IO (Array Word8)
86+readArrayUpto 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
98+
99+-------------------------------------------------------------------------------
100+-- Stream of Arrays IO
101+-------------------------------------------------------------------------------
102+
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
117+
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 #-}
124+toChunksWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
125+toChunksWithBufferOf 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 ()
134+
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 #-}
142+readChunksWithBufferOf :: MonadIO m => Unfold m (Int, Handle) (Array Word8)
143+readChunksWithBufferOf = 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)
152+
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 #-}
164+toChunks :: (IsStream t, MonadIO m) => Handle -> t m (Array Word8)
165+toChunks = toChunksWithBufferOf defaultChunkSize
166+
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.Memory.Array.Types.defaultChunkSize'. The
170+-- size of arrays in the resulting stream are therefore less than or equal to
171+-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
172+--
173+-- @since 0.7.0
174+{-# INLINE readChunks #-}
175+readChunks :: MonadIO m => Unfold m Handle (Array Word8)
176+readChunks = UF.supplyFirst readChunksWithBufferOf defaultChunkSize
177+
178+-------------------------------------------------------------------------------
179+-- Read File to Stream
180+-------------------------------------------------------------------------------
181+
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.
185+
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 #-}
191+readWithBufferOf :: MonadIO m => Unfold m (Int, Handle) Word8
192+readWithBufferOf = UF.concat readChunksWithBufferOf A.read
193+
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 #-}
199+toStreamWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
200+toStreamWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h
201+-}
202+
203+-- TODO
204+-- Generate a stream of elements of the given type from a file 'Handle'.
205+-- read :: (IsStream t, MonadIO m, Storable a) => Handle -> t m a
206+--
207+-- | Unfolds a file handle into a byte stream. IO requests to the device are
208+-- performed in sizes of
209+-- 'Streamly.Internal.Memory.Array.Types.defaultChunkSize'.
210+--
211+-- @since 0.7.0
212+-- XXX exception handling
213+-- XXX this is not stream read
214+{-# INLINE read #-}
215+read :: MonadIO m => Unfold m String String
216+read = UF.lmapM (liftIO . Dir.getDirectoryContents) UF.fromList
217+
218+{-# INLINE readFiles #-}
219+readFiles :: MonadIO m => Unfold m String String
220+readFiles = UF.filterM (fmap not . liftIO . Dir.doesDirectoryExist) read
221+
222+{-
223+-- | Generate a byte stream from a file 'Handle'.
224+--
225+-- /Internal/
226+{-# INLINE toStream #-}
227+toStream :: (IsStream t, MonadIO m) => Handle -> t m Word8
228+toStream = AS.concat . toChunks
229+-}
230+
231+{-
232+-------------------------------------------------------------------------------
233+-- Writing
234+-------------------------------------------------------------------------------
235+
236+-------------------------------------------------------------------------------
237+-- Array IO (output)
238+-------------------------------------------------------------------------------
239+
240+-- | Write an 'Array' to a file handle.
241+--
242+-- @since 0.7.0
243+{-# INLINABLE writeArray #-}
244+writeArray :: Storable a => Handle -> Array a -> IO ()
245+writeArray _ arr | A.length arr == 0 = return ()
246+writeArray h Array{..} = withForeignPtr aStart $ \p -> hPutBuf h p aLen
247+ where
248+ aLen =
249+ let p = unsafeForeignPtrToPtr aStart
250+ in aEnd `minusPtr` p
251+
252+-------------------------------------------------------------------------------
253+-- Stream of Arrays IO
254+-------------------------------------------------------------------------------
255+
256+-------------------------------------------------------------------------------
257+-- Writing
258+-------------------------------------------------------------------------------
259+
260+-- | Write a stream of arrays to a handle.
261+--
262+-- @since 0.7.0
263+{-# INLINE fromChunks #-}
264+fromChunks :: (MonadIO m, Storable a)
265+ => Handle -> SerialT m (Array a) -> m ()
266+fromChunks h m = S.mapM_ (liftIO . writeArray h) m
267+
268+-- | @fromChunksWithBufferOf bufsize handle stream@ writes a stream of arrays
269+-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
270+-- The chunk size is only a maximum and the actual writes could be smaller as
271+-- we do not split the arrays to fit exactly to the specified size.
272+--
273+-- @since 0.7.0
274+{-# INLINE fromChunksWithBufferOf #-}
275+fromChunksWithBufferOf :: (MonadIO m, Storable a)
276+ => Int -> Handle -> SerialT m (Array a) -> m ()
277+fromChunksWithBufferOf n h xs = fromChunks h $ AS.compact n xs
278+
279+-- | @fromStreamWithBufferOf bufsize handle stream@ writes @stream@ to @handle@
280+-- in chunks of @bufsize@. A write is performed to the IO device as soon as we
281+-- collect the required input size.
282+--
283+-- @since 0.7.0
284+{-# INLINE fromStreamWithBufferOf #-}
285+fromStreamWithBufferOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
286+fromStreamWithBufferOf n h m = fromChunks h $ S.arraysOf n m
287+-- fromStreamWithBufferOf n h m = fromChunks h $ AS.arraysOf n m
288+
289+-- > write = 'writeWithBufferOf' A.defaultChunkSize
290+--
291+-- | Write a byte stream to a file handle. Accumulates the input in chunks of
292+-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing.
293+--
294+-- NOTE: This may perform better than the 'write' fold, you can try this if you
295+-- need some extra perf boost.
296+--
297+-- @since 0.7.0
298+{-# INLINE fromStream #-}
299+fromStream :: MonadIO m => Handle -> SerialT m Word8 -> m ()
300+fromStream = fromStreamWithBufferOf defaultChunkSize
301+
302+-- | Write a stream of arrays to a handle. Each array in the stream is written
303+-- to the device as a separate IO request.
304+--
305+-- @since 0.7.0
306+{-# INLINE writeChunks #-}
307+writeChunks :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
308+writeChunks h = FL.drainBy (liftIO . writeArray h)
309+
310+-- | @writeChunksWithBufferOf bufsize handle@ writes a stream of arrays
311+-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
312+-- We never split an array, if a single array is bigger than the specified size
313+-- it emitted as it is. Multiple arrays are coalesed as long as the total size
314+-- remains below the specified size.
315+--
316+-- @since 0.7.0
317+{-# INLINE writeChunksWithBufferOf #-}
318+writeChunksWithBufferOf :: (MonadIO m, Storable a)
319+ => Int -> Handle -> Fold m (Array a) ()
320+writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)
321+
322+-- GHC buffer size dEFAULT_FD_BUFFER_SIZE=8192 bytes.
323+--
324+-- XXX test this
325+-- Note that if you use a chunk size less than 8K (GHC's default buffer
326+-- size) then you are advised to use 'NOBuffering' mode on the 'Handle' in case you
327+-- do not want buffering to occur at GHC level as well. Same thing applies to
328+-- writes as well.
329+
330+-- | @writeWithBufferOf reqSize handle@ writes the input stream to @handle@.
331+-- Bytes in the input stream are collected into a buffer until we have a chunk
332+-- of @reqSize@ and then written to the IO device.
333+--
334+-- @since 0.7.0
335+{-# INLINE writeWithBufferOf #-}
336+writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
337+writeWithBufferOf n h = FL.lchunksOf n (writeNUnsafe n) (writeChunks h)
338+
339+-- > write = 'writeWithBufferOf' A.defaultChunkSize
340+--
341+-- | Write a byte stream to a file handle. Accumulates the input in chunks of
342+-- up to 'Streamly.Internal.Memory.Array.Types.defaultChunkSize' before writing
343+-- to the IO device.
344+--
345+-- @since 0.7.0
346+{-# INLINE write #-}
347+write :: MonadIO m => Handle -> Fold m Word8 ()
348+write = writeWithBufferOf defaultChunkSize
349+-}