git.iko.soy/streamly
on master help sign in
Simplify code in processWithToken
27d277cHarendra Kumar
src/Streamly/Streams/Ahead.hs+9−9
compare with master
unifiedsplit
@@ -399,7 +399,7 @@ processWithToken q heap st sv winfo action sno = do
399 -- XXX deduplicate stop in all invocations
399 -- XXX deduplicate stop in all invocations
400 let stop = do
400 let stop = do
401 liftIO (incrementYieldLimit sv)
401 liftIO (incrementYieldLimit sv)
402− loopWithToken sno
402+ loopWithToken (sno + 1)
403
403
404 unStream action st stop (singleOutput sno) (yieldOutput sno)
404 unStream action st stop (singleOutput sno) (yieldOutput sno)
405
405
@@ -408,7 +408,7 @@ processWithToken q heap st sv winfo action sno = do
408 singleOutput seqNo a = do
408 singleOutput seqNo a = do
409 continue <- liftIO $ sendYield sv winfo (ChildYield a)
409 continue <- liftIO $ sendYield sv winfo (ChildYield a)
410 if continue
410 if continue
411− then loopWithToken seqNo
411+ then loopWithToken (seqNo + 1)
412 else do
412 else do
413 liftIO $ updateHeapSeq heap (seqNo + 1)
413 liftIO $ updateHeapSeq heap (seqNo + 1)
414 drainHeap q heap st sv winfo
414 drainHeap q heap st sv winfo
@@ -423,7 +423,7 @@ processWithToken q heap st sv winfo action sno = do
423 then do
423 then do
424 let stop = do
424 let stop = do
425 liftIO (incrementYieldLimit sv)
425 liftIO (incrementYieldLimit sv)
426− loopWithToken seqNo
426+ loopWithToken (seqNo + 1)
427 unStream r st stop
427 unStream r st stop
428 (singleOutput seqNo)
428 (singleOutput seqNo)
429 (yieldOutput seqNo)
429 (yieldOutput seqNo)
@@ -433,27 +433,27 @@ processWithToken q heap st sv winfo action sno = do
433 liftIO $ incrementYieldLimit sv
433 liftIO $ incrementYieldLimit sv
434 drainHeap q heap st sv winfo
434 drainHeap q heap st sv winfo
435
435
436− loopWithToken prevSeqNo = do
436+ loopWithToken nextSeqNo = do
437 work <- dequeueAhead q
437 work <- dequeueAhead q
438 case work of
438 case work of
439 Nothing -> do
439 Nothing -> do
440− liftIO $ updateHeapSeq heap (prevSeqNo + 1)
440+ liftIO $ updateHeapSeq heap nextSeqNo
441 workLoopAhead q heap st sv winfo
441 workLoopAhead q heap st sv winfo
442
442
443 Just (m, seqNo) -> do
443 Just (m, seqNo) -> do
444 yieldLimitOk <- liftIO $ decrementYieldLimit sv
444 yieldLimitOk <- liftIO $ decrementYieldLimit sv
445 if yieldLimitOk
445 if yieldLimitOk
446 then
446 then
447− if seqNo == prevSeqNo + 1
447+ if seqNo == nextSeqNo
448 then do
448 then do
449 let stop = do
449 let stop = do
450 liftIO (incrementYieldLimit sv)
450 liftIO (incrementYieldLimit sv)
451− loopWithToken seqNo
451+ loopWithToken (seqNo + 1)
452 unStream m st stop
452 unStream m st stop
453 (singleOutput seqNo)
453 (singleOutput seqNo)
454 (yieldOutput seqNo)
454 (yieldOutput seqNo)
455 else do
455 else do
456− liftIO $ updateHeapSeq heap (prevSeqNo + 1)
456+ liftIO $ updateHeapSeq heap nextSeqNo
457 liftIO (incrementYieldLimit sv)
457 liftIO (incrementYieldLimit sv)
458 -- To avoid a race when another thread puts something
458 -- To avoid a race when another thread puts something
459 -- on the heap and goes away, the consumer will not get
459 -- on the heap and goes away, the consumer will not get
@@ -465,7 +465,7 @@ processWithToken q heap st sv winfo action sno = do
465 liftIO $ reEnqueueAhead sv q m
465 liftIO $ reEnqueueAhead sv q m
466 workLoopAhead q heap st sv winfo
466 workLoopAhead q heap st sv winfo
467 else do
467 else do
468− liftIO $ updateHeapSeq heap (prevSeqNo + 1)
468+ liftIO $ updateHeapSeq heap nextSeqNo
469 liftIO $ reEnqueueAhead sv q m
469 liftIO $ reEnqueueAhead sv q m
470 liftIO $ incrementYieldLimit sv
470 liftIO $ incrementYieldLimit sv
471 drainHeap q heap st sv winfo
471 drainHeap q heap st sv winfo