a3c32bc · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

Fix foldStreamSVar space leak

a3c32bcHarendra Kumar7 years ago

Changelog.md+5−5

compare with master

unifiedsplit

@@ -6,11 +6,11 @@

6 cases.

7* Fix a bug in `tap`, `chunksOf` that caused memory consumption to

8 increase in some cases.

9−* Fix a space leak in `async` streams that caused memory consumption

9+* Fix a space leak in concurrent streams (`async`, `wAsync`, and `ahead`) that

10− to increase with the number of elements in the stream, especially when

10+ caused memory consumption to increase with the number of elements in the

11− built with `-threaded` and used with `-N` RTS option. The issue occurs

11+ stream, especially when built with `-threaded` and used with `-N` RTS option.

12− only in cases when a worker thread happens to be used continuously for a

12+ The issue occurs only in cases when a worker thread happens to be used

13− long time.

13+ continuously for a long time.

15### Behavior change

src/Streamly/Internal/Data/Stream/Ahead.hs+55−29

@@ -36,6 +36,7 @@ import Control.Monad.IO.Class (MonadIO(..))

36import Control.Monad.Reader.Class (MonadReader(..))

37import Control.Monad.State.Class (MonadState(..))

38import Control.Monad.Trans.Class (MonadTrans(lift))

39+import Control.Monad.Trans.Control (MonadBaseControl (..))

39import Data.Heap (Heap, Entry(..))

40import Data.Heap (Heap, Entry(..))

40import Data.IORef (IORef, readIORef, atomicModifyIORef, writeIORef)

41import Data.IORef (IORef, readIORef, atomicModifyIORef, writeIORef)

41import Data.Maybe (fromJust)

42import Data.Maybe (fromJust)

@@ -49,8 +50,7 @@ import qualified Data.Heap as H

49import Streamly.Internal.Data.Stream.SVar (fromSVar)

50import Streamly.Internal.Data.Stream.SVar (fromSVar)

50import Streamly.Internal.Data.SVar

51import Streamly.Internal.Data.SVar

51import Streamly.Internal.Data.Stream.StreamK

52import Streamly.Internal.Data.Stream.StreamK

52− (IsStream(..), Stream, mkStream, foldStream, foldStreamShared,

53+ (IsStream(..), Stream, mkStream, foldStream, foldStreamShared)

53− foldStreamSVar)

54import qualified Streamly.Internal.Data.Stream.StreamK as K

55import qualified Streamly.Internal.Data.Stream.StreamD as D

@@ -212,7 +212,8 @@ abortExecution q sv winfo m = do

212--

213-- In both cases we give the drainer a chance to run more often.

214--

215−processHeap :: MonadIO m

215+processHeap

216+ :: (MonadIO m, MonadBaseControl IO m)

216 => IORef ([Stream m a], Int)

217 => IORef ([Stream m a], Int)

217 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

218 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

218 -> State Stream m a

219 -> State Stream m a

@@ -316,7 +317,8 @@ processHeap q heap st sv winfo entry sno stopping = loopHeap sno entry

316 runStreamWithYieldLimit continue seqNo r

317 runStreamWithYieldLimit continue seqNo r

317

318

