b33e9c2 · streamly · git.iko.soy

git.iko.soy/streamly

on master help sign in

performance improvements after maxrate changes

b33e9c2Harendra Kumar8 years ago

src/Streamly/SVar.hs+34−28

compare with master

unifiedsplit

@@ -561,6 +561,29 @@ doFork action exHandler =

561 exHandler

562 runInIO (return tid)

563

564+-- XXX Can we make access to remainingYields and yieldRateInfo fields in sv

565+-- faster, along with the fields in sv required by send?

566+-- XXX make it noinline

567+{-# INLINE decrementYieldLimit #-}

568+decrementYieldLimit :: SVar t m a -> IO Bool

569+decrementYieldLimit sv =

570+ case remainingYields sv of

571+ Nothing -> return True

572+ Just ref -> do

573+ r <- atomicModifyIORefCAS ref $ \x ->

574+ (if x >= 1 then x - 1 else 0, x)

575+ return $ r >= 1

576+

577+-- XXX increment is not safe when the yield limit was already 0

578+-- and therefore was not decremented. We need yield limit to be

579+-- int and negative if we want this to work.

580+{-# INLINE incrementYieldLimit #-}

581+incrementYieldLimit :: SVar t m a -> IO ()

582+incrementYieldLimit sv =

583+ case remainingYields sv of

584+ Nothing -> return ()

585+ Just ref -> atomicModifyIORefCAS_ ref (+ 1)

586+

564-- XXX exception safety of all atomic/MVar operations

587-- XXX exception safety of all atomic/MVar operations

565

588

566-- TBD Each worker can have their own queue and the consumer can empty one

589-- TBD Each worker can have their own queue and the consumer can empty one

@@ -648,27 +671,6 @@ workerRateControl sv yinfo winfo = do

648 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt

671 beyondMaxRate <- checkRatePeriodic sv yinfo winfo cnt

649 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)

672 return $ not (isBeyondMaxYield cnt winfo || beyondMaxRate)

650

673

651−-- XXX Can we make access to remainingYields and yieldRateInfo fields in sv

652−-- faster, along with the fields in sv required by send?

653−-- XXX make it noinline

654−decrementYieldLimit :: SVar t m a -> IO Bool

655−decrementYieldLimit sv =

656− case remainingYields sv of

657− Nothing -> return True

658− Just ref -> do

659− r <- atomicModifyIORefCAS ref $ \x ->

660− (if x >= 1 then x - 1 else 0, x)

661− return $ r >= 1

662−

663−-- XXX increment is not safe when the yield limit was already 0

664−-- and therefore was not decremented. We need yield limit to be

665−-- int and negative if we want this to work.

666−incrementYieldLimit :: SVar t m a -> IO ()

667−incrementYieldLimit sv =

668− case remainingYields sv of

669− Nothing -> return ()

670− Just ref -> atomicModifyIORefCAS_ ref (+ 1)

671−

672-- XXX we should do rate control here but not latency update in case of ahead

674-- XXX we should do rate control here but not latency update in case of ahead

673-- streams. latency update must be done when we yield directly to outputQueue

675-- streams. latency update must be done when we yield directly to outputQueue

674-- or when we yield to heap.

676-- or when we yield to heap.

@@ -682,18 +684,22 @@ sendYield sv winfo msg = do

682 Just yinfo -> workerRateControl sv yinfo winfo

684 Just yinfo -> workerRateControl sv yinfo winfo

683 return $ r && rateLimitOk

685 return $ r && rateLimitOk

684

686

687+{-# INLINE workerStopUpdate #-}

688+workerStopUpdate :: WorkerInfo -> YieldRateInfo -> Int -> IO ()

689+workerStopUpdate winfo info n = do

690+ i <- readIORef (workerLatencyPeriod info)

691+ when (i /= 0) $ workerUpdateLatency info winfo

692+ when (n == 1) $ do

693+ t <- getTime Monotonic

694+ writeIORef (workerStopTimeStamp info) t

695+

685{-# INLINABLE sendStop #-}

696{-# INLINABLE sendStop #-}

686sendStop :: SVar t m a -> WorkerInfo -> IO ()

697sendStop :: SVar t m a -> WorkerInfo -> IO ()

687sendStop sv winfo = do

698sendStop sv winfo = do

688 n <- liftIO $ atomicModifyIORefCAS (workerCount sv) $ \n -> (n - 1, n)

699 n <- liftIO $ atomicModifyIORefCAS (workerCount sv) $ \n -> (n - 1, n)

689 case yieldRateInfo sv of

700 case yieldRateInfo sv of

690 Nothing -> return ()

701 Nothing -> return ()

691− Just info -> do

702+ Just info -> workerStopUpdate winfo info n

692− i <- readIORef (workerLatencyPeriod info)

693− when (i /= 0) $ workerUpdateLatency info winfo

694− when (n == 1) $ do

695− t <- getTime Monotonic

696− writeIORef (workerStopTimeStamp info) t

697 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)

703 myThreadId >>= \tid -> void $ send sv (ChildStop tid Nothing)

698

704

699-------------------------------------------------------------------------------

705-------------------------------------------------------------------------------

@@ -1099,7 +1105,7 @@ getWorkerLatency yinfo = do

1099

1105

1100 let pendingCount = colCount + count

1106 let pendingCount = colCount + count

1101 pendingTime = colTime + time

1107 pendingTime = colTime + time

1102− -- Take the average of previous and the latest

1108+ -- XXX Take the average of previous and the latest

1103 new =

1109 new =

1104 if pendingCount > 0

1110 if pendingCount > 0

1105 then ((pendingTime `div` (fromIntegral pendingCount)) + prev)

1111 then ((pendingTime `div` (fromIntegral pendingCount)) + prev)

src/Streamly/Streams/Ahead.hs+162−102

@@ -127,14 +127,104 @@ import Prelude hiding (map)

127-- updates the heap count and the other updates the yield count.

128-- we can pass that as an update function instead.

129

130−workLoopAhead :: MonadIO m

130+processHeap :: MonadIO m

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

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

133 -> State Stream m a

134 -> SVar Stream m a

135 -> WorkerInfo

136+ -> AheadHeapEntry Stream m a

137+ -> Int

136 -> m ()

138 -> m ()

139+processHeap q heap st sv winfo entry sno = loopHeap sno entry

140+

141+ where

142+

143+ loopHeap seqNo ent = do

144+#ifdef DIAGNOSTICS

145+ liftIO $ do

146+ maxHp <- readIORef (maxHeapSize sv)

147+ (hp, _) <- readIORef heap

148+ when (H.size hp > maxHp) $ writeIORef (maxHeapSize sv) (H.size hp)

149+#endif

150+ case ent of

151+ AheadEntryPure a -> do

152+ -- Use 'send' directly so that we do not account this in worker

153+ -- latency as this will not be the real latency.

154+ continue <- liftIO $ send sv (ChildYield a)

155+ if continue

156+ then nextHeap seqNo

157+ else stopHeap seqNo

158+ AheadEntryStream r -> runStreamWithYieldLimit True seqNo r

159+

160+ stopHeap seqNo = liftIO $ do

161+ atomicModifyIORef heap $ \(h, _) -> ((h, seqNo + 1), ())

162+ sendStop sv winfo

163+

164+ nextHeap prevSeqNo = do

165+ -- XXX use "dequeueIfSeqential prevSeqNo" instead of always

166+ -- updating the sequence number in heap.

167+ liftIO $ atomicModifyIORef heap $ \(h, _) -> ((h, prevSeqNo + 1), ())

168+ ent <- liftIO $ dequeueFromHeap heap

169+ case ent of

170+ Just (Entry seqNo hent) -> loopHeap seqNo hent

171+ Nothing -> do

172+ work <- dequeueAhead q

173+ case work of

174+ Nothing -> liftIO $ sendStop sv winfo

175+ Just (m, seqNo) -> do

176+ yieldLimitOk <- liftIO $ decrementYieldLimit sv

177+ if yieldLimitOk

178+ then do

179+ if seqNo == prevSeqNo + 1

180+ then processWithToken q heap st sv winfo m seqNo

181+ else processWithoutToken q heap st sv winfo m seqNo

182+ else liftIO $ do

183+ incrementYieldLimit sv

184+ sendStop sv winfo

185+

186+ singleStreamFromHeap seqNo a = do

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

188+ if continue

189+ then nextHeap seqNo

190+ else stopHeap seqNo

191+

192+ runStreamWithYieldLimit continue seqNo r = do

193+ yieldLimitOk <- liftIO $ decrementYieldLimit sv

194+ if continue && yieldLimitOk

195+ then do

196+ let stop = do

197+ liftIO (incrementYieldLimit sv)

198+ nextHeap seqNo

199+ unStream r st stop

200+ (singleStreamFromHeap seqNo)

201+ (yieldStreamFromHeap seqNo)

202+ else liftIO $ do

203+ atomicModifyIORef heap $ \(h, _) ->

204+ ((H.insert (Entry seqNo (AheadEntryStream r)) h, seqNo), ())

205+ incrementYieldLimit sv

206+ sendStop sv winfo

207+

208+ yieldStreamFromHeap seqNo a r = do

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

210+ runStreamWithYieldLimit continue seqNo r

211+

212+processWithoutToken :: MonadIO m

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

214+ -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

215+ -> State Stream m a

216+ -> SVar Stream m a

217+ -> WorkerInfo

218+ -> Stream m a

219+ -> Int

220+ -> m ()

221+processWithoutToken q heap st sv winfo m sno = do

222+ -- we have already decremented the yield limit for m

223+ let stop = do

224+ liftIO (incrementYieldLimit sv)

137−workLoopAhead q heap st sv winfo = runHeap

225+ workLoopAhead q heap st sv winfo

226+

227+ unStream m st stop (singleToHeap sno) (yieldToHeap sno)

138

228

139 where

229 where

140

230

@@ -170,7 +260,7 @@ workLoopAhead q heap st sv winfo = runHeap

170 if heapOk

260 if heapOk

171 then

261 then

172 case yieldRateInfo sv of

262 case yieldRateInfo sv of

173− Nothing -> runHeap

263+ Nothing -> workLoopAhead q heap st sv winfo

174 Just yinfo -> do

264 Just yinfo -> do

175 -- check the stop condition under a lock before actually

265 -- check the stop condition under a lock before actually

176 -- stopping so that the whole herd does not stop at once.

266 -- stopping so that the whole herd does not stop at once.

@@ -179,7 +269,7 @@ workLoopAhead q heap st sv winfo = runHeap

179 if rateOk

269 if rateOk

180 then do

270 then do

181 liftIO $ putMVar (workerStopMVar yinfo) ()

271 liftIO $ putMVar (workerStopMVar yinfo) ()

182− runHeap

272+ workLoopAhead q heap st sv winfo

183 else do

273 else do

184 liftIO $ sendStop sv winfo

274 liftIO $ sendStop sv winfo

185 liftIO $ putMVar (workerStopMVar yinfo) ()

275 liftIO $ putMVar (workerStopMVar yinfo) ()

@@ -188,125 +278,103 @@ workLoopAhead q heap st sv winfo = runHeap

188 singleToHeap seqNo a = toHeap seqNo (AheadEntryPure a)

278 singleToHeap seqNo a = toHeap seqNo (AheadEntryPure a)

189 yieldToHeap seqNo a r = toHeap seqNo (AheadEntryStream (a `K.cons` r))

279 yieldToHeap seqNo a r = toHeap seqNo (AheadEntryStream (a `K.cons` r))

190

280

281+processWithToken :: MonadIO m

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

283+ -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

284+ -> State Stream m a

285+ -> SVar Stream m a

286+ -> WorkerInfo

287+ -> Stream m a

288+ -> Int

289+ -> m ()

290+processWithToken q heap st sv winfo action sno = do

291+ let stop = do

292+ liftIO (incrementYieldLimit sv)

293+ loopWithToken sno

294+

295+ unStream action st stop

296+ (singleOutput sno)

297+ (yieldOutput sno)

298+

299+ where

300+

191 singleOutput seqNo a = do

301 singleOutput seqNo a = do

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

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

193 if continue

303 if continue

194− then runQueueToken seqNo

304+ then loopWithToken seqNo

195 else liftIO $ do

305 else liftIO $ do

196 atomicModifyIORef heap $ \(h, _) -> ((h, seqNo + 1), ())

306 atomicModifyIORef heap $ \(h, _) -> ((h, seqNo + 1), ())

197 sendStop sv winfo

307 sendStop sv winfo

198

308

199− singleFromHeap seqNo a = do

200− continue <- liftIO $ send sv (ChildYield a)

201− liftIO $ atomicModifyIORef heap $ \(h, _) -> ((h, seqNo + 1), ())

202− if continue

203− then runHeap

204− else liftIO $ sendStop sv winfo

205−

206− singleStreamFromHeap seqNo a = do

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

208− liftIO $ atomicModifyIORef heap $ \(h, _) -> ((h, seqNo + 1), ())

209− if continue

210− then runHeap

211− else liftIO $ sendStop sv winfo

212−

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

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

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

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

215 -- "unstream" calls in this function must increment yield limit on stop.

311 -- "unstream" calls in this function must increment yield limit on stop.

216 yieldOutput seqNo a r = do

312 yieldOutput seqNo a r = do

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

218− yieldLimitOk <- liftIO $ decrementYieldLimit sv

219− if continue && yieldLimitOk

220− then do

221− let stop = liftIO (incrementYieldLimit sv) >> runQueueToken seqNo

222− unStream r st stop

223− (singleOutput seqNo)

224− (yieldOutput seqNo)

225− else liftIO $ do

226− atomicModifyIORef heap $ \(h, _) ->

227− ((H.insert (Entry seqNo (AheadEntryStream r)) h, seqNo), ())

228− incrementYieldLimit sv

229− sendStop sv winfo

230−

231− yieldStreamFromHeap seqNo a r = do

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

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

233 yieldLimitOk <- liftIO $ decrementYieldLimit sv

314 yieldLimitOk <- liftIO $ decrementYieldLimit sv

234 if continue && yieldLimitOk

315 if continue && yieldLimitOk

235 then do

316 then do

236 let stop = do

317 let stop = do

237− liftIO $ atomicModifyIORef heap $ \(h, _) ->

238− ((h, seqNo + 1), ())

239 liftIO (incrementYieldLimit sv)

318 liftIO (incrementYieldLimit sv)

240− runHeap

319+ loopWithToken seqNo

241 unStream r st stop

320 unStream r st stop

242− (singleStreamFromHeap seqNo)

321+ (singleOutput seqNo)

243− (yieldStreamFromHeap seqNo)

322+ (yieldOutput seqNo)

244 else liftIO $ do

323 else liftIO $ do

245 atomicModifyIORef heap $ \(h, _) ->

324 atomicModifyIORef heap $ \(h, _) ->

246− ((H.insert (Entry seqNo (AheadEntryStream r)) h, seqNo), ())

325+ ((H.insert (Entry seqNo (AheadEntryStream r)) h, seqNo), ())

247 incrementYieldLimit sv

326 incrementYieldLimit sv

248 sendStop sv winfo

327 sendStop sv winfo

249

328

250− {-# INLINE runQueueToken #-}

251− runQueueToken prevSeqNo = do

329+ loopWithToken prevSeqNo = do

252 work <- dequeueAhead q

330 work <- dequeueAhead q

253 case work of

331 case work of

254 Nothing -> do

332 Nothing -> do

255 liftIO $ atomicModifyIORef heap $ \(h, _) ->

333 liftIO $ atomicModifyIORef heap $ \(h, _) ->

256 ((h, prevSeqNo + 1), ())

334 ((h, prevSeqNo + 1), ())

257− runHeap

335+

336+ ent <- liftIO $ dequeueFromHeap heap

337+ case ent of

338+ Just (Entry seqNo hent) ->

339+ processHeap q heap st sv winfo hent seqNo

340+ Nothing -> liftIO $ sendStop sv winfo

341+

258 Just (m, seqNo) -> do

342 Just (m, seqNo) -> do

259 yieldLimitOk <- liftIO $ decrementYieldLimit sv

343 yieldLimitOk <- liftIO $ decrementYieldLimit sv

260 if yieldLimitOk

344 if yieldLimitOk

261 then do

345 then do

262 if seqNo == prevSeqNo + 1

346 if seqNo == prevSeqNo + 1

263 then do

347 then do

348+ let stop = do

264− let stop = liftIO (incrementYieldLimit sv)

349+ liftIO (incrementYieldLimit sv)

265− >> runQueueToken seqNo

350+ loopWithToken seqNo

266 unStream m st stop

351 unStream m st stop

267 (singleOutput seqNo)

352 (singleOutput seqNo)

268 (yieldOutput seqNo)

353 (yieldOutput seqNo)

269 else do

354 else do

270 liftIO $ atomicModifyIORef heap $ \(h, _) ->

355 liftIO $ atomicModifyIORef heap $ \(h, _) ->

271− ((h, prevSeqNo + 1), ())

356+ ((h, prevSeqNo + 1), ())

272− let stop = liftIO (incrementYieldLimit sv)

357+ -- XXX put it back on the queue and start with the Heap?

273− >> runHeap

358+ processWithoutToken q heap st sv winfo m seqNo

274− unStream m st stop

275− (singleToHeap seqNo)

276− (yieldToHeap seqNo)

277 else liftIO $ do

359 else liftIO $ do

278 atomicModifyIORef heap $ \(h, _) ->

360 atomicModifyIORef heap $ \(h, _) ->

279− ((h, prevSeqNo + 1), ())

361+ ((h, prevSeqNo + 1), ())

280 incrementYieldLimit sv

362 incrementYieldLimit sv

281 sendStop sv winfo

363 sendStop sv winfo

282

364

283− runQueueNoToken = do

365+-- XXX the yield limit changes increased the performance overhead by 30-40%.

284− work <- dequeueAhead q

366+-- Just like AsyncT we can use an implementation without yeidlimit and even

285− case work of

367+-- without pacing code to keep the performance higher in the unlimited and

286− Nothing -> runHeap

368+-- unpaced case.

287− Just (m, seqNo) -> do

369+{-# INLINE workLoopAhead #-}

288− yieldLimitOk <- liftIO $ decrementYieldLimit sv

370+workLoopAhead :: MonadIO m

289− if yieldLimitOk

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

290− then do

372+ -> IORef (Heap (Entry Int (AheadHeapEntry Stream m a)) , Int)

291− if seqNo == 0

373+ -> State Stream m a

292− then do

374+ -> SVar Stream m a

293− let stop = liftIO (incrementYieldLimit sv)

375+ -> WorkerInfo

294− >> runQueueToken seqNo

376+ -> m ()

295− unStream m st stop

377+workLoopAhead q heap st sv winfo = do

296− (singleOutput seqNo)

297− (yieldOutput seqNo)

298− else do

299− let stop = liftIO (incrementYieldLimit sv)

300− >> runHeap

301− unStream m st stop

302− (singleToHeap seqNo)

303− (yieldToHeap seqNo)

304− else liftIO $ do

305− incrementYieldLimit sv

306− sendStop sv winfo

307−

308− {-# NOINLINE runHeap #-}

309− runHeap = do

310#ifdef DIAGNOSTICS

378#ifdef DIAGNOSTICS

311 liftIO $ do

379 liftIO $ do

312 maxHp <- readIORef (maxHeapSize sv)

380 maxHp <- readIORef (maxHeapSize sv)

@@ -316,7 +384,7 @@ workLoopAhead q heap st sv winfo = runHeap

316 ent <- liftIO $ dequeueFromHeap heap

384 ent <- liftIO $ dequeueFromHeap heap

317 case ent of

385 case ent of

318 Nothing -> do

386 Nothing -> do

319− -- Before we pick up the next item from the work queue we check

387+ -- Before we execute the next item from the work queue we check

320 -- if we are beyond the yield limit. It is better to check the

388 -- if we are beyond the yield limit. It is better to check the

321 -- yield limit before we pick up the next item. Otherwise we

389 -- yield limit before we pick up the next item. Otherwise we

322 -- may have already started more tasks even though we may have

390 -- may have already started more tasks even though we may have

@@ -325,8 +393,6 @@ workLoopAhead q heap st sv winfo = runHeap

325 -- workers may go away without picking up work and yielding a

393 -- workers may go away without picking up work and yielding a

326 -- value.

394 -- value.

327 --

395 --

328− -- XXX do the same for Async stream as well.

329− --

330 -- Rate control can be done either based on actual yields in

396 -- Rate control can be done either based on actual yields in

331 -- the output queue or based on any yield either to the heap or

397 -- the output queue or based on any yield either to the heap or

332 -- to the output queue. In both cases we may have one issue or

398 -- to the output queue. In both cases we may have one issue or

@@ -334,26 +400,20 @@ workLoopAhead q heap st sv winfo = runHeap

334 -- output queue because it makes the code common to both async

400 -- output queue because it makes the code common to both async

335 -- and ahead streams.

401 -- and ahead streams.

336 --

402 --

337− done <- queueEmptyAhead q

403+ work <- dequeueAhead q

338− if done

339− -- XXX need to run heap here if not empty

340− then liftIO $ sendStop sv winfo

341− else runQueueNoToken

342− Just (Entry seqNo hent) -> do

343− case hent of

404+ case work of

344− -- Do not account this in worker latency as this will not

405+ Nothing -> liftIO $ sendStop sv winfo

345− -- be the real latency.

406+ Just (m, seqNo) -> do

346− AheadEntryPure a -> singleFromHeap seqNo a

407+ yieldLimitOk <- liftIO $ decrementYieldLimit sv

347− AheadEntryStream r -> do

408+ if yieldLimitOk

348− let stp = do

409+ then do

349− liftIO $ atomicModifyIORef heap $ \(h, _) ->

410+ if seqNo == 0

350− ((h, seqNo + 1), ())

411+ then processWithToken q heap st sv winfo m seqNo

412+ else processWithoutToken q heap st sv winfo m seqNo

413+ else liftIO $ do

351− liftIO (incrementYieldLimit sv)

414+ incrementYieldLimit sv

352− runHeap

415+ sendStop sv winfo

353−

416+ Just (Entry seqNo hent) -> processHeap q heap st sv winfo hent seqNo

354− unStream r st stp

355− (singleStreamFromHeap seqNo)

356− (yieldStreamFromHeap seqNo)

357

417

358-------------------------------------------------------------------------------

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

359-- WAhead

419-- WAhead

src/Streamly/Streams/Async.hs+120−12

@@ -77,6 +77,45 @@ workLoopLIFO q st sv winfo = run

78 where

80+ run = do

81+ work <- dequeue

82+ case work of

83+ Nothing -> liftIO $ sendStop sv winfo

84+ Just m -> unStream m st run single yieldk

85+

86+ single a = do

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

88+ if res then run else liftIO $ sendStop sv winfo

89+

90+ yieldk a r = do

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

92+ if res

93+ then unStream r st run single yieldk

94+ else liftIO $ do

95+ enqueueLIFO sv q r

96+ sendStop sv winfo

97+

98+ dequeue = liftIO $ atomicModifyIORefCAS q $ \case

99+ [] -> ([], Nothing)

100+ x : xs -> (xs, Just x)

101+

102+-- We duplicate workLoop for yield limit and no limit cases because it has

103+-- around 40% performance overhead in the worst case.

104+--

105+-- XXX we can pass yinfo directly as an argument here so that we do not have to

106+-- make a check every time.

107+{-# INLINE workLoopLIFOLimited #-}

108+workLoopLIFOLimited

109+ :: MonadIO m

110+ => IORef [Stream m a]

111+ -> State Stream m a

112+ -> SVar Stream m a

113+ -> WorkerInfo

114+ -> m ()

115+workLoopLIFOLimited q st sv winfo = run

116+

117+ where

118+

80 run = do

119 run = do

81 work <- dequeue

120 work <- dequeue

82 case work of

121 case work of

@@ -139,6 +178,36 @@ workLoopFIFO q st sv winfo = run

139

178

140 where

179 where

141

180

181+ run = do

182+ work <- liftIO $ tryPopR q

183+ case work of

184+ Nothing -> liftIO $ sendStop sv winfo

185+ Just m -> unStream m st run single yieldk

186+

187+ single a = do

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

189+ if res then run else liftIO $ sendStop sv winfo

190+

191+ yieldk a r = do

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

193+ if res

194+ then unStream r st run single yieldk

195+ else liftIO $ do

196+ enqueueFIFO sv q r

197+ sendStop sv winfo

198+

199+{-# INLINE workLoopFIFOLimited #-}

200+workLoopFIFOLimited

201+ :: MonadIO m

202+ => LinkedQueue (Stream m a)

203+ -> State Stream m a

204+ -> SVar Stream m a

205+ -> WorkerInfo

206+ -> m ()

207+workLoopFIFOLimited q st sv winfo = run

208+

209+ where

210+

142 run = do

211 run = do

143 work <- liftIO $ tryPopR q

212 work <- liftIO $ tryPopR q

144 case work of

213 case work of

@@ -197,7 +266,9 @@ getLifoSVar st = do

197 maxHs <- newIORef 0

266 maxHs <- newIORef 0

198 maxWq <- newIORef 0

267 maxWq <- newIORef 0

199#endif

268#endif

200− let isWorkFinished sv = do

269+ let isWorkFinished _ = null <$> readIORef q

270+

271+ let isWorkFinishedLimited sv = do

201 yieldsDone <-

272 yieldsDone <-

202 case remainingYields sv of

273 case remainingYields sv of

203 Just ref -> do

274 Just ref -> do

@@ -207,7 +278,7 @@ getLifoSVar st = do

207 qEmpty <- null <$> readIORef q

278 qEmpty <- null <$> readIORef q

208 return $ qEmpty || yieldsDone

279 return $ qEmpty || yieldsDone

209

280

210− let getSVar sv readOutput postProc = SVar

281+ let getSVar sv readOutput postProc workDone wloop = SVar

211 { outputQueue = outQ

282 { outputQueue = outQ

212 , remainingYields = yl

283 , remainingYields = yl

213 , maxBufferLimit = getMaxBuffer st

284 , maxBufferLimit = getMaxBuffer st

@@ -217,9 +288,9 @@ getLifoSVar st = do

217 , readOutputQ = readOutput sv

288 , readOutputQ = readOutput sv

218 , postProcess = postProc sv

289 , postProcess = postProc sv

219 , workerThreads = running

290 , workerThreads = running

220− , workLoop = workLoopLIFO q st{streamVar = Just sv} sv

291+ , workLoop = wloop q st{streamVar = Just sv} sv

221 , enqueue = enqueueLIFO sv q

292 , enqueue = enqueueLIFO sv q

222− , isWorkDone = isWorkFinished sv

293+ , isWorkDone = workDone sv

223 , needDoorBell = wfw

294 , needDoorBell = wfw

224 , svarStyle = AsyncVar

295 , svarStyle = AsyncVar

225 , workerCount = active

296 , workerCount = active

@@ -237,8 +308,26 @@ getLifoSVar st = do

237

308

238 let sv =

309 let sv =

239 case getMaxStreamRate st of

310 case getMaxStreamRate st of

311+ Nothing ->

312+ case getYieldLimit st of

240− Nothing -> getSVar sv readOutputQBounded postProcessBounded

313+ Nothing -> getSVar sv readOutputQBounded

314+ postProcessBounded

315+ isWorkFinished

316+ workLoopLIFO

317+ Just _ -> getSVar sv readOutputQBounded

318+ postProcessBounded

319+ isWorkFinishedLimited

320+ workLoopLIFOLimited

321+ Just _ ->

322+ case getYieldLimit st of

241− Just _ -> getSVar sv readOutputQPaced postProcessPaced

323+ Nothing -> getSVar sv readOutputQPaced

324+ postProcessPaced

325+ isWorkFinished

326+ workLoopLIFO

327+ Just _ -> getSVar sv readOutputQPaced

328+ postProcessPaced

329+ isWorkFinishedLimited

330+ workLoopLIFOLimited

242 in return sv

331 in return sv

243

332

244getFifoSVar :: MonadAsync m => State Stream m a -> IO (SVar Stream m a)

333getFifoSVar :: MonadAsync m => State Stream m a -> IO (SVar Stream m a)

@@ -262,7 +351,8 @@ getFifoSVar st = do

262 maxWq <- newIORef 0

351 maxWq <- newIORef 0

263#endif

352#endif

264

353

265− let isWorkFinished sv = do

354+ let isWorkFinished _ = nullQ q

355+ let isWorkFinishedLimited sv = do

266 yieldsDone <-

356 yieldsDone <-

267 case remainingYields sv of

357 case remainingYields sv of

268 Just ref -> do

358 Just ref -> do

@@ -272,7 +362,7 @@ getFifoSVar st = do

272 qEmpty <- nullQ q

362 qEmpty <- nullQ q

273 return $ qEmpty || yieldsDone

363 return $ qEmpty || yieldsDone

274

364

275− let getSVar sv readOutput postProc = SVar

365+ let getSVar sv readOutput postProc workDone wloop = SVar

276 { outputQueue = outQ

366 { outputQueue = outQ

277 , remainingYields = yl

367 , remainingYields = yl

278 , maxBufferLimit = getMaxBuffer st

368 , maxBufferLimit = getMaxBuffer st

@@ -282,9 +372,9 @@ getFifoSVar st = do

282 , readOutputQ = readOutput sv

372 , readOutputQ = readOutput sv

283 , postProcess = postProc sv

373 , postProcess = postProc sv

284 , workerThreads = running

374 , workerThreads = running

285− , workLoop = workLoopFIFO q st{streamVar = Just sv} sv

375+ , workLoop = wloop q st{streamVar = Just sv} sv

286 , enqueue = enqueueFIFO sv q

376 , enqueue = enqueueFIFO sv q

287− , isWorkDone = isWorkFinished sv

377+ , isWorkDone = workDone sv

288 , needDoorBell = wfw

378 , needDoorBell = wfw

289 , svarStyle = WAsyncVar

379 , svarStyle = WAsyncVar

290 , workerCount = active

380 , workerCount = active

@@ -302,8 +392,26 @@ getFifoSVar st = do

302

392

303 let sv =

393 let sv =

304 case getMaxStreamRate st of

394 case getMaxStreamRate st of

395+ Nothing ->

396+ case getYieldLimit st of

305− Nothing -> getSVar sv readOutputQBounded postProcessBounded

397+ Nothing -> getSVar sv readOutputQBounded

398+ postProcessBounded

399+ isWorkFinished

400+ workLoopFIFO

401+ Just _ -> getSVar sv readOutputQBounded

402+ postProcessBounded

403+ isWorkFinishedLimited

404+ workLoopFIFOLimited

405+ Just _ ->

406+ case getYieldLimit st of

306− Just _ -> getSVar sv readOutputQPaced postProcessPaced

407+ Nothing -> getSVar sv readOutputQPaced

408+ postProcessPaced

409+ isWorkFinished

410+ workLoopFIFO

411+ Just _ -> getSVar sv readOutputQPaced

412+ postProcessPaced

413+ isWorkFinishedLimited

414+ workLoopFIFOLimited

307 in return sv

415 in return sv

308

416

309{-# INLINABLE newAsyncVar #-}

417{-# INLINABLE newAsyncVar #-}

src/Streamly/Streams/SVar.hs+2−5

@@ -78,11 +78,8 @@ fromStreamVar sv = Stream $ \st stp sng yld -> do

78 processEvents (ev : es) = Stream $ \st stp sng yld -> do

79 let rest = processEvents es

80 case ev of

81− ChildYield a -> do

81+ ChildYield a -> yld a rest

82−-- void $ dispatchWorkerPaced sv

83− yld a rest

84 ChildStop tid e -> do

82 ChildStop tid e -> do

85− -- void $ dispatchWorkerPaced sv

86 accountThread sv tid

83 accountThread sv tid

87 case e of

84 case e of

88 Nothing -> unStream rest (rstState st) stp sng yld

85 Nothing -> unStream rest (rstState st) stp sng yld

@@ -104,7 +101,7 @@ toSVar sv m = toStreamVar sv (toStream m)

104-- XXX need to write these in direct style otherwise they will break fusion.

101-- XXX need to write these in direct style otherwise they will break fusion.

105--

102--

106-- | Specify the maximum number of threads that can be spawned concurrently

103-- | Specify the maximum number of threads that can be spawned concurrently

107−-- when using concurrent streams. This values denotes maximum concurrent

104+-- when using concurrent streams. This value denotes maximum concurrent

108-- requests, or tasks in progress at any given point of time. Note that this is

105-- requests, or tasks in progress at any given point of time. Note that this is

109-- not the grand total number of threads but maximum threads at an individual

106-- not the grand total number of threads but maximum threads at an individual

110-- point of concurrency.

107-- point of concurrency.

test/MaxRate.hs+2−3

@@ -52,7 +52,7 @@ measureRate' desc t rate consumerDelay producerDelay dur = do

52 else randomRIO ( round $ toMicroSecs t1

53 , round $ toMicroSecs t2)

54 when (r > 0) $ do

55− t1 <- getTime Monotonic

55+ -- t1 <- getTime Monotonic

56 threadDelay r

57 -- t2 <- getTime Monotonic

58 -- let delta = fromIntegral (toNanoSecs (t2 - t1)) / 1000000000

@@ -87,7 +87,6 @@ main = hspec $ do

87 let rates = [1, 10, 100, 1000, 10000, 25000]

88 in describe "asyncly no consumer delay and 1 sec producer delay" $ do

89 forM_ rates (\r -> measureRate "asyncly" asyncly r 0 1 range)

90− -}

92 let rates = [1, 10, 100, 1000, 10000, 25000]

91 let rates = [1, 10, 100, 1000, 10000, 25000]

93 in describe "asyncly no consumer delay and variable producer delay" $ do

92 in describe "asyncly no consumer delay and variable producer delay" $ do

@@ -102,7 +101,7 @@ main = hspec $ do

102 in describe "wAsyncly no consumer delay and 1 sec producer delay" $ do

101 in describe "wAsyncly no consumer delay and 1 sec producer delay" $ do

103 forM_ rates (\r -> measureRate "wAsyncly" wAsyncly r 0 1 range)

102 forM_ rates (\r -> measureRate "wAsyncly" wAsyncly r 0 1 range)

104

103

105− let rates = [1, 10, 100, 1000, 10000, 100000]

104+ let rates = [1, 10, 100, 1000, 10000, 100000, 1000000]

106 in describe "aheadly no consumer delay no producer delay" $ do

105 in describe "aheadly no consumer delay no producer delay" $ do

107 forM_ rates (\r -> measureRate "aheadly" aheadly r 0 0 range)

106 forM_ rates (\r -> measureRate "aheadly" aheadly r 0 0 range)

108

107