@@ -8,6 +8,11 @@
88-- Maintainer : streamly@composewell.com
99-- Stability : experimental
1010-- Portability : GHC
11+--
12+-- The fundamental singleton APIs are 'getChunk' and 'putChunk' and the
13+-- fundamental stream APIs built on top of those are 'readChunksWithBufferOf'
14+-- and 'writeChunks'. Rest of this module is just combinatorial programming
15+-- using these.
1116
1217module Streamly.Internal.FileSystem.Handle
1318 (
@@ -22,7 +27,7 @@ module Streamly.Internal.FileSystem.Handle
2227 , toBytesWithBufferOf
2328
2429 -- -- * Array Read
25− -- , readArrayUpto
30+ , getChunk
2631 -- , readArrayOf
2732 , readChunks
2833 , readChunksWithBufferOf
@@ -45,7 +50,7 @@ module Streamly.Internal.FileSystem.Handle
4550 , putBytesWithBufferOf
4651
4752 -- -- * Array Write
48− , writeArray
53+ , putChunk
4954 , writeChunks
5055 , writeChunksWithBufferOf
5156
@@ -75,8 +80,8 @@ module Streamly.Internal.FileSystem.Handle
7580
7681 -- , readChunksFrom
7782 -- , readChunksFromTo
78− -- , readChunksFromToWithBufferOf
79− -- , readChunksFromThenToWithBufferOf
83+ -- , readChunksFromToWith
84+ -- , readChunksFromThenToWith
8085
8186 -- , writeIndex
8287 -- , writeFrom -- start writing at the given position
@@ -88,8 +93,8 @@ module Streamly.Internal.FileSystem.Handle
8893 --
8994 -- , writeChunksFrom
9095 -- , writeChunksFromTo
91− -- , writeChunksFromToWithBufferOf
92− -- , writeChunksFromThenToWithBufferOf
96+ -- , writeChunksFromToWith
97+ -- , writeChunksFromThenToWith
9398 )
9499where
95100
@@ -124,6 +129,18 @@ import qualified Streamly.Internal.Data.Stream.IsStream as S
124129import qualified Streamly.Data.Array.Foreign as A
125130import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
126131
132+-- $setup
133+-- >>> import qualified Streamly.Data.Array.Foreign as Array
134+-- >>> import qualified Streamly.Data.Fold as Fold
135+-- >>> import qualified Streamly.Data.Unfold as Unfold
136+-- >>> import qualified Streamly.FileSystem.Handle as Handle
137+-- >>> import qualified Streamly.Prelude as Stream
138+--
139+-- >>> import qualified Streamly.Internal.Data.Stream.IsStream as Stream
140+-- >>> import qualified Streamly.Internal.Data.Unfold as Unfold (supplyFirst)
141+-- >>> import qualified Streamly.Internal.FileSystem.Handle as Handle
142+-- >>> import qualified Streamly.Internal.System.IO as IO (defaultChunkSize)
143+
127144-------------------------------------------------------------------------------
128145-- References
129146-------------------------------------------------------------------------------
@@ -145,9 +162,9 @@ import qualified Streamly.Internal.Data.Stream.StreamD.Type as D
145162-- handle it blocks until some data becomes available. If data is available
146163-- then it immediately returns that data without blocking. It reads a maximum
147164-- of up to the size requested.
148−{-# INLINABLE readArrayUpto #-}
149−readArrayUpto :: Int -> Handle -> IO (Array Word8)
150−readArrayUpto size h = do
165+{-# INLINABLE getChunk #-}
166+getChunk :: Int -> Handle -> IO (Array Word8)
167+getChunk size h = do
151168 ptr <- mallocPlainForeignPtrBytes size
152169 -- ptr <- mallocPlainForeignPtrAlignedBytes size (alignment (undefined :: Word8))
153170 withForeignPtr ptr $ \p -> do
@@ -171,7 +188,7 @@ _toChunksWithBufferOf size h = go
171188 where
172189 -- XXX use cons/nil instead
173190 go = mkStream $ \_ yld _ stp -> do
174− arr <- liftIO $ readArrayUpto size h
191+ arr <- liftIO $ getChunk size h
175192 if A.length arr == 0
176193 then stp
177194 else yld arr go
@@ -180,14 +197,17 @@ _toChunksWithBufferOf size h = go
180197-- handle @handle@. The maximum size of a single array is limited to @size@.
181198-- The actual size read may be less than or equal to @size@.
182199--
200+-- >>> toChunksWithBufferOf size h = Stream.unfold Handle.readChunksWithBufferOf (size, h)
201+--
183202-- @since 0.7.0
184203{-# INLINE_NORMAL toChunksWithBufferOf #-}
185−toChunksWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m (Array Word8)
204+toChunksWithBufferOf :: (IsStream t, MonadIO m) =>
205+ Int -> Handle -> t m (Array Word8)
186206toChunksWithBufferOf size h = D.fromStreamD (D.Stream step ())
187207 where
188208 {-# INLINE_LATE step #-}
189209 step _ _ = do
190− arr <- liftIO $ readArrayUpto size h
210+ arr <- liftIO $ getChunk size h
191211 return $
192212 case A.length arr of
193213 0 -> D.Stop
@@ -205,7 +225,7 @@ readChunksWithBufferOf = Unfold step return
205225 where
206226 {-# INLINE_LATE step #-}
207227 step (size, h) = do
208− arr <- liftIO $ readArrayUpto size h
228+ arr <- liftIO $ getChunk size h
209229 return $
210230 case A.length arr of
211231 0 -> D.Stop
@@ -218,7 +238,7 @@ readChunksWithBufferOf = Unfold step return
218238-- @defaultChunkSize@. The actual size read may be less than or equal to
219239-- @defaultChunkSize@.
220240--
221−-- > toChunks = toChunksWithBufferOf defaultChunkSize
241+-- >>> toChunks = Handle.toChunksWithBufferOf IO.defaultChunkSize
222242--
223243-- @since 0.7.0
224244{-# INLINE toChunks #-}
@@ -231,6 +251,8 @@ toChunks = toChunksWithBufferOf defaultChunkSize
231251-- size of arrays in the resulting stream are therefore less than or equal to
232252-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'.
233253--
254+-- >>> readChunks = Unfold.supplyFirst IO.defaultChunkSize Handle.readChunksWithBufferOf
255+--
234256-- @since 0.7.0
235257{-# INLINE readChunks #-}
236258readChunks :: MonadIO m => Unfold m Handle (Array Word8)
@@ -247,6 +269,8 @@ readChunks = UF.supplyFirst defaultChunkSize readChunksWithBufferOf
247269-- | Unfolds the tuple @(bufsize, handle)@ into a byte stream, read requests
248270-- to the IO device are performed using buffers of @bufsize@.
249271--
272+-- >>> readWithBufferOf = Unfold.many Handle.readChunksWithBufferOf Array.read
273+--
250274-- @since 0.7.0
251275{-# INLINE readWithBufferOf #-}
252276readWithBufferOf :: MonadIO m => Unfold m (Int, Handle) Word8
@@ -255,10 +279,12 @@ readWithBufferOf = UF.many readChunksWithBufferOf A.read
255279-- | @toBytesWithBufferOf bufsize handle@ reads a byte stream from a file
256280-- handle, reads are performed in chunks of up to @bufsize@.
257281--
282+-- >>> toBytesWithBufferOf size h = Stream.unfoldMany Array.read $ Handle.toChunksWithBufferOf size h
283+--
258284-- /Pre-release/
259285{-# INLINE toBytesWithBufferOf #-}
260286toBytesWithBufferOf :: (IsStream t, MonadIO m) => Int -> Handle -> t m Word8
261−toBytesWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h
287+toBytesWithBufferOf size h = AS.concat $ toChunksWithBufferOf size h
262288
263289-- TODO
264290-- Generate a stream of elements of the given type from a file 'Handle'.
@@ -268,13 +294,17 @@ toBytesWithBufferOf chunkSize h = AS.concat $ toChunksWithBufferOf chunkSize h
268294-- performed in sizes of
269295-- 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize'.
270296--
297+-- >>> read = Unfold.many Handle.readChunks Array.read
298+--
271299-- @since 0.7.0
272300{-# INLINE read #-}
273301read :: MonadIO m => Unfold m Handle Word8
274−read = UF.supplyFirst defaultChunkSize readWithBufferOf
302+read = UF.many readChunks A.read
275303
276304-- | Generate a byte stream from a file 'Handle'.
277305--
306+-- >>> toBytes h = Stream.unfoldMany Array.read $ Handle.toChunks h
307+--
278308-- /Pre-release/
279309{-# INLINE toBytes #-}
280310toBytes :: (IsStream t, MonadIO m) => Handle -> t m Word8
@@ -291,11 +321,14 @@ toBytes = AS.concat . toChunks
291321-- | Write an 'Array' to a file handle.
292322--
293323-- @since 0.7.0
294−{-# INLINABLE writeArray #-}
295−writeArray :: Storable a => Handle -> Array a -> IO ()
296−writeArray _ arr | A.length arr == 0 = return ()
297−writeArray h Array{..} = unsafeWithForeignPtr aStart $ \p -> hPutBuf h p aLen
324+{-# INLINABLE putChunk #-}
325+putChunk :: (MonadIO m, Storable a) => Handle -> Array a -> m ()
326+putChunk _ arr | A.length arr == 0 = return ()
327+putChunk h Array{..} =
328+ liftIO $ unsafeWithForeignPtr aStart $ \p -> hPutBuf h p aLen
329+
298330 where
331+
299332 aLen =
300333 let p = unsafeForeignPtrToPtr aStart
301334 in aEnd `minusPtr` p
@@ -311,12 +344,18 @@ writeArray h Array{..} = unsafeWithForeignPtr aStart $ \p -> hPutBuf h p aLen
311344-- XXX use an unfold to fromObjects or fromUnfold so that we can put any object
312345-- | Write a stream of arrays to a handle.
313346--
347+-- >>> putChunks h = Stream.mapM_ (Handle.putChunk h)
348+--
314349-- @since 0.7.0
315350{-# INLINE putChunks #-}
316351putChunks :: (MonadIO m, Storable a)
317352 => Handle -> SerialT m (Array a) -> m ()
318−putChunks h = S.mapM_ (liftIO . writeArray h)
353+putChunks h = S.mapM_ (putChunk h)
319354
355+-- XXX AS.compact can be written idiomatically in terms of foldMany, just like
356+-- AS.concat is written in terms of foldMany. Once that is done we can write
357+-- idiomatic def in the docs.
358+--
320359-- | @putChunksWithBufferOf bufsize handle stream@ writes a stream of arrays
321360-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
322361-- The chunk size is only a maximum and the actual writes could be smaller as
@@ -332,20 +371,22 @@ putChunksWithBufferOf n h xs = putChunks h $ AS.compact n xs
332371-- in chunks of @bufsize@. A write is performed to the IO device as soon as we
333372-- collect the required input size.
334373--
374+-- >>> putBytesWithBufferOf n h m = Handle.putChunks h $ Stream.arraysOf n m
375+--
335376-- @since 0.7.0
336377{-# INLINE putBytesWithBufferOf #-}
337378putBytesWithBufferOf :: MonadIO m => Int -> Handle -> SerialT m Word8 -> m ()
338379putBytesWithBufferOf n h m = putChunks h $ S.arraysOf n m
339380-- putBytesWithBufferOf n h m = putChunks h $ AS.arraysOf n m
340381
341−-- > write = 'writeWithBufferOf' A.defaultChunkSize
342−--
343382-- | Write a byte stream to a file handle. Accumulates the input in chunks of
344383-- up to 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize' before writing.
345384--
346385-- NOTE: This may perform better than the 'write' fold, you can try this if you
347386-- need some extra perf boost.
348387--
388+-- >>> putBytes = Handle.putBytesWithBufferOf IO.defaultChunkSize
389+--
349390-- @since 0.7.0
350391{-# INLINE putBytes #-}
351392putBytes :: MonadIO m => Handle -> SerialT m Word8 -> m ()
@@ -354,15 +395,22 @@ putBytes = putBytesWithBufferOf defaultChunkSize
354395-- | Write a stream of arrays to a handle. Each array in the stream is written
355396-- to the device as a separate IO request.
356397--
398+-- writeChunks h = Fold.drainBy (Handle.putChunk h)
399+--
357400-- @since 0.7.0
358401{-# INLINE writeChunks #-}
359402writeChunks :: (MonadIO m, Storable a) => Handle -> Fold m (Array a) ()
360−writeChunks h = FL.drainBy (liftIO . writeArray h)
403+writeChunks h = FL.drainBy (putChunk h)
361404
362405{-# INLINE writeChunks2 #-}
363406writeChunks2 :: (MonadIO m, Storable a) => Fold2 m Handle (Array a) ()
364−writeChunks2 = Fold2 (\h arr -> liftIO $ writeArray h arr >> return h) return (\_ -> return ())
407+writeChunks2 =
408+ Fold2 (\h arr -> liftIO $ putChunk h arr >> return h)
409+ return
410+ (\_ -> return ())
365411
412+-- XXX lpackArraysChunksOf should be written idiomatically
413+--
366414-- | @writeChunksWithBufferOf bufsize handle@ writes a stream of arrays
367415-- to @handle@ after coalescing the adjacent arrays in chunks of @bufsize@.
368416-- We never split an array, if a single array is bigger than the specified size
@@ -383,10 +431,14 @@ writeChunksWithBufferOf n h = lpackArraysChunksOf n (writeChunks h)
383431-- do not want buffering to occur at GHC level as well. Same thing applies to
384432-- writes as well.
385433
434+-- XXX Maybe we should have a Fold.arraysOf like we have Stream.arraysOf
435+--
386436-- | @writeWithBufferOf reqSize handle@ writes the input stream to @handle@.
387437-- Bytes in the input stream are collected into a buffer until we have a chunk
388438-- of @reqSize@ and then written to the IO device.
389439--
440+-- >>> writeWithBufferOf n h = Fold.chunksOf n (Array.writeNUnsafe n) (Handle.writeChunks h)
441+--
390442-- @since 0.7.0
391443{-# INLINE writeWithBufferOf #-}
392444writeWithBufferOf :: MonadIO m => Int -> Handle -> Fold m Word8 ()
@@ -396,12 +448,12 @@ writeWithBufferOf n h = FL.chunksOf n (writeNUnsafe n) (writeChunks h)
396448writeWithBufferOf2 :: MonadIO m => Int -> Fold2 m Handle Word8 ()
397449writeWithBufferOf2 n = FL.chunksOf2 n (writeNUnsafe n) writeChunks2
398450
399−-- > write = 'writeWithBufferOf' A.defaultChunkSize
400−--
401451-- | Write a byte stream to a file handle. Accumulates the input in chunks of
402452-- up to 'Streamly.Internal.Data.Array.Foreign.Type.defaultChunkSize' before writing
403453-- to the IO device.
404454--
455+-- >>> write = Handle.writeWithBufferOf IO.defaultChunkSize
456+--
405457-- @since 0.7.0
406458{-# INLINE write #-}
407459write :: MonadIO m => Handle -> Fold m Word8 ()