@@ -92,7 +92,6 @@ module Streamly.Internal.Data.Array.Foreign.Mut.Type
93 -- * Stream of arrays
93 -- * Stream of arrays
94 , arraysOf
94 , arraysOf
96 , writeChunks
95 , writeChunks
98 -- * Utilities
97 -- * Utilities
@@ -527,10 +526,10 @@ arraysOf n (D.Stream step state) =
528-- XXX buffer to a list instead?
527-- XXX buffer to a list instead?
529-- | Buffer the stream into arrays in memory.
528-- | Buffer the stream into arrays in memory.
530−{-# INLINE bufferChunks #-}
529+{-# INLINE arrayStreamKFromStreamD #-}
531−bufferChunks :: (MonadIO m, Storable a) =>
530+arrayStreamKFromStreamD :: (MonadIO m, Storable a) =>
532 D.Stream m a -> m (K.Stream m (Array a))
531 D.Stream m a -> m (K.Stream m (Array a))
533−bufferChunks m = D.foldr K.cons K.nil $ arraysOf defaultChunkSize m
532+arrayStreamKFromStreamD m = D.foldr K.cons K.nil $ arraysOf defaultChunkSize m
535-------------------------------------------------------------------------------
534-------------------------------------------------------------------------------
536-- Streams of arrays - Flattening
535-- Streams of arrays - Flattening
@@ -898,17 +897,19 @@ writeNUnsafe = writeNUnsafeWith newArray
899-- | Buffer a stream into a stream of arrays.
898-- | Buffer a stream into a stream of arrays.
901−-- @writeChunks = Fold.many Fold.toStream (Array.writeN n)@
900+-- >>> writeChunks n = FL.many (writeN n) FL.toStreamK
903-- Breaking an array into an array stream can be useful to consume a large
902-- Breaking an array into an array stream can be useful to consume a large
904-- array sequentially such that memory of the array is released incrementatlly.
903-- array sequentially such that memory of the array is released incrementatlly.
905+-- See also: 'arrayStreamKFromStreamD'.
906-- /Unimplemented/
907-- /Unimplemented/
908{-# INLINE_NORMAL writeChunks #-}
909{-# INLINE_NORMAL writeChunks #-}
909−writeChunks :: -- (MonadIO m, Storable a) =>
910+writeChunks :: (MonadIO m, Storable a) =>
910− Int -> Fold m a (D.Stream m (Array a))
911+ Int -> Fold m a (K.Stream n (Array a))
911−writeChunks = undefined -- Fold.many Fold.toStream (Array.writeN n)
912+writeChunks n = FL.many (writeN n) FL.toStreamK
913-- XXX Compare toArrayMinChunk with fromStreamD which uses an array of streams
914-- XXX Compare toArrayMinChunk with fromStreamD which uses an array of streams
914-- implementation. We can write this using writeChunks above if that is faster.
915-- implementation. We can write this using writeChunks above if that is faster.
@@ -1018,6 +1019,20 @@ fromListN n xs = fromStreamDN n $ D.fromList xs
1018-- convert stream to a single array
1019-- convert stream to a single array
1019-------------------------------------------------------------------------------
1020-------------------------------------------------------------------------------
1022+{-# INLINE arrayStreamKLength #-}
1023+arrayStreamKLength :: (Monad m, Storable a) => K.Stream m (Array a) -> m Int
1024+arrayStreamKLength as = K.foldl' (+) 0 (K.map length as)
1026+-- | Convert an array stream to an array. Note that this requires peak memory
1027+-- that is double the size of the array stream.
1029+{-# INLINE fromArrayStreamK #-}
1030+fromArrayStreamK :: (Storable a, MonadIO m) =>
1031+ K.Stream m (Array a) -> m (Array a)
1032+fromArrayStreamK as = do
1033+ len <- arrayStreamKLength as
1034+ fromStreamDN len $ D.unfoldMany read $ D.fromStreamK as
1021-- CAUTION: a very large number (millions) of arrays can degrade performance
1036-- CAUTION: a very large number (millions) of arrays can degrade performance
1022-- due to GC overhead because we need to buffer the arrays before we flatten
1037-- due to GC overhead because we need to buffer the arrays before we flatten
1023-- all the arrays.
1038-- all the arrays.
@@ -1035,10 +1050,7 @@ fromListN n xs = fromStreamDN n $ D.fromList xs
1036{-# INLINE fromStreamD #-}
1051{-# INLINE fromStreamD #-}
1037fromStreamD :: (MonadIO m, Storable a) => D.Stream m a -> m (Array a)
1052fromStreamD :: (MonadIO m, Storable a) => D.Stream m a -> m (Array a)
1038−fromStreamD m = do
1053+fromStreamD m = arrayStreamKFromStreamD m >>= fromArrayStreamK
1039− buffered <- bufferChunks m
1040− len <- K.foldl' (+) 0 (K.map length buffered)
1041− fromStreamDN len $ D.unfoldMany read $ D.fromStreamK buffered
1043-- | Create an 'Array' from a list. The list must be of finite size.
1055-- | Create an 'Array' from a list. The list must be of finite size.