318{-# NOINLINE drainHeap #-}

319{-# NOINLINE drainHeap #-}

319−drainHeap :: MonadIO m

320+drainHeap

321+ :: (MonadIO m, MonadBaseControl IO m)

320 => IORef ([Stream m a], Int)

322 => IORef ([Stream m a], Int)

321 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

323 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

322 -> State Stream m a

324 -> State Stream m a

@@ -331,8 +333,10 @@ drainHeap q heap st sv winfo = do

331 _ -> liftIO $ sendStop sv winfo

333 _ -> liftIO $ sendStop sv winfo

332

334

333data HeapStatus = HContinue | HStop

335data HeapStatus = HContinue | HStop

336+data WorkerStatus = Continue | Suspend

334

337

335−processWithoutToken :: MonadIO m

338+processWithoutToken

339+ :: (MonadIO m, MonadBaseControl IO m)

336 => IORef ([Stream m a], Int)

340 => IORef ([Stream m a], Int)

337 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

341 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

338 -> State Stream m a

342 -> State Stream m a

@@ -351,12 +355,18 @@ processWithoutToken q heap st sv winfo m seqNo = do

351 -- put it on heap. So we have to put a null entry on heap even when

355 -- put it on heap. So we have to put a null entry on heap even when

352 -- we stop.

356 -- we stop.

353 toHeap AheadEntryNull

357 toHeap AheadEntryNull

358+ mrun = runInIO $ svarMrun sv

354

359

355− foldStreamSVar sv st

360+ r <- liftIO $ mrun $

361+ foldStreamShared st

356− (\a r -> toHeap $ AheadEntryStream $ K.cons a r)

362+ (\a r -> toHeap $ AheadEntryStream $ K.cons a r)

357− (toHeap . AheadEntryPure)

363+ (toHeap . AheadEntryPure)

358− stop

364+ stop

359− m

365+ m

366+ res <- restoreM r

367+ case res of

368+ Continue -> workLoopAhead q heap st sv winfo

369+ Suspend -> drainHeap q heap st sv winfo

360

370

361 where

371 where

362

372

@@ -377,8 +387,6 @@ processWithoutToken q heap st sv winfo m seqNo = do

377 writeIORef (maxHeapSize $ svarStats sv) (H.size newHp)

387 writeIORef (maxHeapSize $ svarStats sv) (H.size newHp)

378

388

379 heapOk <- liftIO $ underMaxHeap sv newHp

389 heapOk <- liftIO $ underMaxHeap sv newHp

380− let drainAndStop = drainHeap q heap st sv winfo

381− mainLoop = workLoopAhead q heap st sv winfo

382 status <-

390 status <-

383 case yieldRateInfo sv of

391 case yieldRateInfo sv of

384 Nothing -> return HContinue

392 Nothing -> return HContinue

@@ -394,11 +402,14 @@ processWithoutToken q heap st sv winfo m seqNo = do

394 if heapOk

402 if heapOk

395 then

403 then

396 case status of

404 case status of

397− HContinue -> mainLoop

405+ HContinue -> return Continue

398− HStop -> drainAndStop

406+ HStop -> return Suspend

399− else drainAndStop

407+ else return Suspend

408+

409+data TokenWorkerStatus = TokenContinue Int | TokenSuspend

400

410

401−processWithToken :: MonadIO m

411+processWithToken

412+ :: (MonadIO m, MonadBaseControl IO m)

402 => IORef ([Stream m a], Int)

413 => IORef ([Stream m a], Int)

403 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

414 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

404 -> State Stream m a

415 -> State Stream m a

@@ -412,19 +423,26 @@ processWithToken q heap st sv winfo action sno = do

412 -- XXX deduplicate stop in all invocations

423 -- XXX deduplicate stop in all invocations

413 let stop = do

424 let stop = do

414 liftIO (incrementYieldLimit sv)

425 liftIO (incrementYieldLimit sv)

415− loopWithToken (sno + 1)

426+ return $ TokenContinue (sno + 1)

427+ mrun = runInIO $ svarMrun sv

416

428

429+ r <- liftIO $ mrun $

417− foldStreamSVar sv st (yieldOutput sno) (singleOutput sno) stop action

430+ foldStreamShared st (yieldOutput sno) (singleOutput sno) stop action

431+

432+ res <- restoreM r

433+ case res of

434+ TokenContinue seqNo -> loopWithToken seqNo

435+ TokenSuspend -> drainHeap q heap st sv winfo

418

436

419 where

437 where

420

438

421 singleOutput seqNo a = do

439 singleOutput seqNo a = do

422 continue <- liftIO $ sendYield sv winfo (ChildYield a)

440 continue <- liftIO $ sendYield sv winfo (ChildYield a)

423 if continue

441 if continue

424− then loopWithToken (seqNo + 1)

442+ then return $ TokenContinue (seqNo + 1)

425 else do

443 else do

426 liftIO $ updateHeapSeq heap (seqNo + 1)

444 liftIO $ updateHeapSeq heap (seqNo + 1)

427− drainHeap q heap st sv winfo

445+ return TokenSuspend

428

446

429 -- XXX use a wrapper function around stop so that we never miss

447 -- XXX use a wrapper function around stop so that we never miss

430 -- incrementing the yield in a stop continuation. Essentiatlly all

448 -- incrementing the yield in a stop continuation. Essentiatlly all

@@ -436,7 +454,7 @@ processWithToken q heap st sv winfo action sno = do

436 then do

454 then do

437 let stop = do

455 let stop = do

438 liftIO (incrementYieldLimit sv)

456 liftIO (incrementYieldLimit sv)

439− loopWithToken (seqNo + 1)

457+ return $ TokenContinue (seqNo + 1)

440 foldStreamShared st

458 foldStreamShared st

441 (yieldOutput seqNo)

459 (yieldOutput seqNo)

442 (singleOutput seqNo)

460 (singleOutput seqNo)

@@ -446,7 +464,7 @@ processWithToken q heap st sv winfo action sno = do

446 let ent = Entry seqNo (AheadEntryStream r)

464 let ent = Entry seqNo (AheadEntryStream r)

447 liftIO $ requeueOnHeapTop heap ent seqNo

465 liftIO $ requeueOnHeapTop heap ent seqNo

448 liftIO $ incrementYieldLimit sv

466 liftIO $ incrementYieldLimit sv

449− drainHeap q heap st sv winfo

467+ return TokenSuspend

450

468

451 loopWithToken nextSeqNo = do

469 loopWithToken nextSeqNo = do

452 work <- dequeueAhead q

470 work <- dequeueAhead q

@@ -467,12 +485,19 @@ processWithToken q heap st sv winfo action sno = do

467 then do

485 then do

468 let stop = do

486 let stop = do

469 liftIO (incrementYieldLimit sv)

487 liftIO (incrementYieldLimit sv)

470− loopWithToken (seqNo + 1)

488+ return $ TokenContinue (seqNo + 1)

489+ mrun = runInIO $ svarMrun sv

490+ r <- liftIO $ mrun $

471− foldStreamShared st

491+ foldStreamShared st

472− (yieldOutput seqNo)

492+ (yieldOutput seqNo)

473− (singleOutput seqNo)

493+ (singleOutput seqNo)

474− stop

494+ stop

475− m

495+ m

496+ res <- restoreM r

497+ case res of

498+ TokenContinue seqNo1 -> loopWithToken seqNo1

499+ TokenSuspend -> drainHeap q heap st sv winfo

500+

476 else

501 else

477 -- To avoid a race when another thread puts something

502 -- To avoid a race when another thread puts something

478 -- on the heap and goes away, the consumer will not get

503 -- on the heap and goes away, the consumer will not get

@@ -496,7 +521,8 @@ processWithToken q heap st sv winfo action sno = do

496

521

497-- XXX we can remove the sv parameter as it can be derived from st

522-- XXX we can remove the sv parameter as it can be derived from st

498

523

499−workLoopAhead :: MonadIO m

524+workLoopAhead

525+ :: (MonadIO m, MonadBaseControl IO m)

500 => IORef ([Stream m a], Int)

526 => IORef ([Stream m a], Int)

501 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

527 -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)), Maybe Int)

