7650587 · streamly · git.iko.soy

git.iko.soy/streamly

Implement fold in terms of fold_.

- Add test for fold_.

commit7650587baa7bdbfd8e319eaee62c6a50329b1223 authorPranay Sashank <pranaysashank@composewell.com> dateThu 28 Jan 2021, 13:01 (6 years ago) parent3cd4b10

17 files changed+81−46

compare with master

unifiedsplit

src/Streamly/Internal/Data/Array.hs+2−2

@@ -151,7 +151,7 @@ fromStreamDN limit str = do

151151

152152{-# INLINE fromStreamD #-}

153153fromStreamD :: MonadIO m => D.Stream m a -> m (Array a)

154−fromStreamD = D.foldOnce write

154+fromStreamD = D.fold write

155155

156156{-# INLINABLE fromListN #-}

157157fromListN :: Int -> [a] -> Array a

@@ -187,7 +187,7 @@ toStreamRev = D.fromStreamD . toStreamDRev

187187

188188{-# INLINE fold #-}

189189fold :: Monad m => Fold m a b -> Array a -> m b

190−fold f arr = D.foldOnce f (toStreamD arr)

190+fold f arr = D.fold f (toStreamD arr)

191191

192192{-# INLINE streamFold #-}

193193streamFold :: Monad m => (SerialT m a -> m b) -> Array a -> m b

src/Streamly/Internal/Data/Array/Foreign.hs+3−3

@@ -201,7 +201,7 @@ fromStreamN n m = do

201201-- /Pre-release/

202202{-# INLINE fromStream #-}

203203fromStream :: (MonadIO m, Storable a) => SerialT m a -> m (Array a)

204−fromStream = P.foldOnce A.write

204+fromStream = P.fold A.write

205205-- write m = A.fromStreamD $ D.toStreamD m

206206

207207-------------------------------------------------------------------------------

@@ -492,7 +492,7 @@ runPipe f arr = P.runPipe (toArrayMinChunk (length arr)) $ f (A.read arr)

492492streamTransform :: forall m a b. (MonadIO m, Storable a, Storable b)

493493 => (SerialT m a -> SerialT m b) -> Array a -> m (Array b)

494494streamTransform f arr =

495− P.foldOnce (A.toArrayMinChunk (alignment (undefined :: a)) (length arr))

495+ P.fold (A.toArrayMinChunk (alignment (undefined :: a)) (length arr))

496496 $ f (A.toStream arr)

497497

498498-------------------------------------------------------------------------------

@@ -566,7 +566,7 @@ asCString arr act = do

566566-- /Pre-release/

567567{-# INLINE fold #-}

568568fold :: forall m a b. (MonadIO m, Storable a) => Fold m a b -> Array a -> m b

569−fold f arr = P.foldOnce f (A.toStream arr :: Serial.SerialT m a)

569+fold f arr = P.fold f (A.toStream arr :: Serial.SerialT m a)

570570

571571-- | Fold an array using a stream fold operation.

572572--

src/Streamly/Internal/Data/Array/Prim/MutTypesInclude.hs+1−1

@@ -216,7 +216,7 @@ fromStreamDN limit str = do

216216

217217{-# INLINE fromStreamD #-}

218218fromStreamD :: (MonadIO m, Prim a) => D.Stream m a -> m (Array a)

219−fromStreamD str = D.foldOnce write str

219+fromStreamD str = D.fold write str

220220

221221{-# INLINABLE fromListNM #-}

222222fromListNM :: (MonadIO m, Prim a) => Int -> [a] -> m (Array a)

src/Streamly/Internal/Data/Array/PrimInclude.hs+2−2

@@ -45,7 +45,7 @@ fromStreamN n m = do

4545-- /Pre-release/

4646{-# INLINE fromStream #-}

4747fromStream :: (MonadIO m, Prim a) => SerialT m a -> m (Array a)

48−fromStream = P.foldOnce A.write

48+fromStream = P.fold A.write

4949-- write m = A.fromStreamD $ D.toStreamD m

5050

5151-------------------------------------------------------------------------------

@@ -128,7 +128,7 @@ null arr = length arr == 0

128128-- /Pre-release/

129129{-# INLINE fold #-}

130130fold :: forall m a b. (MonadIO m, Prim a) => Fold m a b -> Array a -> m b

131−fold f arr = P.foldOnce f (toStream arr :: Serial.SerialT m a)

131+fold f arr = P.fold f (toStream arr :: Serial.SerialT m a)

132132

133133-- | Fold an array using a stream fold operation.

134134--

src/Streamly/Internal/Data/SmallArray.hs+1−1

@@ -169,7 +169,7 @@ toStreamRev = D.fromStreamD . toStreamDRev

169169

170170{-# INLINE fold #-}

171171fold :: Monad m => Fold m a b -> SmallArray a -> m b

172−fold f arr = D.foldOnce f (toStreamD arr)

172+fold f arr = D.fold f (toStreamD arr)

173173

174174{-# INLINE streamFold #-}

175175streamFold :: Monad m => (SerialT m a -> m b) -> SmallArray a -> m b

src/Streamly/Internal/Data/Stream/IsStream/Common.hs+10−1

@@ -23,6 +23,7 @@ module Streamly.Internal.Data.Stream.IsStream.Common

2323

2424 -- * Elimination

2525 , fold

26+ , fold_

2627

2728 -- * Transformation

2829 , scanlMAfter'

@@ -248,7 +249,15 @@ relTimesWith = fmap snd . timesWith

248249-- @since 0.7.0

249250{-# INLINE fold #-}

250251fold :: Monad m => Fold m a b -> SerialT m a -> m b

251−fold = P.foldOnce

252+fold fl strm = do

253+ (b, _) <- fold_ fl strm

254+ return $! b

255+

256+{-# INLINE fold_ #-}

257+fold_ :: Monad m => Fold m a b -> SerialT m a -> m (b, SerialT m a)

258+fold_ fl strm = do

259+ (b, str) <- D.fold_ fl $ D.toStreamD strm

260+ return $! (b, D.fromStreamD str)

252261

253262------------------------------------------------------------------------------

254263-- Transformation

src/Streamly/Internal/Data/Stream/IsStream/Eliminate.hs+3−1

@@ -20,6 +20,7 @@ module Streamly.Internal.Data.Stream.IsStream.Eliminate

2020 -- * Running a 'Fold'

2121 -- See "Streamly.Internal.Data.Fold".

2222 fold

23+ , fold_

2324

2425 -- * Running a 'Parser'

2526 -- "Streamly.Internal.Data.Parser".

@@ -150,7 +151,8 @@ import Foreign.Storable (Storable)

150151import Streamly.Internal.Data.Parser (Parser (..))

151152import Streamly.Internal.Data.SVar (MonadAsync, defState)

152153import Streamly.Internal.Data.Stream.IsStream.Common

153− (fold, drop, findIndices, reverse, splitOnSeq, take, takeWhile)

154+ ( fold, fold_, drop, findIndices, reverse, splitOnSeq, take

155+ , takeWhile)

154156import Streamly.Internal.Data.Stream.Prelude (toStreamS)

155157import Streamly.Internal.Data.Stream.StreamD (fromStreamD, toStreamD)

156158import Streamly.Internal.Data.Stream.StreamK (IsStream)

src/Streamly/Internal/Data/Stream/Prelude.hs+4−4

@@ -33,7 +33,7 @@ module Streamly.Internal.Data.Stream.Prelude

3333 , foldlx'

3434 , foldlMx'

3535 , foldl'

36− , foldOnce

36+ , fold

3737

3838 -- Lazy left folds are useful only for reversing the stream

3939 , foldlS

@@ -188,9 +188,9 @@ foldlT :: (Monad m, IsStream t, Monad (s m), MonadTrans s)

188188 => (s m b -> a -> s m b) -> s m b -> t m a -> s m b

189189foldlT f z s = S.foldlT f z (toStreamS s)

190190

191−{-# INLINE foldOnce #-}

192−foldOnce :: (Monad m, IsStream t) => Fold m a b -> t m a -> m b

193−foldOnce fld m = S.foldOnce fld $ toStreamS m

191+{-# INLINE fold #-}

192+fold :: (Monad m, IsStream t) => Fold m a b -> t m a -> m b

193+fold fld m = S.fold fld $ toStreamS m

194194

195195------------------------------------------------------------------------------

196196-- Scans

src/Streamly/Internal/Data/Stream/SVar.hs+3−3

@@ -374,10 +374,10 @@ toSVarParallel st sv xs =

374374 where

375375

376376 {-# NOINLINE work #-}

377− work info = D.foldOnce (write sv info) xs

377+ work info = D.fold (write sv info) xs

378378

379379 {-# NOINLINE workLim #-}

380− workLim info = D.foldOnce (writeLimited sv info) xs

380+ workLim info = D.fold (writeLimited sv info) xs

381381

382382 {-# NOINLINE forkWithDiag #-}

383383 forkWithDiag = do

@@ -567,7 +567,7 @@ newFoldSVarF stt f = do

567567 where

568568

569569 {-# NOINLINE work #-}

570− work sv = void $ D.foldOnce f $ fromProducerD sv

570+ work sv = void $ D.fold f $ fromProducerD sv

571571

572572-------------------------------------------------------------------------------

573573-- Process events received by the producer thread from the consumer side

src/Streamly/Internal/Data/Stream/StreamD/Eliminate.hs+1−1

@@ -13,7 +13,7 @@

1313module Streamly.Internal.Data.Stream.StreamD.Eliminate

1414 (

1515 -- * Running a 'Fold'

16− foldOnce -- XXX rename to "fold"

16+ fold

1717

1818 -- -- * Running a 'Parser'

1919 , parse

src/Streamly/Internal/Data/Stream/StreamD/Nesting.hs+4−4

@@ -1762,8 +1762,8 @@ splitOnSeq patArr (Fold fstep initial done) (Stream step state) =

17621762 rh1 <- liftIO $ RB.unsafeInsert rb rh x

17631763 if idx == maxIndex

17641764 then do

1765− let fold = RB.unsafeFoldRing (RB.ringBound rb)

1766− let !ringHash = fold addCksum 0 rb

1765+ let fld = RB.unsafeFoldRing (RB.ringBound rb)

1766+ let !ringHash = fld addCksum 0 rb

17671767 if ringHash == patHash

17681768 then skip $ SplitOnSeqKRCheck fs s rb rh1

17691769 else skip $ SplitOnSeqKRLoop fs s rb rh1 ringHash

@@ -2137,8 +2137,8 @@ splitOnSuffixSeq withSep patArr (Fold fstep initial done) (Stream step state) =

21372137 if idx /= maxIndex

21382138 then go SPEC (idx + 1) rh1 s fs1

21392139 else skip $

2140− let fold = RB.unsafeFoldRing (RB.ringBound rb)

2141− !ringHash = fold addCksum 0 rb

2140+ let fld = RB.unsafeFoldRing (RB.ringBound rb)

2141+ !ringHash = fld addCksum 0 rb

21422142 in if ringHash == patHash

21432143 then SplitOnSuffixSeqKRCheck fs1 s rb rh1

21442144 else SplitOnSuffixSeqKRLoop

src/Streamly/Internal/Data/Stream/StreamD/Transform.hs+1−1

@@ -346,7 +346,7 @@ pollCounts predicate transf fld (Stream step state) = Stream step' Nothing

346346 -- However, an Int on a 32-bit machine may overflow quickly.

347347 countVar <- liftIO $ Prim.newIORef (0 :: Int)

348348 tid <- forkManaged

349− $ void $ foldOnce fld

349+ $ void $ fold fld

350350 $ transf $ Prim.toStreamD countVar

351351 return $ Skip (Just (countVar, tid, state))

352352

src/Streamly/Internal/Data/Stream/StreamD/Type.hs+16−7

@@ -43,7 +43,8 @@ module Streamly.Internal.Data.Stream.StreamD.Type

4343 , fromStreamD

4444

4545 -- * Running a 'Fold'

46− , foldOnce

46+ , fold

47+ , fold_

4748

4849 -- * Right Folds

4950 , foldrT

@@ -268,13 +269,19 @@ toStreamD = fromStreamK . K.toStream

268269-- Running a 'Fold'

269270------------------------------------------------------------------------------

270271

271−{-# INLINE_NORMAL foldOnce #-}

272−foldOnce :: (Monad m) => Fold m a b -> Stream m a -> m b

273−foldOnce (Fold fstep begin done) (Stream step state) = do

272+{-# INLINE_NORMAL fold #-}

273+fold :: (Monad m) => Fold m a b -> Stream m a -> m b

274+fold fld strm = do

275+ (b, _) <- fold_ fld strm

276+ return b

277+

278+{-# INLINE_NORMAL fold_ #-}

279+fold_ :: Monad m => Fold m a b -> Stream m a -> m (b, Stream m a)

280+fold_ (Fold fstep begin done) (Stream step state) = do

274281 res <- begin

275282 case res of

276283 FL.Partial fs -> go SPEC fs state

277− FL.Done fb -> return fb

284+ FL.Done fb -> return $! (fb, Stream step state)

278285

279286 where

280287

@@ -285,10 +292,12 @@ foldOnce (Fold fstep begin done) (Stream step state) = do

285292 Yield x s -> do

286293 res <- fstep fs x

287294 case res of

288− FL.Done b -> return b

295+ FL.Done b -> return $! (b, Stream step s)

289296 FL.Partial fs1 -> go SPEC fs1 s

290297 Skip s -> go SPEC fs s

291− Stop -> done fs

298+ Stop -> do

299+ b <- done fs

300+ return $! (b, Stream (\ _ _ -> return Stop) ())

292301

293302------------------------------------------------------------------------------

294303-- Right Folds

src/Streamly/Internal/Data/Stream/StreamK.hs+4−4

@@ -88,7 +88,7 @@ module Streamly.Internal.Data.Stream.StreamK

8888 , foldlT

8989 , foldlx'

9090 , foldlMx'

91− , foldOnce

91+ , fold

9292

9393 -- ** Specialized Folds

9494 , drain

@@ -400,9 +400,9 @@ foldlMx' step begin done m = go begin m

400400 yieldk a r = acc >>= \b -> step b a >>= \x -> go (return x) r

401401 in foldStream defState yieldk single stop m1

402402

403−{-# INLINABLE foldOnce #-}

404−foldOnce :: (IsStream t, Monad m) => FL.Fold m a b -> t m a -> m b

405−foldOnce (FL.Fold step begin done) m = do

403+{-# INLINABLE fold #-}

404+fold :: (IsStream t, Monad m) => FL.Fold m a b -> t m a -> m b

405+fold (FL.Fold step begin done) m = do

406406 res <- begin

407407 case res of

408408 FL.Partial fs -> go fs m

src/Streamly/Internal/Unicode/Stream.hs+1−1

@@ -182,7 +182,7 @@ utf8d =

182182 unsafePerformIO

183183 -- Aligning to cacheline makes a barely noticeable difference

184184 -- XXX currently alignment is not implemented for unmanaged allocation

185− $ D.foldOnce (A.writeNAlignedUnmanaged 64 (length decodeTable))

185+ $ D.fold (A.writeNAlignedUnmanaged 64 (length decodeTable))

186186 (D.fromList decodeTable)

187187

188188-- | Return element at the specified index without checking the bounds.

test/Streamly/Test/Data/Fold.hs+23−8

@@ -1,7 +1,7 @@

11module Main (main) where

33import Data.Semigroup (Sum(..), getSum)

4−import Streamly.Test.Common (checkListEqual)

4+import Streamly.Test.Common (checkListEqual, listEquals)

55import Test.QuickCheck

66 ( Gen

77 , Property

@@ -18,7 +18,8 @@ import Test.QuickCheck.Monadic (monadicIO, assert, run)

1818

1919import qualified Prelude

2020import qualified Streamly.Internal.Data.Fold as F

21−import qualified Streamly.Internal.Data.Stream.IsStream as S

21+import qualified Streamly.Prelude as S

22+import qualified Streamly.Internal.Data.Stream.IsStream as Stream

2223import qualified Streamly.Data.Fold as FL

2324

2425import Prelude hiding

@@ -60,10 +61,9 @@ rollingHashFirstN =

6061head :: [Int] -> Expectation

6162head ls = S.fold FL.head (S.fromList ls) `shouldReturn` headl ls

6263

63− where

64−

65− headl [] = Nothing

66− headl (x:_) = Just x

64+headl :: [a] -> Maybe a

65+headl [] = Nothing

66+headl (x:_) = Just x

6767

6868length :: [Int] -> Expectation

6969length ls = S.fold FL.length (S.fromList ls) `shouldReturn` Prelude.length ls

@@ -424,15 +424,28 @@ many =

424424 monadicIO $ do

425425 let strm = S.fromList lst

426426 r1 <- S.fold (F.many (split i) F.toList) strm

427− r2 <- S.toList $ S.foldMany (split i) strm

427+ r2 <- S.toList $ Stream.foldMany (split i) strm

428428 assert $ r1 == r2

429429

430430 where

431431

432432 split i = F.take i F.toList

433433

434+headAndRest :: [Int] -> Property

435+headAndRest ls = monadicIO $ do

436+ (mbh, rest) <- run $ Stream.fold_ FL.head (S.fromList ls)

437+ rests <- run $ S.toList rest

438+ assert (mbh == headl ls)

439+ listEquals (==) rests (taill ls)

440+

441+ where

442+

443+ taill :: [a] -> [a]

444+ taill [] = []

445+ taill (_:xs) = xs

446+

434447main :: IO ()

435−main = hspec $

448+main = hspec $ do

436449 describe "Fold" $ do

437450 -- Folds

438451 -- Accumulators

@@ -510,3 +523,5 @@ main = hspec $

510523 prop "many" Main.many

511524 -- concatMap

512525 -- chunksOf

526+ describe "FoldAndReturn s" $ do

527+ prop "head from fold_" headAndRest

test/Streamly/Test/FileSystem/Handle.hs+2−2

@@ -120,7 +120,7 @@ testWrite hfold =

120120 writeFile fpathWrite ""

121121 h <- openFile fpathWrite ReadWriteMode

122122 hSeek h AbsoluteSeek 0

123− Stream.fold (hfold h) $ Stream.fromList list

123+ _ <- Stream.fold (hfold h) $ Stream.fromList list

124124 hFlush h

125125 hSeek h AbsoluteSeek 0

126126 ls <- Stream.toList $ Stream.unfold Handle.read h

@@ -144,7 +144,7 @@ testWriteWithChunk =

144144 hr <- openFile fpathRead ReadMode

145145 hw <- openFile fpathWrite ReadWriteMode

146146 hSeek hw AbsoluteSeek 0

147− Stream.fold (Handle.writeChunks hw)

147+ _ <- Stream.fold (Handle.writeChunks hw)

148148 $ Stream.unfold Handle.readChunksWithBufferOf (1024, hr)

149149 hFlush hw

150150 hSeek hw AbsoluteSeek 0