157c63f · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Implement Array/Stream funcs in Array/MutArray

157c63fHarendra Kumar3 years ago

core/src/DocTestDataArray.hs+1

compare with master

unifiedsplit

@@ -16,4 +16,5 @@

16For APIs that have not been released yet.

18>>> import qualified Streamly.Internal.Data.Array as Array

19+>>> import qualified Streamly.Internal.Data.Stream as Stream

19-}

20-}

core/src/Streamly/Internal/Data/Array.hs+95−1

@@ -66,6 +66,17 @@ module Streamly.Internal.Data.Array

66 , streamFold

67 , fold

69+ -- * Stream of Arrays

70+

71+ -- XXX these are probably not very useful to have in this module as we can

72+ -- express these idiomatically using streams.

73+ , interpose

74+ , interposeSuffix

75+ , intercalateSuffix

76+

77+ , compactOnByte

78+ , compactOnByteSuffix

79+

69 -- * Serialization

80 -- * Serialization

70 , encodeAs

81 , encodeAs

71 , serialize

82 , serialize

@@ -93,10 +104,11 @@ import Foreign.Storable (Storable)

93import Streamly.Internal.Data.Unbox (Unbox(..))

104import Streamly.Internal.Data.Unbox (Unbox(..))

94import Prelude hiding (length, null, last, map, (!!), read, concat)

105import Prelude hiding (length, null, last, map, (!!), read, concat)

106

96−import Streamly.Internal.Data.MutByteArray.Type (PinnedState(..))

107+import Streamly.Internal.Data.MutByteArray.Type (PinnedState(..), MutByteArray)

97import Streamly.Internal.Data.Serialize.Type (Serialize)

108import Streamly.Internal.Data.Serialize.Type (Serialize)

98import Streamly.Internal.Data.Fold.Type (Fold(..))

109import Streamly.Internal.Data.Fold.Type (Fold(..))

99import Streamly.Internal.Data.Stream (Stream)

110import Streamly.Internal.Data.Stream (Stream)

111+import Streamly.Internal.Data.SVar.Type (adaptState)