502 -> State Stream m a

528 -> State Stream m a

src/Streamly/Internal/Data/Stream/Async.hs+68−30

@@ -39,6 +39,7 @@ where

39import Control.Concurrent (myThreadId)

40import Control.Monad.Base (MonadBase(..), liftBaseDefault)

41import Control.Monad.Catch (MonadThrow, throwM)

42+import Control.Monad.Trans.Control (MonadBaseControl (..))

42import Control.Concurrent.MVar (newEmptyMVar)

43import Control.Concurrent.MVar (newEmptyMVar)

43-- import Control.Monad.Error.Class (MonadError(..))

44-- import Control.Monad.Error.Class (MonadError(..))

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

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

@@ -59,8 +60,7 @@ import Streamly.Internal.Data.Atomics (atomicModifyIORefCAS)

59import Streamly.Internal.Data.Stream.SVar (fromSVar)

60import Streamly.Internal.Data.Stream.SVar (fromSVar)

60import Streamly.Internal.Data.SVar

61import Streamly.Internal.Data.SVar

61import Streamly.Internal.Data.Stream.StreamK

62import Streamly.Internal.Data.Stream.StreamK

62− (IsStream(..), Stream, mkStream, foldStream, adapt, foldStreamShared,

63+ (IsStream(..), Stream, mkStream, foldStream, adapt, foldStreamShared)

63− foldStreamSVar)

