49542d2 · streamly · git.iko.soy

git.iko.soy/streamly

Fix ahead style nil stream merging live lock

commit49542d2445ae3955167c6fb3bfd7f82980581943 authorHarendra Kumar <harendra.kumar@gmail.com> dateSun 23 Sep 2018, 13:35 (8 years ago) parent27d277c

3 files changed+17−2

compare with master

unifiedsplit

Changelog.md+8 src/Streamly/SVar.hs+2−1 src/Streamly/Streams/Ahead.hs+7−1

Changelog.md+8

@@ -1,3 +1,11 @@

1+## Unreleased

3+### Bug Fixes

5+* Fixed a livelock in ahead style streams. The problem manifests sometimes when

6+ multiple streams are merged together in ahead style and one of them is a nil

7+ stream.

19## 0.5.1

210

311* Performance improvements, especially space consumption, for concurrent

src/Streamly/SVar.hs+2−1

@@ -189,7 +189,8 @@ data ChildEvent a =

189189

190190-- | Sorting out-of-turn outputs in a heap for Ahead style streams

191191data AheadHeapEntry (t :: (* -> *) -> * -> *) m a =

192− AheadEntryPure a

192+ AheadEntryNull

193+ | AheadEntryPure a

193194 | AheadEntryStream (t m a)

194195

195196------------------------------------------------------------------------------

src/Streamly/Streams/Ahead.hs+7−1

@@ -230,6 +230,7 @@ processHeap q heap st sv winfo entry sno stopping = loopHeap sno entry

230230

231231 loopHeap seqNo ent =

232232 case ent of

233+ AheadEntryNull -> nextHeap seqNo

233234 AheadEntryPure a -> do

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

235236 -- latency as this will not be the real latency.

@@ -338,7 +339,12 @@ processWithoutToken q heap st sv winfo m sno = do

338339 -- we have already decremented the yield limit for m

339340 let stop = do

340341 liftIO (incrementYieldLimit sv)

341− workLoopAhead q heap st sv winfo

342+ -- If the stream stops without yielding anything, and we do not put

343+ -- anything on heap, but if heap was waiting for this seq number

344+ -- then it will keep waiting forever, because we are never going to

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

346+ -- we stop.

347+ toHeap sno AheadEntryNull

342348

343349 unStream m st stop (singleToHeap sno) (yieldToHeap sno)

344350