[Issue 527] fix send goroutine blocked #530
Conversation
jiangbo9510
changed the title
| @@ -490,21 +490,36 @@ func (p *partitionProducer) failTimeoutMessages() { | |||
|
|
|||
| // iterate at most viewSize items | |||
| for i := 0; i < viewSize; i++ { | |||
| item := p.pendingQueue.Poll() | |||
| tickerNeedWaiting := time.Duration(0) | |||
| item := p.pendingQueue.CompareAndPoll( | |||
There was a problem hiding this comment.
This should also be used to refactor the pendingQueue.Peek() at line 464
There was a problem hiding this comment.
This should also be used to refactor the
pendingQueue.Peek()at line 464
@merlimat There is not necessarily. pendingQueue.Peek()` at line 464 is to check the first item of queue whether timeout. It not modify the queue.
There was a problem hiding this comment.
the bug reason is peek head item and determine if it timeout, and then get a snapshot of pendingQueue, range snapshot and poll all the items in snapshot even if it is not all timeout.
the pendingQueue.Peek() at line 464 just peek(not poll) the head item in queue and check(logically correct), there are no actual operation to pendingQueue, so it looks like no need to refactor
|
@wolfstudy When this pull request merge to the master branch? |
|
@merlimat please review my code and merge this pull request to master branch. |
|
cc / @cckellogg @merlimat PTAL |
|
@jiangbo9510 Please merger master code for failed action CI |
@wolfstudy |
|
@wolfstudy It's anothor problem. If it occured,All send goroutins will blocked by the dead lock. In my case, Sometimes some send goroutins blocked. and others goroutins can send message successful. |
Fix Issue: #527
Motivation
I add a function to compare and poll block queue in atomic. To fix poll a wrong item in
failTimeoutMessages()Modifications
This change is a trivial rework / code cleanup without any test coverage.
Does this pull request potentially affect one of the following parts:
Documentation