100import Streamly.Internal.Data.Tuple.Strict (Tuple3Fused'(..))

112import Streamly.Internal.Data.Tuple.Strict (Tuple3Fused'(..))

101import Streamly.Internal.Data.Unfold.Type (Unfold(..))

113import Streamly.Internal.Data.Unfold.Type (Unfold(..))

102import Streamly.Internal.System.IO (unsafeInlineIO)

114import Streamly.Internal.System.IO (unsafeInlineIO)

@@ -539,3 +551,85 @@ deserialize arr@(Array {..}) = unsafeInlineIO $ do

539 Serialize.deserializeAt arrStart arrContents (arrStart + lenArr)

551 Serialize.deserializeAt arrStart arrContents (arrStart + lenArr)

540 assertM(off == arrStart + lenArr)

552 assertM(off == arrStart + lenArr)

541 pure val

553 pure val

554+

555+-------------------------------------------------------------------------------

556+-- Streams of Arrays

557+-------------------------------------------------------------------------------

558+

559+-- TODO: efficiently compare two streams of arrays. Two streams can have chunks

560+-- of different sizes, we can handle that in the stream comparison abstraction.

561+-- This could be useful e.g. to fast compare whether two files differ.

562+

563+-- | Insert the given element between arrays and flatten.

564+--

565+-- >>> interpose x = Stream.interpose x Array.reader

566+--

567+{-# INLINE interpose #-}

568+interpose :: (Monad m, Unbox a) => a -> Stream m (Array a) -> Stream m a

569+interpose x = D.interpose x reader

570+

571+data FlattenState s =

572+ OuterLoop s

573+ | InnerLoop s !MutByteArray !Int !Int

574+

575+-- | Insert the given element after each array and flatten. This is similar to

576+-- unlines.

577+--

578+-- >>> interposeSuffix x = Stream.interposeSuffix x Array.reader

579+--

580+{-# INLINE_NORMAL interposeSuffix #-}

581+interposeSuffix :: forall m a. (MonadIO m, Unbox a)

582+ => a -> Stream m (Array a) -> Stream m a

583+-- This does not require MonadIO constraint.

584+-- interposeSuffix x = D.interposeSuffix x reader

585+interposeSuffix sep (D.Stream step state) = D.Stream step' (OuterLoop state)

586+

587+ where

588+

589+ {-# INLINE_LATE step' #-}

590+ step' gst (OuterLoop st) = do

591+ r <- step (adaptState gst) st

592+ return $ case r of

593+ D.Yield Array{..} s ->

594+ D.Skip (InnerLoop s arrContents arrStart arrEnd)

595+ D.Skip s -> D.Skip (OuterLoop s)

596+ D.Stop -> D.Stop

597+

598+ step' _ (InnerLoop st _ p end) | p == end =

599+ return $ D.Yield sep $ OuterLoop st

600+

601+ step' _ (InnerLoop st contents p end) = do

602+ x <- liftIO $ peekAt p contents

603+ return $ D.Yield x (InnerLoop st contents (INDEX_NEXT(p,a)) end)

604+

605+-- | Insert the given array after each array and flatten.

606+--

607+-- >>> intercalateSuffix = Stream.intercalateSuffix Array.reader

608+--

609+{-# INLINE intercalateSuffix #-}

610+intercalateSuffix :: (Monad m, Unbox a)

611+ => Array a -> Stream m (Array a) -> Stream m a

612+intercalateSuffix = D.intercalateSuffix reader

613+

614+-- | Split a stream of arrays on a given separator byte, dropping the separator

615+-- and coalescing all the arrays between two separators into a single array.

616+--

617+{-# INLINE compactOnByte #-}

618+compactOnByte

619+ :: (MonadIO m)

620+ => Word8

621+ -> Stream m (Array Word8)

622+ -> Stream m (Array Word8)

623+compactOnByte byte =

624+ fmap unsafeFreeze . MA.compactOnByte byte . fmap unsafeThaw

625+

626+-- | Like 'compactOnByte' considers the separator in suffix position instead of

627+-- infix position.

628+{-# INLINE compactOnByteSuffix #-}

629+compactOnByteSuffix

630+ :: (MonadIO m)

631+ => Word8

632+ -> Stream m (Array Word8)

633+ -> Stream m (Array Word8)

634+compactOnByteSuffix byte =

635+ fmap unsafeFreeze . MA.compactOnByteSuffix byte . fmap unsafeThaw

core/src/Streamly/Internal/Data/Array/Stream.hs+11−176

@@ -19,13 +19,13 @@ module Streamly.Internal.Data.Array.Stream

19 , Array.bufferChunks

21 -- * Flattening to elements

22− , concat

22+ , Array.concat

23 , Array.flattenArrays

24− , concatRev

24+ , Array.concatRev

25 , Array.flattenArraysRev

26− , interpose

26+ , Array.interpose

27− , interposeSuffix

27+ , Array.interposeSuffix

28− , intercalateSuffix

28+ , Array.intercalateSuffix

29 , unlines

31 -- * Elimination

@@ -90,7 +90,6 @@ import Prelude hiding (null, last, (!!), read, concat, unlines)

91import Streamly.Data.Fold (Fold)

92import Streamly.Internal.Data.Array.Type (Array(..))

93−import Streamly.Internal.Data.MutByteArray.Type (MutByteArray)

94import Streamly.Internal.Data.Fold.Chunked (ChunkFold(..))

93import Streamly.Internal.Data.Fold.Chunked (ChunkFold(..))

95import Streamly.Internal.Data.Parser (ParseError(..))

94import Streamly.Internal.Data.Parser (ParseError(..))

96import Streamly.Internal.Data.Stream (Stream)

95import Streamly.Internal.Data.Stream (Stream)

@@ -98,11 +97,8 @@ import Streamly.Internal.Data.StreamK (StreamK, fromStream, toStream)

98import Streamly.Internal.Data.SVar.Type (adaptState, defState)

97import Streamly.Internal.Data.SVar.Type (adaptState, defState)

99import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))

98import Streamly.Internal.Data.Tuple.Strict (Tuple'(..))

100

101−import qualified Streamly.Data.Fold as FL

102import qualified Streamly.Internal.Data.Array as A

100import qualified Streamly.Internal.Data.Array as A

103import qualified Streamly.Internal.Data.Array as Array

101import qualified Streamly.Internal.Data.Array as Array

104−import qualified Streamly.Internal.Data.MutArray.Type as MA

105−import qualified Streamly.Internal.Data.MutArray.Stream as AS

106import qualified Streamly.Internal.Data.Fold.Type as FL (Fold(..), Step(..))

102import qualified Streamly.Internal.Data.Fold.Type as FL (Fold(..), Step(..))

107import qualified Streamly.Internal.Data.Parser as PR

103import qualified Streamly.Internal.Data.Parser as PR

108import qualified Streamly.Internal.Data.Parser as PRD

104import qualified Streamly.Internal.Data.Parser as PRD

@@ -114,110 +110,19 @@ import qualified Streamly.Internal.Data.StreamK as K

114-- XXX Since these are immutable arrays MonadIO constraint can be removed from

110-- XXX Since these are immutable arrays MonadIO constraint can be removed from

115-- most places.

111-- most places.

116

112

117−-------------------------------------------------------------------------------

118−-- Append

119−-------------------------------------------------------------------------------

120−

121−-- XXX efficiently compare two streams of arrays. Two streams can have chunks

122−-- of different sizes, we can handle that in the stream comparison abstraction.

123−-- This could be useful e.g. to fast compare whether two files differ.

124−

125−-- | Convert a stream of arrays into a stream of their elements.

126−--

127−-- Same as the following:

128−--

129−-- > concat = Stream.unfoldMany Array.read

130−--

131−-- @since 0.7.0

132−{-# INLINE concat #-}

133−concat :: (Monad m, Unbox a) => Stream m (Array a) -> Stream m a

134−-- concat m = fromStreamD $ A.flattenArrays (toStreamD m)

135−-- concat m = fromStreamD $ D.concatMap A.toStreamD (toStreamD m)

136−concat = D.unfoldMany A.reader

137−

138−-- | Convert a stream of arrays into a stream of their elements reversing the

139−-- contents of each array before flattening.

140−--

141−-- > concatRev = Stream.unfoldMany Array.readerRev

142−--

143−-- @since 0.7.0

144−{-# INLINE concatRev #-}

145−concatRev :: (Monad m, Unbox a) => Stream m (Array a) -> Stream m a

146−-- concatRev m = fromStreamD $ A.flattenArraysRev (toStreamD m)

147−concatRev = D.unfoldMany A.readerRev

148−

149-------------------------------------------------------------------------------

113-------------------------------------------------------------------------------

150-- Intersperse and append

114-- Intersperse and append

151-------------------------------------------------------------------------------

115-------------------------------------------------------------------------------

152

116

153−-- | Flatten a stream of arrays after inserting the given element between

154−-- arrays.

155−--

156−-- /Pre-release/

157−{-# INLINE interpose #-}

158−interpose :: (Monad m, Unbox a) => a -> Stream m (Array a) -> Stream m a

159−interpose x = D.interpose x A.reader

160−

161−{-# INLINE intercalateSuffix #-}

162−intercalateSuffix :: (Monad m, Unbox a)

163− => Array a -> Stream m (Array a) -> Stream m a

164−intercalateSuffix = D.intercalateSuffix A.reader

165−

166−-- | Flatten a stream of arrays appending the given element after each

167−-- array.

168−--

169−-- @since 0.7.0

170−{-# INLINE interposeSuffix #-}

171−interposeSuffix :: (Monad m, Unbox a)

172− => a -> Stream m (Array a) -> Stream m a

173−-- interposeSuffix x = fromStreamD . A.unlines x . toStreamD

174−interposeSuffix x = D.interposeSuffix x A.reader

175−

176−data FlattenState s =

177− OuterLoop s

178− | InnerLoop s !MutByteArray !Int !Int

179−

180−-- XXX This is a special case of interposeSuffix, can be removed.

181−-- XXX Remove monadIO constraint

182{-# INLINE_NORMAL unlines #-}

117{-# INLINE_NORMAL unlines #-}

183unlines :: forall m a. (MonadIO m, Unbox a)

118unlines :: forall m a. (MonadIO m, Unbox a)

184 => a -> D.Stream m (Array a) -> D.Stream m a

119 => a -> D.Stream m (Array a) -> D.Stream m a

185−unlines sep (D.Stream step state) = D.Stream step' (OuterLoop state)

120+unlines = Array.interposeSuffix

186− where

187− {-# INLINE_LATE step' #-}

188− step' gst (OuterLoop st) = do

189− r <- step (adaptState gst) st

190− return $ case r of

191− D.Yield Array{..} s ->

192− D.Skip (InnerLoop s arrContents arrStart arrEnd)

193− D.Skip s -> D.Skip (OuterLoop s)

194− D.Stop -> D.Stop

195−

196− step' _ (InnerLoop st _ p end) | p == end =

197− return $ D.Yield sep $ OuterLoop st

198−

199− step' _ (InnerLoop st contents p end) = do

200− x <- liftIO $ peekByteIndex p contents

201− return $ D.Yield x (InnerLoop st contents (INDEX_NEXT(p,a)) end)

202

121

203-------------------------------------------------------------------------------

122-------------------------------------------------------------------------------

204-- Compact

123-- Compact

205-------------------------------------------------------------------------------

124-------------------------------------------------------------------------------

206

125

207−-- XXX These would not be needed once we implement compactLEFold, see

208−-- module Streamly.Internal.Data.Stream.MutChunked

209−--

210−-- XXX Note that this thaws immutable arrays for appending, that may be

211−-- problematic if multiple users do the same thing, however, immutable arrays

212−-- would usually have no capacity to append, therefore, a copy will be forced

213−-- anyway. Confirm this. We can forcefully trim the array capacity before thaw

214−-- to ensure this.

215−{-# INLINE_NORMAL packArraysChunksOf #-}

216−packArraysChunksOf :: (MonadIO m, Unbox a)

217− => Int -> D.Stream m (Array a) -> D.Stream m (Array a)

218−packArraysChunksOf n str =

219− D.map A.unsafeFreeze $ AS.packArraysChunksOf n $ D.map A.unsafeThaw str

220−

221-- XXX instead of writing two different versions of this operation, we should

126-- XXX instead of writing two different versions of this operation, we should

222-- write it as a pipe.

127-- write it as a pipe.

223--

128--

@@ -225,8 +130,7 @@ packArraysChunksOf n str =

225{-# INLINE_NORMAL lpackArraysChunksOf #-}

130{-# INLINE_NORMAL lpackArraysChunksOf #-}

226lpackArraysChunksOf :: (MonadIO m, Unbox a)

131lpackArraysChunksOf :: (MonadIO m, Unbox a)

227 => Int -> Fold m (Array a) () -> Fold m (Array a) ()

132 => Int -> Fold m (Array a) () -> Fold m (Array a) ()

228−lpackArraysChunksOf n fld =

133+lpackArraysChunksOf = Array.lCompactGE

229− FL.lmap A.unsafeThaw $ AS.lpackArraysChunksOf n (FL.lmap A.unsafeFreeze fld)

230

134

231-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a

135-- | Coalesce adjacent arrays in incoming stream to form bigger arrays of a

232-- maximum specified size in bytes.

136-- maximum specified size in bytes.

@@ -235,7 +139,7 @@ lpackArraysChunksOf n fld =

235{-# INLINE compact #-}

139{-# INLINE compact #-}

236compact :: (MonadIO m, Unbox a)

140compact :: (MonadIO m, Unbox a)

237 => Int -> Stream m (Array a) -> Stream m (Array a)

141 => Int -> Stream m (Array a) -> Stream m (Array a)

238−compact = packArraysChunksOf

142+compact = Array.compactLE

239

143

240-- | Given a stream of arrays, splice them all together to generate a single

144-- | Given a stream of arrays, splice them all together to generate a single

241-- array. The stream must be /finite/.

145-- array. The stream must be /finite/.

@@ -243,71 +147,12 @@ compact = packArraysChunksOf

243-- @since 0.7.0

147-- @since 0.7.0

244{-# INLINE toArray #-}

148{-# INLINE toArray #-}

245toArray :: (MonadIO m, Unbox a) => Stream m (Array a) -> m (Array a)

149toArray :: (MonadIO m, Unbox a) => Stream m (Array a) -> m (Array a)

246−toArray s =

150+toArray = Array.fromChunks

247− fmap A.unsafeFreeze $ MA.fromChunksRealloced (fmap A.unsafeThaw s)

248

151

249-------------------------------------------------------------------------------

152-------------------------------------------------------------------------------

250-- Split

153-- Split

251-------------------------------------------------------------------------------

154-------------------------------------------------------------------------------

252

155

253−data SplitState s arr

254− = Initial s

255− | Buffering s arr

256− | Splitting s arr

257− | Yielding arr (SplitState s arr)

258− | Finishing

259−

260−-- | Split a stream of arrays on a given separator byte, dropping the separator

261−-- and coalescing all the arrays between two separators into a single array.

262−--

263−-- @since 0.7.0

264−{-# INLINE_NORMAL _splitOn #-}

265−_splitOn

266− :: MonadIO m

267− => Word8

268− -> D.Stream m (Array Word8)

269− -> D.Stream m (Array Word8)

270−_splitOn byte (D.Stream step state) = D.Stream step' (Initial state)

271−

272− where

273−

274− {-# INLINE_LATE step' #-}

275− step' gst (Initial st) = do

276− r <- step gst st

277− case r of

278− D.Yield arr s -> do

279− (arr1, marr2) <- A.breakOn byte arr

280− return $ case marr2 of

281− Nothing -> D.Skip (Buffering s arr1)

282− Just arr2 -> D.Skip (Yielding arr1 (Splitting s arr2))

283− D.Skip s -> return $ D.Skip (Initial s)

284− D.Stop -> return D.Stop

285−

286− step' gst (Buffering st buf) = do

287− r <- step gst st

288− case r of

289− D.Yield arr s -> do

290− (arr1, marr2) <- A.breakOn byte arr

291− -- XXX Should use MutArray.splice

292− buf' <- A.splice buf arr1

293− return $ case marr2 of

294− Nothing -> D.Skip (Buffering s buf')

295− Just x -> D.Skip (Yielding buf' (Splitting s x))

296− D.Skip s -> return $ D.Skip (Buffering s buf)

297− D.Stop -> return $

298− if A.byteLength buf == 0

299− then D.Stop

300− else D.Skip (Yielding buf Finishing)

301−

302− step' _ (Splitting st buf) = do

303− (arr1, marr2) <- A.breakOn byte buf

304− return $ case marr2 of

305− Nothing -> D.Skip $ Buffering st arr1

306− Just arr2 -> D.Skip $ Yielding arr1 (Splitting st arr2)

307−

308− step' _ (Yielding arr next) = return $ D.Yield arr next

309− step' _ Finishing = return D.Stop

310−

311-- XXX Remove MonadIO constraint.

156-- XXX Remove MonadIO constraint.

312-- | Split a stream of arrays on a given separator byte, dropping the separator

157-- | Split a stream of arrays on a given separator byte, dropping the separator

313-- and coalescing all the arrays between two separators into a single array.

158-- and coalescing all the arrays between two separators into a single array.

@@ -319,11 +164,7 @@ splitOn

319 => Word8

164 => Word8

320 -> Stream m (Array Word8)

165 -> Stream m (Array Word8)

321 -> Stream m (Array Word8)

166 -> Stream m (Array Word8)

322−splitOn byte =

167+splitOn = Array.compactOnByte

323− fmap A.unsafeFreeze

324− -- XXX use spliceExp and rightSize?

325− . D.splitInnerBy (MA.breakOn byte) MA.splice

326− . fmap A.unsafeThaw

327

168

328{-# INLINE splitOnSuffix #-}

169{-# INLINE splitOnSuffix #-}

329splitOnSuffix

170splitOnSuffix

@@ -331,13 +172,7 @@ splitOnSuffix

331 => Word8

172 => Word8

332 -> Stream m (Array Word8)

173 -> Stream m (Array Word8)

333 -> Stream m (Array Word8)

174 -> Stream m (Array Word8)

334−-- splitOn byte s = fromStreamD $ A.splitOn byte $ toStreamD s

335−splitOnSuffix byte =

175+splitOnSuffix = Array.compactOnByteSuffix

336− fmap A.unsafeFreeze

337− -- XXX use spliceExp and rightSize?

338− . D.splitInnerBySuffix

339− (\arr -> MA.byteLength arr == 0) (MA.breakOn byte) MA.splice

340− . fmap A.unsafeThaw

341

176

342-------------------------------------------------------------------------------

177-------------------------------------------------------------------------------

343-- Elimination - Running folds

178-- Elimination - Running folds

core/src/Streamly/Internal/Data/Array/Type.hs+72−1

@@ -69,6 +69,7 @@ module Streamly.Internal.Data.Array.Type

69 , fromByteStr#

70 , fromByteStr

71 , fromPtrN

72+ , fromChunks

72 , fromChunksK

73 , fromChunksK

74 -- ** Reading

75 -- ** Reading

@@ -122,6 +123,10 @@ module Streamly.Internal.Data.Array.Type

122

123

123 -- *** Compact

124 -- *** Compact

124 -- | Append the arrays in a stream to form a stream of larger arrays.

125 -- | Append the arrays in a stream to form a stream of larger arrays.

126+ , compactLE

127+ , fCompactGE

128+ , lCompactGE

129+ , compactGE

125

130

126 -- ** Deprecated

131 -- ** Deprecated

127 , unsafeIndex

132 , unsafeIndex

@@ -166,6 +171,7 @@ import Text.Read (readPrec)

166import Prelude hiding (Foldable(..), concat, read, unlines, splitAt)

171import Prelude hiding (Foldable(..), concat, read, unlines, splitAt)

167

172

168import qualified GHC.Exts as Exts

173import qualified GHC.Exts as Exts

174+import qualified Streamly.Internal.Data.Fold.Type as Fold

169import qualified Streamly.Internal.Data.MutArray.Type as MA

175import qualified Streamly.Internal.Data.MutArray.Type as MA

170import qualified Streamly.Internal.Data.Stream.Type as D

176import qualified Streamly.Internal.Data.Stream.Type as D

171import qualified Streamly.Internal.Data.StreamK.Type as K

177import qualified Streamly.Internal.Data.StreamK.Type as K

@@ -480,6 +486,56 @@ flattenArraysRev :: forall m a. (MonadIO m, Unbox a)

480 => D.Stream m (Array a) -> D.Stream m a

486 => D.Stream m (Array a) -> D.Stream m a

481flattenArraysRev = concatRev

487flattenArraysRev = concatRev

482

488

489+-------------------------------------------------------------------------------

490+-- Compact

491+-------------------------------------------------------------------------------

492+

493+-- XXX Note that this thaws immutable arrays for appending, that may be

494+-- problematic if multiple users do the same thing, however, thawed immutable

495+-- arrays would have no capacity to append, therefore, a copy will be forced

496+-- anyway.

497+

498+-- | Scan @compactLE n@ coalesces adjacent arrays in the input stream

499+-- only if the combined size would be less than or equal to n.

500+{-# INLINE_NORMAL compactLE #-}

501+compactLE :: (MonadIO m, Unbox a)

502+ => Int -> Stream m (Array a) -> Stream m (Array a)

503+compactLE n stream =

504+ D.map unsafeFreeze $ MA.rCompactLE n $ D.map unsafeThaw stream

505+

506+-- | Fold @fCompactGE n@ coalesces adjacent arrays in the input stream

507+-- until the size becomes greater than or equal to n.

508+--

509+{-# INLINE_NORMAL fCompactGE #-}

510+fCompactGE :: (MonadIO m, Unbox a) => Int -> Fold m (Array a) (Array a)

511+fCompactGE n = fmap unsafeFreeze $ Fold.lmap unsafeThaw $ MA.fCompactGE n

512+

513+-- | @compactGE n stream@ coalesces adjacent arrays in the @stream@ until

514+-- the size becomes greater than or equal to @n@.

515+--

516+-- >>> compactGE n = Stream.foldMany (Array.fCompactGE n)

517+--

518+{-# INLINE compactGE #-}

519+compactGE ::

520+ (MonadIO m, Unbox a)

521+ => Int -> Stream m (Array a) -> Stream m (Array a)

522+compactGE n stream =

523+ D.map unsafeFreeze $ MA.compactGE n $ D.map unsafeThaw stream

524+

525+-- | Like 'compactGE' but for transforming folds instead of stream.

526+--

527+-- >>> lCompactGE n = Fold.many (Array.fCompactGE n)

528+--

529+{-# INLINE_NORMAL lCompactGE #-}

530+lCompactGE :: (MonadIO m, Unbox a)

531+ => Int -> Fold m (Array a) () -> Fold m (Array a) ()

532+lCompactGE n fld =

533+ Fold.lmap unsafeThaw $ MA.lCompactGE n (Fold.lmap unsafeFreeze fld)

534+

535+-------------------------------------------------------------------------------

536+-- Splitting

537+-------------------------------------------------------------------------------

538+

483-- Drops the separator byte

539-- Drops the separator byte

484{-# INLINE breakOn #-}

540{-# INLINE breakOn #-}

485breakOn :: MonadIO m

541breakOn :: MonadIO m

@@ -802,14 +858,29 @@ fromByteStr (Ptr addr#) = fromByteStr# addr#

802-- that is double the size of the array stream.

858-- that is double the size of the array stream.

803--

859--

804{-# INLINE fromChunksK #-}

860{-# INLINE fromChunksK #-}

805−fromChunksK :: (Unbox a, MonadIO m) => StreamK m (Array a) -> m (Array a)

861+fromChunksK :: (MonadIO m, Unbox a) => StreamK m (Array a) -> m (Array a)

806fromChunksK stream =

862fromChunksK stream =

863+ -- We buffer the entire stream and then allocate the target array of the

864+ -- same size, thus requiring double the memory.

807 fmap unsafeFreeze $ MA.fromChunksK $ fmap unsafeThaw stream

865 fmap unsafeFreeze $ MA.fromChunksK $ fmap unsafeThaw stream

808

866

809{-# DEPRECATED fromArrayStreamK "Please use fromChunksK instead." #-}

867{-# DEPRECATED fromArrayStreamK "Please use fromChunksK instead." #-}

810fromArrayStreamK :: (Unbox a, MonadIO m) => StreamK m (Array a) -> m (Array a)

868fromArrayStreamK :: (Unbox a, MonadIO m) => StreamK m (Array a) -> m (Array a)

811fromArrayStreamK = fromChunksK

869fromArrayStreamK = fromChunksK

812

870

871+-- | Given a stream of arrays, splice them all together to generate a single

872+-- array. The stream must be /finite/.

873+--

874+{-# INLINE fromChunks #-}

875+fromChunks :: (MonadIO m, Unbox a) => Stream m (Array a) -> m (Array a)

876+fromChunks s =

877+ -- XXX Check which implementation is better

878+ -- This may also require double the memory as we double the space every

879+ -- time, when copying the last array we may have reallocated almost double

880+ -- the space required before we right size it.

881+ fmap unsafeFreeze $ MA.fromChunksRealloced (fmap unsafeThaw s)

882+ -- fromChunkStreamK $ D.toStreamK s

883+

813-------------------------------------------------------------------------------

884-------------------------------------------------------------------------------

814-- Instances

885-- Instances

815-------------------------------------------------------------------------------

886-------------------------------------------------------------------------------

core/src/Streamly/Internal/Data/MutArray.hs+95−1

@@ -14,6 +14,8 @@ module Streamly.Internal.Data.MutArray

14 , sliceIndexerFromLen

15 , slicerFromLen

16 , compactLE

17+ , compactOnByte

18+ , compactOnByteSuffix

17 -- * Unboxed IORef

19 -- * Unboxed IORef

18 , module Streamly.Internal.Data.IORef.Unboxed

20 , module Streamly.Internal.Data.IORef.Unboxed

@@ -26,11 +28,13 @@ where

26#include "inline.hs"

28#include "inline.hs"

28import Control.Monad.IO.Class (MonadIO(..))

30import Control.Monad.IO.Class (MonadIO(..))

31+import Data.Word (Word8)

29import Streamly.Internal.Data.Stream.Type (Stream)

32import Streamly.Internal.Data.Stream.Type (Stream)

30import Streamly.Internal.Data.Unbox (Unbox)

33import Streamly.Internal.Data.Unbox (Unbox)

31import Streamly.Internal.Data.Unfold.Type (Unfold(..))

34import Streamly.Internal.Data.Unfold.Type (Unfold(..))

36+import qualified Streamly.Internal.Data.Stream.Type as Stream

33−-- import qualified Streamly.Internal.Data.Stream.Nesting as Stream

37+import qualified Streamly.Internal.Data.Stream.Nesting as Stream

34-- import qualified Streamly.Internal.Data.Stream.Transform as Stream

38-- import qualified Streamly.Internal.Data.Stream.Transform as Stream

35import qualified Streamly.Internal.Data.Unfold as Unfold

39import qualified Streamly.Internal.Data.Unfold as Unfold

@@ -88,5 +92,95 @@ getSlicesFromLen = slicerFromLen

88{-# INLINE compactLE #-}

92{-# INLINE compactLE #-}

89compactLE :: (MonadIO m, Unbox a) =>

93compactLE :: (MonadIO m, Unbox a) =>

90 Int -> Stream m (MutArray a) -> Stream m (MutArray a)

94 Int -> Stream m (MutArray a) -> Stream m (MutArray a)

95+-- XXX compactLE can be moved to MutArray/Type if we are not using the parser

96+-- to implement it.

91-- compactLE n = Stream.catRights . Stream.parseManyD (pCompactLE n)

97-- compactLE n = Stream.catRights . Stream.parseManyD (pCompactLE n)

92compactLE = rCompactLE

98compactLE = rCompactLE

99+

100+data SplitState s arr

101+ = Initial s

102+ | Buffering s arr

103+ | Splitting s arr

104+ | Yielding arr (SplitState s arr)

105+ | Finishing

106+

107+-- | Split a stream of arrays on a given separator byte, dropping the separator

108+-- and coalescing all the arrays between two separators into a single array.

109+--

110+{-# INLINE_NORMAL _compactOnByteCustom #-}

111+_compactOnByteCustom

112+ :: MonadIO m

113+ => Word8

114+ -> Stream m (MutArray Word8)

115+ -> Stream m (MutArray Word8)

116+_compactOnByteCustom byte (Stream.Stream step state) =

117+ Stream.Stream step' (Initial state)

118+

119+ where

120+

121+ {-# INLINE_LATE step' #-}

122+ step' gst (Initial st) = do

123+ r <- step gst st

124+ case r of

125+ Stream.Yield arr s -> do

126+ (arr1, marr2) <- breakOn byte arr

127+ return $ case marr2 of

128+ Nothing -> Stream.Skip (Buffering s arr1)

129+ Just arr2 -> Stream.Skip (Yielding arr1 (Splitting s arr2))

130+ Stream.Skip s -> return $ Stream.Skip (Initial s)

131+ Stream.Stop -> return Stream.Stop

132+

133+ step' gst (Buffering st buf) = do

134+ r <- step gst st

135+ case r of

136+ Stream.Yield arr s -> do

137+ (arr1, marr2) <- breakOn byte arr

138+ -- XXX Use spliceExp instead and then rightSize?

139+ buf1 <- splice buf arr1

140+ return $ case marr2 of

141+ Nothing -> Stream.Skip (Buffering s buf1)

142+ Just x -> Stream.Skip (Yielding buf1 (Splitting s x))

143+ Stream.Skip s -> return $ Stream.Skip (Buffering s buf)

144+ Stream.Stop -> return $

145+ if byteLength buf == 0

146+ then Stream.Stop

147+ else Stream.Skip (Yielding buf Finishing)

148+

149+ step' _ (Splitting st buf) = do

150+ (arr1, marr2) <- breakOn byte buf

151+ return $ case marr2 of

152+ Nothing -> Stream.Skip $ Buffering st arr1

153+ Just arr2 -> Stream.Skip $ Yielding arr1 (Splitting st arr2)

154+

155+ step' _ (Yielding arr next) = return $ Stream.Yield arr next

156+ step' _ Finishing = return Stream.Stop

157+

158+-- XXX implement predicate based version of this

159+-- XXX Naming of predicate based vs custom version

160+

161+-- | Split a stream of arrays on a given separator byte, dropping the separator

162+-- and coalescing all the arrays between two separators into a single array.

163+--

164+{-# INLINE compactOnByte #-}

165+compactOnByte

166+ :: (MonadIO m)

167+ => Word8

168+ -> Stream m (MutArray Word8)

169+ -> Stream m (MutArray Word8)

170+-- XXX compare perf of custom vs idiomatic version

171+-- compactOnByte = _compactOnByteCustom

172+-- XXX use spliceExp and rightSize?

173+compactOnByte byte = Stream.splitInnerBy (breakOn byte) splice

174+

175+-- | Like 'compactOnByte' considers the separator in suffix position instead of

176+-- infix position.

177+{-# INLINE compactOnByteSuffix #-}

178+compactOnByteSuffix

179+ :: (MonadIO m)

180+ => Word8

181+ -> Stream m (MutArray Word8)

182+ -> Stream m (MutArray Word8)

183+compactOnByteSuffix byte =

184+ -- XXX use spliceExp and rightSize?

185+ Stream.splitInnerBySuffix

186+ (\arr -> byteLength arr == 0) (breakOn byte) splice