@@ -64,8 +64,9 @@ module Streamly.Internal.Memory.Array
6464 -- list.
6565
6666 , A.toList
67− , A.read
68− , readRev
67+ , toStream
68+ , toStreamRev
69+ , read
6970
7071 -- * Random Access
7172 , length
@@ -96,19 +97,25 @@ where
9697
9798import Control.Monad.IO.Class (MonadIO(..))
9899-- import Data.Functor.Identity (Identity)
99−import Foreign.ForeignPtr (withForeignPtr)
100+import Foreign.ForeignPtr (withForeignPtr, touchForeignPtr)
101+import Foreign.Ptr (plusPtr)
100102import Foreign.Storable (Storable(..))
101103import Prelude hiding (length, null, last, map, (!!), read, concat)
102104
105+import GHC.ForeignPtr (ForeignPtr(..))
106+import GHC.Ptr (Ptr(..))
107+
108+import Streamly.Internal.Data.Fold.Types (Fold(..))
109+import Streamly.Internal.Data.Unfold.Types (Unfold(..))
103110import Streamly.Internal.Memory.Array.Types (Array(..), length)
104111import Streamly.Streams.Serial (SerialT)
105112import Streamly.Streams.StreamK.Type (IsStream)
106−import Streamly.Internal.Data.Fold.Types (Fold(..))
107113
108114import qualified Streamly.Internal.Memory.Array.Types as A
109115import qualified Streamly.Streams.Prelude as P
110116import qualified Streamly.Streams.Serial as Serial
111117import qualified Streamly.Streams.StreamD as D
118+import qualified Streamly.Streams.StreamK as K
112119
113120-------------------------------------------------------------------------------
114121-- Construction
@@ -151,16 +158,56 @@ fromStream = P.runFold A.write
151158-- Elimination
152159-------------------------------------------------------------------------------
153160
161+-- | Convert an 'Array' into a stream.
162+--
163+-- /Internal/
164+{-# INLINE_EARLY toStream #-}
165+toStream :: (Monad m, K.IsStream t, Storable a) => Array a -> t m a
166+toStream = D.fromStreamD . A.toStreamD
167+-- XXX add fallback to StreamK rule
168+-- {-# RULES "Streamly.Array.read fallback to StreamK" [1]
169+-- forall a. S.readK (read a) = K.fromArray a #-}
170+
154171-- | Convert an 'Array' into a stream in reverse order.
155172--
156−-- @since 0.7.0
157−{-# INLINE_EARLY readRev #-}
158−readRev :: (Monad m, IsStream t, Storable a) => Array a -> t m a
159−readRev = D.fromStreamD . A.toStreamDRev
173+-- /Internal/
174+{-# INLINE_EARLY toStreamRev #-}
175+toStreamRev :: (Monad m, IsStream t, Storable a) => Array a -> t m a
176+toStreamRev = D.fromStreamD . A.toStreamDRev
160177-- XXX add fallback to StreamK rule
161178-- {-# RULES "Streamly.Array.readRev fallback to StreamK" [1]
162179-- forall a. S.toStreamK (readRev a) = K.revFromArray a #-}
163180
181+data ReadUState a = ReadUState
182+ {-# UNPACK #-} !(ForeignPtr a) -- foreign ptr with end of array pointer
183+ {-# UNPACK #-} !(Ptr a) -- current pointer
184+
185+-- | Unfold an array into a stream.
186+--
187+-- @since 0.7.0
188+{-# INLINE_NORMAL read #-}
189+read :: forall m a. (Monad m, Storable a) => Unfold m (Array a) a
190+read = Unfold step inject
191+ where
192+
193+ inject (Array (ForeignPtr start contents) (Ptr end) _) =
194+ return $ ReadUState (ForeignPtr end contents) (Ptr start)
195+
196+ {-# INLINE_LATE step #-}
197+ step (ReadUState fp@(ForeignPtr end _) p) | p == (Ptr end) =
198+ let x = A.unsafeInlineIO $ touchForeignPtr fp
199+ in x `seq` return D.Stop
200+ step (ReadUState fp p) = do
201+ -- unsafeInlineIO allows us to run this in Identity monad for pure
202+ -- toList/foldr case which makes them much faster due to not
203+ -- accumulating the list and fusing better with the pure consumers.
204+ --
205+ -- This should be safe as the array contents are guaranteed to be
206+ -- evaluated/written to before we peek at them.
207+ let !x = A.unsafeInlineIO $ peek p
208+ return $ D.Yield x
209+ (ReadUState fp (p `plusPtr` (sizeOf (undefined :: a))))
210+
164211-- | > null arr = length arr == 0
165212--
166213-- /Internal/
@@ -375,18 +422,18 @@ streamTransform :: forall m a b. (MonadIO m, Storable a, Storable b)
375422 => (SerialT m a -> SerialT m b) -> Array a -> m (Array b)
376423streamTransform f arr =
377424 P.runFold (A.toArrayMinChunk (alignment (undefined :: a)) (length arr))
378− $ f (A.read arr)
425+ $ f (toStream arr)
379426
380427-- | Fold an array using a 'Fold'.
381428--
382429-- /Internal/
383430{-# INLINE fold #-}
384431fold :: forall m a b. (MonadIO m, Storable a) => Fold m a b -> Array a -> m b
385−fold f arr = P.runFold f $ (A.read arr :: Serial.SerialT m a)
432+fold f arr = P.runFold f $ (toStream arr :: Serial.SerialT m a)
386433
387434-- | Fold an array using a stream fold operation.
388435--
389436-- /Internal/
390437{-# INLINE streamFold #-}
391438streamFold :: (MonadIO m, Storable a) => (SerialT m a -> m b) -> Array a -> m b
392−streamFold f arr = f (A.read arr)
439+streamFold f arr = f (toStream arr)