Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[Issue859] fix panic when waiting on left requests to drain upon connection close #860

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

[Issue859] fix panic when waiting on left requests to drain upon connection close #860

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9c2a61b
preemtively increament waitgroup for wait on left requests to drain
zzzming Oct 12, 2022
fdaf53c
remove waitgroup used to drain left incoming requests
zzzming Oct 16, 2022
1ae5cad
remove unrelated comments
zzzming Oct 16, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 0 additions & 11 deletions pulsar/internal/connection.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -144,7 +144,6 @@ type connection struct {

log log.Logger

incomingRequestsWG sync.WaitGroup
incomingRequestsCh chan *request
incomingCmdCh chan *incomingCmd
closeCh chan interface{}
Expand Down Expand Up @@ -347,10 +346,6 @@ func (c *connection) waitUntilReady() error {
}

func (c *connection) failLeftRequestsWhenClose() {
// wait for outstanding incoming requests to complete before draining
// and closing the channel
c.incomingRequestsWG.Wait()

ch := c.incomingRequestsCh
go func() {
// send a nil message to drain instead of
Expand Down Expand Up @@ -598,9 +593,6 @@ func (c *connection) Write(data Buffer) {

func (c *connection) SendRequest(requestID uint64, req *pb.BaseCommand,
callback func(command *pb.BaseCommand, err error)) {
c.incomingRequestsWG.Add(1)
defer c.incomingRequestsWG.Done()

state := c.getState()
if state == connectionClosed || state == connectionClosing {
callback(req, ErrConnectionClosed)
Expand All @@ -620,9 +612,6 @@ func (c *connection) SendRequest(requestID uint64, req *pb.BaseCommand,
}

func (c *connection) SendRequestNoWait(req *pb.BaseCommand) error {
c.incomingRequestsWG.Add(1)
defer c.incomingRequestsWG.Done()

state := c.getState()
if state == connectionClosed || state == connectionClosing {
return ErrConnectionClosed
Expand Down