git.iko.soy/streamly
on master help sign in
Make changes for "reader" based impl of concatChunks
ffea484Harendra Kumar
core/src/Streamly/Internal/Data/Array.hs−51
compare with master
unifiedsplit
@@ -23,11 +23,6 @@ module Streamly.Internal.Data.Array
23 -- Monadic Folds
23 -- Monadic Folds
24 , writeLastN
24 , writeLastN
25
25
26− -- * Unfolds
27− , reader
28− , readerUnsafe
29− , producer -- experimental
30−
31 -- * Random Access
26 -- * Random Access
32 -- , (!!)
27 -- , (!!)
33 , getIndex
28 , getIndex
@@ -99,7 +94,6 @@ import Prelude hiding (length, null, last, map, (!!), read, concat)
99import Streamly.Internal.Data.MutByteArray.Type (PinnedState(..))
94import Streamly.Internal.Data.MutByteArray.Type (PinnedState(..))
100import Streamly.Internal.Data.Serialize.Type (Serialize)
95import Streamly.Internal.Data.Serialize.Type (Serialize)
101import Streamly.Internal.Data.Fold.Type (Fold(..))
96import Streamly.Internal.Data.Fold.Type (Fold(..))
102−import Streamly.Internal.Data.Producer.Type (Producer(..))
103import Streamly.Internal.Data.Stream (Stream)
97import Streamly.Internal.Data.Stream (Stream)
104import Streamly.Internal.Data.Tuple.Strict (Tuple3Fused'(..))
98import Streamly.Internal.Data.Tuple.Strict (Tuple3Fused'(..))
105import Streamly.Internal.Data.Unfold.Type (Unfold(..))
99import Streamly.Internal.Data.Unfold.Type (Unfold(..))
@@ -110,8 +104,6 @@ import qualified Streamly.Internal.Data.MutByteArray.Type as MBA
110import qualified Streamly.Internal.Data.MutArray as MA
104import qualified Streamly.Internal.Data.MutArray as MA
111import qualified Streamly.Internal.Data.Array.Type as A
105import qualified Streamly.Internal.Data.Array.Type as A
112import qualified Streamly.Internal.Data.Fold as FL
106import qualified Streamly.Internal.Data.Fold as FL
113−import qualified Streamly.Internal.Data.Producer.Type as Producer
114−import qualified Streamly.Internal.Data.Producer as Producer
115import qualified Streamly.Internal.Data.Ring as RB
107import qualified Streamly.Internal.Data.Ring as RB
116import qualified Streamly.Internal.Data.Stream as D
108import qualified Streamly.Internal.Data.Stream as D
117import qualified Streamly.Internal.Data.Stream as Stream
109import qualified Streamly.Internal.Data.Stream as Stream
@@ -148,49 +140,6 @@ import Streamly.Internal.Data.Array.Type
148-- Elimination
140-- Elimination
149-------------------------------------------------------------------------------
141-------------------------------------------------------------------------------
150
142
151−{-# INLINE_NORMAL producer #-}
152−producer :: forall m a. (Monad m, Unbox a) => Producer m (Array a) a
153−producer =
154− Producer.translate A.unsafeThaw A.unsafeFreeze
155− $ MA.producerWith (return . unsafeInlineIO)
156−
157−-- | Unfold an array into a stream.
158−--
159−{-# INLINE_NORMAL reader #-}
160−reader :: forall m a. (Monad m, Unbox a) => Unfold m (Array a) a
161−reader = Producer.simplify producer
162−
163−-- | Unfold an array into a stream, does not check the end of the array, the
164−-- user is responsible for terminating the stream within the array bounds. For
165−-- high performance application where the end condition can be determined by
166−-- a terminating fold.
167−--
168−-- Written in the hope that it may be faster than "read", however, in the case
169−-- for which this was written, "read" proves to be faster even though the core
170−-- generated with unsafeRead looks simpler.
171−--
172−-- /Pre-release/
173−--
174−{-# INLINE_NORMAL readerUnsafe #-}
175−readerUnsafe :: forall m a. (Monad m, Unbox a) => Unfold m (Array a) a
176−readerUnsafe = Unfold step inject
177− where
178−
179− inject (Array contents start end) =
180− return (ArrayUnsafe contents end start)
181−
182− {-# INLINE_LATE step #-}
183− step (ArrayUnsafe contents end p) = do
184− -- unsafeInlineIO allows us to run this in Identity monad for pure
185− -- toList/foldr case which makes them much faster due to not
186− -- accumulating the list and fusing better with the pure consumers.
187− --
188− -- This should be safe as the array contents are guaranteed to be
189− -- evaluated/written to before we peek at them.
190− let !x = unsafeInlineIO $ peekAt p contents
191− let !p1 = INDEX_NEXT(p,a)
192− return $ D.Yield x (ArrayUnsafe contents end p1)
193−
194-- |
143-- |
195--
144--
196-- >>> null arr = Array.byteLength arr == 0
145-- >>> null arr = Array.byteLength arr == 0
core/src/Streamly/Internal/Data/Array/Type.hs+56−7
@@ -54,6 +54,11 @@ module Streamly.Internal.Data.Array.Type
54 , clone
54 , clone
55 , pinnedClone
55 , pinnedClone
56
56
57+ -- * Unfolds
58+ , reader
59+ , readerUnsafe
60+ , producer -- experimental
61+
57 -- ** Elimination
62 -- ** Elimination
58 , unsafeIndexIO
63 , unsafeIndexIO
59 , getIndexUnsafe
64 , getIndexUnsafe
@@ -76,7 +81,7 @@ module Streamly.Internal.Data.Array.Type
76 , pinnedWriteN
81 , pinnedWriteN
77 , writeNUnsafe
82 , writeNUnsafe
78 , pinnedWriteNUnsafe
83 , pinnedWriteNUnsafe
79− , MA.ArrayUnsafe (..)
84+ -- , MA.ArrayUnsafe (..)
80 , pinnedWriteNAligned
85 , pinnedWriteNAligned
81 , write
86 , write
82 , pinnedWrite
87 , pinnedWrite
@@ -121,6 +126,7 @@ import GHC.Exts (IsList, IsString(..), Addr#)
121
126
122import GHC.IO (unsafePerformIO)
127import GHC.IO (unsafePerformIO)
123import GHC.Ptr (Ptr(..))
128import GHC.Ptr (Ptr(..))
129+import Streamly.Internal.Data.Producer.Type (Producer(..))
124import Streamly.Internal.Data.MutArray.Type (MutArray(..))
130import Streamly.Internal.Data.MutArray.Type (MutArray(..))
125import Streamly.Internal.Data.MutByteArray.Type (MutByteArray)
131import Streamly.Internal.Data.MutByteArray.Type (MutByteArray)
126import Streamly.Internal.Data.Fold.Type (Fold(..))
132import Streamly.Internal.Data.Fold.Type (Fold(..))
@@ -137,6 +143,7 @@ import qualified Streamly.Internal.Data.MutArray.Type as MA
137import qualified Streamly.Internal.Data.Stream.Type as D
143import qualified Streamly.Internal.Data.Stream.Type as D
138import qualified Streamly.Internal.Data.StreamK.Type as K
144import qualified Streamly.Internal.Data.StreamK.Type as K
139import qualified Streamly.Internal.Data.MutByteArray.Type as Unboxed
145import qualified Streamly.Internal.Data.MutByteArray.Type as Unboxed
146+import qualified Streamly.Internal.Data.Producer as Producer
140import qualified Streamly.Internal.Data.Unfold.Type as Unfold
147import qualified Streamly.Internal.Data.Unfold.Type as Unfold
141import qualified Text.ParserCombinators.ReadPrec as ReadPrec
148import qualified Text.ParserCombinators.ReadPrec as ReadPrec
142
149
@@ -414,16 +421,15 @@ pinnedChunksOf :: forall m a. (MonadIO m, Unbox a)
414 => Int -> D.Stream m a -> D.Stream m (Array a)
421 => Int -> D.Stream m a -> D.Stream m (Array a)
415pinnedChunksOf n str = D.map unsafeFreeze $ MA.pinnedChunksOf n str
422pinnedChunksOf n str = D.map unsafeFreeze $ MA.pinnedChunksOf n str
416
423
417−-- | Use the "reader" unfold instead.
424+-- | Convert a stream of arrays into a stream of their elements.
418−--
419−-- @concatChunks = unfoldMany reader@
420--
425--
421−-- We can try this if there are any fusion issues in the unfold.
426+-- >>> concatChunks = Stream.unfoldMany Array.reader
422--
427--
423{-# INLINE_NORMAL concatChunks #-}
428{-# INLINE_NORMAL concatChunks #-}
424−concatChunks :: forall m a. (MonadIO m, Unbox a)
429+concatChunks :: (MonadIO m, Unbox a) => Stream m (Array a) -> Stream m a
425− => D.Stream m (Array a) -> D.Stream m a
430+-- XXX this requires MonadIO whereas the unfoldMany version does not
426concatChunks = MA.concatChunks . D.map unsafeThaw
431concatChunks = MA.concatChunks . D.map unsafeThaw
432+-- concatChunks = D.unfoldMany reader
427
433
428{-# DEPRECATED flattenArrays "Please use \"unfoldMany reader\" instead." #-}
434{-# DEPRECATED flattenArrays "Please use \"unfoldMany reader\" instead." #-}
429flattenArrays :: forall m a. (MonadIO m, Unbox a)
435flattenArrays :: forall m a. (MonadIO m, Unbox a)
@@ -488,6 +494,49 @@ byteLength = MA.byteLength . unsafeThaw
488length :: Unbox a => Array a -> Int
494length :: Unbox a => Array a -> Int
489length arr = MA.length (unsafeThaw arr)
495length arr = MA.length (unsafeThaw arr)
490
496
497+{-# INLINE_NORMAL producer #-}
498+producer :: forall m a. (Monad m, Unbox a) => Producer m (Array a) a
499+producer =
500+ Producer.translate unsafeThaw unsafeFreeze
501+ $ MA.producerWith (return . unsafeInlineIO)
502+
503+-- | Unfold an array into a stream.
504+--
505+{-# INLINE_NORMAL reader #-}
506+reader :: forall m a. (Monad m, Unbox a) => Unfold m (Array a) a
507+reader = Producer.simplify producer
508+
509+-- | Unfold an array into a stream, does not check the end of the array, the
510+-- user is responsible for terminating the stream within the array bounds. For
511+-- high performance application where the end condition can be determined by
512+-- a terminating fold.
513+--
514+-- Written in the hope that it may be faster than "read", however, in the case
515+-- for which this was written, "read" proves to be faster even though the core
516+-- generated with unsafeRead looks simpler.
517+--
518+-- /Pre-release/
519+--
520+{-# INLINE_NORMAL readerUnsafe #-}
521+readerUnsafe :: forall m a. (Monad m, Unbox a) => Unfold m (Array a) a
522+readerUnsafe = Unfold step inject
523+ where
524+
525+ inject (Array contents start end) =
526+ return (MA.ArrayUnsafe contents end start)
527+
528+ {-# INLINE_LATE step #-}
529+ step (MA.ArrayUnsafe contents end p) = do
530+ -- unsafeInlineIO allows us to run this in Identity monad for pure
531+ -- toList/foldr case which makes them much faster due to not
532+ -- accumulating the list and fusing better with the pure consumers.
533+ --
534+ -- This should be safe as the array contents are guaranteed to be
535+ -- evaluated/written to before we peek at them.
536+ let !x = unsafeInlineIO $ peekAt p contents
537+ let !p1 = INDEX_NEXT(p,a)
538+ return $ D.Yield x (MA.ArrayUnsafe contents end p1)
539+
491-- | Unfold an array into a stream in reverse order.
540-- | Unfold an array into a stream in reverse order.
492--
541--
493{-# INLINE_NORMAL readerRev #-}
542{-# INLINE_NORMAL readerRev #-}