@@ -64,8 +64,9 @@ module Streamly.Internal.Memory.Array
66 , A.toList
66 , A.toList
70 -- * Random Access
71 -- * Random Access
@@ -96,19 +97,25 @@ where
97import Control.Monad.IO.Class (MonadIO(..))
98import Control.Monad.IO.Class (MonadIO(..))
98-- import Data.Functor.Identity (Identity)
99-- import Data.Functor.Identity (Identity)
99−import Foreign.ForeignPtr (withForeignPtr)
100+import Foreign.ForeignPtr (withForeignPtr, touchForeignPtr)
101+import Foreign.Ptr (plusPtr)
100import Foreign.Storable (Storable(..))
102import Foreign.Storable (Storable(..))
101import Prelude hiding (length, null, last, map, (!!), read, concat)
103import Prelude hiding (length, null, last, map, (!!), read, concat)
105+import GHC.ForeignPtr (ForeignPtr(..))
106+import GHC.Ptr (Ptr(..))
108+import Streamly.Internal.Data.Fold.Types (Fold(..))
109+import Streamly.Internal.Data.Unfold.Types (Unfold(..))
103import Streamly.Internal.Memory.Array.Types (Array(..), length)
110import Streamly.Internal.Memory.Array.Types (Array(..), length)
104import Streamly.Streams.Serial (SerialT)
111import Streamly.Streams.Serial (SerialT)
105import Streamly.Streams.StreamK.Type (IsStream)
112import Streamly.Streams.StreamK.Type (IsStream)
106−import Streamly.Internal.Data.Fold.Types (Fold(..))
108import qualified Streamly.Internal.Memory.Array.Types as A
114import qualified Streamly.Internal.Memory.Array.Types as A
109import qualified Streamly.Streams.Prelude as P
115import qualified Streamly.Streams.Prelude as P
110import qualified Streamly.Streams.Serial as Serial
116import qualified Streamly.Streams.Serial as Serial
111import qualified Streamly.Streams.StreamD as D
117import qualified Streamly.Streams.StreamD as D
118+import qualified Streamly.Streams.StreamK as K
113-------------------------------------------------------------------------------
120-------------------------------------------------------------------------------
114-- Construction
121-- Construction
@@ -151,16 +158,56 @@ fromStream = P.runFold A.write
151-- Elimination
158-- Elimination
152-------------------------------------------------------------------------------
159-------------------------------------------------------------------------------
161+-- | Convert an 'Array' into a stream.
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 #-}
154-- | Convert an 'Array' into a stream in reverse order.
171-- | Convert an 'Array' into a stream in reverse order.
156−-- @since 0.7.0
173+-- /Internal/
157−{-# INLINE_EARLY readRev #-}
174+{-# INLINE_EARLY toStreamRev #-}
158−readRev :: (Monad m, IsStream t, Storable a) => Array a -> t m a
175+toStreamRev :: (Monad m, IsStream t, Storable a) => Array a -> t m a
159−readRev = D.fromStreamD . A.toStreamDRev
176+toStreamRev = D.fromStreamD . A.toStreamDRev
160-- XXX add fallback to StreamK rule
177-- XXX add fallback to StreamK rule
161-- {-# RULES "Streamly.Array.readRev fallback to StreamK" [1]
178-- {-# RULES "Streamly.Array.readRev fallback to StreamK" [1]
162-- forall a. S.toStreamK (readRev a) = K.revFromArray a #-}
179-- forall a. S.toStreamK (readRev a) = K.revFromArray a #-}
181+data ReadUState a = ReadUState
182+ {-# UNPACK #-} !(ForeignPtr a) -- foreign ptr with end of array pointer
183+ {-# UNPACK #-} !(Ptr a) -- current pointer
185+-- | Unfold an array into a stream.
188+{-# INLINE_NORMAL read #-}
189+read :: forall m a. (Monad m, Storable a) => Unfold m (Array a) a
190+read = Unfold step inject
193+ inject (Array (ForeignPtr start contents) (Ptr end) _) =
194+ return $ ReadUState (ForeignPtr end contents) (Ptr start)
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.
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
209+ (ReadUState fp (p `plusPtr` (sizeOf (undefined :: a))))
164-- | > null arr = length arr == 0
211-- | > null arr = length arr == 0
166-- /Internal/
213-- /Internal/
@@ -375,18 +422,18 @@ streamTransform :: forall m a b. (MonadIO m, Storable a, Storable b)
375 => (SerialT m a -> SerialT m b) -> Array a -> m (Array b)
422 => (SerialT m a -> SerialT m b) -> Array a -> m (Array b)
376streamTransform f arr =
423streamTransform f arr =
377 P.runFold (A.toArrayMinChunk (alignment (undefined :: a)) (length arr))
424 P.runFold (A.toArrayMinChunk (alignment (undefined :: a)) (length arr))
378− $ f (A.read arr)
425+ $ f (toStream arr)
380-- | Fold an array using a 'Fold'.
427-- | Fold an array using a 'Fold'.
382-- /Internal/
429-- /Internal/
383{-# INLINE fold #-}
430{-# INLINE fold #-}
384fold :: forall m a b. (MonadIO m, Storable a) => Fold m a b -> Array a -> m b
431fold :: 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)
387-- | Fold an array using a stream fold operation.
434-- | Fold an array using a stream fold operation.
389-- /Internal/
436-- /Internal/
390{-# INLINE streamFold #-}
437{-# INLINE streamFold #-}
391streamFold :: (MonadIO m, Storable a) => (SerialT m a -> m b) -> Array a -> m b
438streamFold :: (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)