64import qualified Streamly.Internal.Data.Stream.StreamK as K

65import qualified Streamly.Internal.Data.Stream.StreamD as D

@@ -70,9 +70,11 @@ import qualified Streamly.Internal.Data.Stream.StreamD as D

70-- Async

71-------------------------------------------------------------------------------

73+data WorkerStatus = Continue | Suspend

74+

73{-# INLINE workLoopLIFO #-}

75{-# INLINE workLoopLIFO #-}

74workLoopLIFO

76workLoopLIFO

75− :: MonadIO m

77+ :: (MonadIO m, MonadBaseControl IO m)

76 => IORef [Stream m a]

78 => IORef [Stream m a]

77 -> State Stream m a

79 -> State Stream m a

78 -> SVar Stream m a

80 -> SVar Stream m a

@@ -82,23 +84,39 @@ workLoopLIFO q st sv winfo = run

83 where

85 where

87+ mrun = runInIO $ svarMrun sv

85 run = do

88 run = do

86 work <- dequeue

89 work <- dequeue

90+ let stop = liftIO $ sendStop sv winfo

87 case work of

91 case work of

88− Nothing -> liftIO $ sendStop sv winfo

92+ Nothing -> stop

89− Just m -> foldStreamSVar sv st yieldk single run m

93+ Just m -> do

94+ -- XXX when we finish we need to send the monadic state back to

95+ -- the parent so that the state can be merged back. We capture

96+ -- and return the state in the stop continuation.

97+ --

98+ -- Instead of using the run function we can just restore the

99+ -- monad state here. That way it can work easily for

100+ -- distributed case as well.

101+ r <- liftIO $ mrun $

102+ foldStreamShared st yieldk single (return Continue) m

103+ res <- restoreM r

104+ case res of

105+ Continue -> run

106+ Suspend -> stop

107

91 single a = do

108 single a = do

92 res <- liftIO $ sendYield sv winfo (ChildYield a)

109 res <- liftIO $ sendYield sv winfo (ChildYield a)

93− if res then run else liftIO $ sendStop sv winfo

110+ return $ if res then Continue else Suspend

111

95 yieldk a r = do

112 yieldk a r = do

96 res <- liftIO $ sendYield sv winfo (ChildYield a)

113 res <- liftIO $ sendYield sv winfo (ChildYield a)

97 if res

114 if res

98− then foldStreamShared st yieldk single run r

115+ then foldStreamShared st yieldk single (return Continue) r

99 else liftIO $ do

116 else liftIO $ do

117+ -- XXX we also need to save the monadic state here

100 enqueueLIFO sv q r

118 enqueueLIFO sv q r

101− sendStop sv winfo

119+ return Suspend

102

120

103 dequeue = liftIO $ atomicModifyIORefCAS q $ \case

121 dequeue = liftIO $ atomicModifyIORefCAS q $ \case

104 [] -> ([], Nothing)

122 [] -> ([], Nothing)

@@ -111,7 +129,7 @@ workLoopLIFO q st sv winfo = run

111-- make a check every time.

129-- make a check every time.

112{-# INLINE workLoopLIFOLimited #-}

130{-# INLINE workLoopLIFOLimited #-}

113workLoopLIFOLimited

131workLoopLIFOLimited

114− :: MonadIO m

132+ :: (MonadIO m, MonadBaseControl IO m)

115 => IORef [Stream m a]

133 => IORef [Stream m a]

116 -> State Stream m a

134 -> State Stream m a

117 -> SVar Stream m a

135 -> SVar Stream m a

@@ -121,10 +139,13 @@ workLoopLIFOLimited q st sv winfo = run

121

139

122 where

140 where

123

141

142+ mrun = runInIO $ svarMrun sv

143+ incrContinue = liftIO (incrementYieldLimit sv) >> return Continue

124 run = do

144 run = do

125 work <- dequeue

145 work <- dequeue

146+ let stop = liftIO $ sendStop sv winfo

126 case work of

147 case work of

127− Nothing -> liftIO $ sendStop sv winfo

148+ Nothing -> stop

128 Just m -> do

149 Just m -> do

129 -- XXX This is just a best effort minimization of concurrency

150 -- XXX This is just a best effort minimization of concurrency

130 -- to the yield limit. If the stream is made of concurrent

151 -- to the yield limit. If the stream is made of concurrent

@@ -135,8 +156,12 @@ workLoopLIFOLimited q st sv winfo = run

135 yieldLimitOk <- liftIO $ decrementYieldLimit sv

156 yieldLimitOk <- liftIO $ decrementYieldLimit sv

136 if yieldLimitOk

157 if yieldLimitOk

137 then do

158 then do

138− let stop = liftIO (incrementYieldLimit sv) >> run

159+ r <- liftIO $ mrun $

139− foldStreamSVar sv st yieldk single stop m

160+ foldStreamShared st yieldk single incrContinue m

161+ res <- restoreM r

162+ case res of

163+ Continue -> run

164+ Suspend -> stop

140 -- Avoid any side effects, undo the yield limit decrement if we

165 -- Avoid any side effects, undo the yield limit decrement if we

141 -- never yielded anything.

166 -- never yielded anything.

142 else liftIO $ do

167 else liftIO $ do

@@ -146,20 +171,19 @@ workLoopLIFOLimited q st sv winfo = run

146

171

147 single a = do

172 single a = do

148 res <- liftIO $ sendYield sv winfo (ChildYield a)

173 res <- liftIO $ sendYield sv winfo (ChildYield a)

149− if res then run else liftIO $ sendStop sv winfo

174+ return $ if res then Continue else Suspend

150

175

151 -- XXX can we pass on the yield limit downstream to limit the concurrency

176 -- XXX can we pass on the yield limit downstream to limit the concurrency

152 -- of constituent streams.

177 -- of constituent streams.

153 yieldk a r = do

178 yieldk a r = do

154 res <- liftIO $ sendYield sv winfo (ChildYield a)

179 res <- liftIO $ sendYield sv winfo (ChildYield a)

155 yieldLimitOk <- liftIO $ decrementYieldLimit sv

180 yieldLimitOk <- liftIO $ decrementYieldLimit sv

156− let stop = liftIO (incrementYieldLimit sv) >> run

157 if res && yieldLimitOk

181 if res && yieldLimitOk

158− then foldStreamShared st yieldk single stop r

182+ then foldStreamShared st yieldk single incrContinue r

159 else liftIO $ do

183 else liftIO $ do

160 incrementYieldLimit sv

184 incrementYieldLimit sv

161 enqueueLIFO sv q r

185 enqueueLIFO sv q r

162− sendStop sv winfo

186+ return Suspend

163

187

164 dequeue = liftIO $ atomicModifyIORefCAS q $ \case

188 dequeue = liftIO $ atomicModifyIORefCAS q $ \case

165 [] -> ([], Nothing)

189 [] -> ([], Nothing)

@@ -173,7 +197,7 @@ workLoopLIFOLimited q st sv winfo = run

173

197

174{-# INLINE workLoopFIFO #-}

198{-# INLINE workLoopFIFO #-}

175workLoopFIFO

199workLoopFIFO

176− :: MonadIO m

200+ :: (MonadIO m, MonadBaseControl IO m)

177 => LinkedQueue (Stream m a)

201 => LinkedQueue (Stream m a)

178 -> State Stream m a

202 -> State Stream m a

179 -> SVar Stream m a

203 -> SVar Stream m a

@@ -183,27 +207,35 @@ workLoopFIFO q st sv winfo = run

183

207

184 where

208 where

185

209

210+ mrun = runInIO $ svarMrun sv

186 run = do

211 run = do

187 work <- liftIO $ tryPopR q

212 work <- liftIO $ tryPopR q

213+ let stop = liftIO $ sendStop sv winfo

188 case work of

214 case work of

189− Nothing -> liftIO $ sendStop sv winfo

215+ Nothing -> stop

190− Just m -> foldStreamSVar sv st yieldk single run m

216+ Just m -> do

217+ r <- liftIO $ mrun $

218+ foldStreamShared st yieldk single (return Continue) m

219+ res <- restoreM r

220+ case res of

221+ Continue -> run

222+ Suspend -> stop

191

223

192 single a = do

224 single a = do

193 res <- liftIO $ sendYield sv winfo (ChildYield a)

225 res <- liftIO $ sendYield sv winfo (ChildYield a)

194− if res then run else liftIO $ sendStop sv winfo

226+ return $ if res then Continue else Suspend

195

227

196 yieldk a r = do

228 yieldk a r = do

197 res <- liftIO $ sendYield sv winfo (ChildYield a)

229 res <- liftIO $ sendYield sv winfo (ChildYield a)

198 if res

230 if res

199− then foldStreamShared st yieldk single run r

231+ then foldStreamShared st yieldk single (return Continue) r

200 else liftIO $ do

232 else liftIO $ do

201 enqueueFIFO sv q r

233 enqueueFIFO sv q r

202− sendStop sv winfo

234+ return Suspend

203

235

204{-# INLINE workLoopFIFOLimited #-}

236{-# INLINE workLoopFIFOLimited #-}

205workLoopFIFOLimited

237workLoopFIFOLimited

206− :: MonadIO m

238+ :: (MonadIO m, MonadBaseControl IO m)

207 => LinkedQueue (Stream m a)

239 => LinkedQueue (Stream m a)

208 -> State Stream m a

240 -> State Stream m a

209 -> SVar Stream m a

241 -> SVar Stream m a

@@ -213,16 +245,23 @@ workLoopFIFOLimited q st sv winfo = run

213

245

214 where

246 where

215

247

248+ mrun = runInIO $ svarMrun sv

249+ incrContinue = liftIO (incrementYieldLimit sv) >> return Continue

216 run = do

250 run = do

217 work <- liftIO $ tryPopR q

251 work <- liftIO $ tryPopR q

252+ let stop = liftIO $ sendStop sv winfo

218 case work of

253 case work of

219− Nothing -> liftIO $ sendStop sv winfo

254+ Nothing -> stop

220 Just m -> do

255 Just m -> do

221 yieldLimitOk <- liftIO $ decrementYieldLimit sv

256 yieldLimitOk <- liftIO $ decrementYieldLimit sv

222 if yieldLimitOk

257 if yieldLimitOk

223 then do

258 then do

224− let stop = liftIO (incrementYieldLimit sv) >> run

259+ r <- liftIO $ mrun $

225− foldStreamSVar sv st yieldk single stop m

260+ foldStreamShared st yieldk single incrContinue m

261+ res <- restoreM r

262+ case res of

263+ Continue -> run

264+ Suspend -> stop

226 else liftIO $ do

265 else liftIO $ do

227 enqueueFIFO sv q m

266 enqueueFIFO sv q m

228 incrementYieldLimit sv

267 incrementYieldLimit sv

@@ -230,18 +269,17 @@ workLoopFIFOLimited q st sv winfo = run

230

269

231 single a = do

270 single a = do

232 res <- liftIO $ sendYield sv winfo (ChildYield a)

271 res <- liftIO $ sendYield sv winfo (ChildYield a)

233− if res then run else liftIO $ sendStop sv winfo

272+ return $ if res then Continue else Suspend

234

273

235 yieldk a r = do

274 yieldk a r = do

236 res <- liftIO $ sendYield sv winfo (ChildYield a)

275 res <- liftIO $ sendYield sv winfo (ChildYield a)

237 yieldLimitOk <- liftIO $ decrementYieldLimit sv

276 yieldLimitOk <- liftIO $ decrementYieldLimit sv

238− let stop = liftIO (incrementYieldLimit sv) >> run

239 if res && yieldLimitOk

277 if res && yieldLimitOk

240− then foldStreamShared st yieldk single stop r

278+ then foldStreamShared st yieldk single incrContinue r

241 else liftIO $ do

279 else liftIO $ do

242 incrementYieldLimit sv

280 incrementYieldLimit sv

243 enqueueFIFO sv q r

281 enqueueFIFO sv q r

244− sendStop sv winfo

282+ return Suspend

245

283

246-------------------------------------------------------------------------------

284-------------------------------------------------------------------------------

247-- SVar creation

285-- SVar creation

src/Streamly/Internal/Data/Stream/StreamK.hs+1−2

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

47 -- * Elimination Primitives

48 , foldStream

49 , foldStreamShared

50− , foldStreamSVar

52 -- * Transformation Primitives

51 -- * Transformation Primitives

53 , unShare

52 , unShare

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

67 , replicateM

66 , replicateM

68 , fromIndices

67 , fromIndices

69 , fromIndicesM

68 , fromIndicesM

70− , iterate

69+ , iterate

71 , iterateM

70 , iterateM

73 -- ** Conversions

72 -- ** Conversions

src/Streamly/Internal/Data/Stream/StreamK/Type.hs+1−19

@@ -47,7 +47,6 @@ module Streamly.Internal.Data.Stream.StreamK.Type

47 -- * Elimination

48 , foldStream

49 , foldStreamShared

50− , foldStreamSVar

52 -- * foldr/build

51 -- * foldr/build

53 , foldrM

52 , foldrM

@@ -85,8 +84,7 @@ module Streamly.Internal.Data.Stream.StreamK.Type

85 )

84 )

86where

85where

88−import Control.Monad (void, ap, (>=>))

87+import Control.Monad (ap, (>=>))

89−import Control.Monad.IO.Class (MonadIO(liftIO))

90import Control.Monad.Trans.Class (MonadTrans(lift))

88import Control.Monad.Trans.Class (MonadTrans(lift))

91#if __GLASGOW_HASKELL__ >= 800

89#if __GLASGOW_HASKELL__ >= 800

92import Data.Kind (Type)

90import Data.Kind (Type)

@@ -415,22 +413,6 @@ foldStream st yld sng stp m =

415 MkStream k = toStream m

413 MkStream k = toStream m

416 in k (adaptState st) yieldk sng stp

414 in k (adaptState st) yieldk sng stp

417

415

418−-- Run the stream using a run function associated with the SVar that runs the

419−-- streams with a captured snapshot of the monadic state.

420−{-# INLINE foldStreamSVar #-}

421−foldStreamSVar

422− :: (IsStream t, MonadIO m)

423− => SVar Stream m a

424− -> State Stream m a -- state

425− -> (a -> t m a -> m r) -- yield

426− -> (a -> m r) -- singleton

427− -> m r -- stop

428− -> t m a

429− -> m ()

430−foldStreamSVar sv st yld sng stp m =

431− let mrun = runInIO $ svarMrun sv

432− in void $ liftIO $ mrun $ foldStreamShared st yld sng stp m

433−

434-------------------------------------------------------------------------------

416-------------------------------------------------------------------------------

435-- Instances

417-- Instances

436-------------------------------------------------------------------------------

418-------------------------------------------------------------------------------