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

distsql, executor: fix select channel bug that may discard error. #2226

Merged
merged 3 commits into from
Dec 12, 2016
Merged

distsql, executor: fix select channel bug that may discard error. #2226

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
281c88b
distsql, executor: fix race bug that may discard error.
coocood Dec 12, 2016
2ed3af3
*: update test
coocood Dec 12, 2016
4f2ea55
Merge branch 'master' into coocood/select-bug
coocood Dec 12, 2016
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
32 changes: 12 additions & 20 deletions distsql/distsql.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,10 +72,13 @@ type selectResult struct {
resp kv.Response
ignoreData bool

results chan PartialResult
done chan error
results chan resultWithErr
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I sugguest *resultWithErr...
use chan for value instead of ptr is heavy

closed chan struct{}
}

closed chan struct{}
type resultWithErr struct {
result PartialResult
err error
}

func (r *selectResult) Fetch() {
Expand All @@ -87,7 +90,7 @@ func (r *selectResult) fetch() {
for {
reader, err := r.resp.Next()
if err != nil {
r.done <- errors.Trace(err)
r.results <- resultWithErr{err: errors.Trace(err)}
return
}
if reader == nil {
Expand All @@ -104,7 +107,7 @@ func (r *selectResult) fetch() {
go pr.fetch()

select {
case r.results <- pr:
case r.results <- resultWithErr{result: pr}:
case <-r.closed:
// if selectResult called Close() already, make fetch goroutine exit
return
Expand All @@ -113,19 +116,9 @@ func (r *selectResult) fetch() {
}

// Next returns the next row.
func (r *selectResult) Next() (pr PartialResult, err error) {
var ok bool
select {
case pr, ok = <-r.results:
case err = <-r.done:
}
if err != nil {
return nil, err
}
if !ok {
return nil, nil
}
return
func (r *selectResult) Next() (PartialResult, error) {
re := <-r.results
return re.result, errors.Trace(re.err)
}

// SetFields sets select result field types.
Expand Down Expand Up @@ -301,8 +294,7 @@ func Select(client kv.Client, req *tipb.SelectRequest, keyRanges []kv.KeyRange,
}
result := &selectResult{
resp: resp,
results: make(chan PartialResult, 5),
done: make(chan error, 1),
results: make(chan resultWithErr, 5),
closed: make(chan struct{}),
}
// If Aggregates is not nil, we should set result fields latter.
Expand Down
3 changes: 1 addition & 2 deletions distsql/distsql_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,8 +59,7 @@ func (s *testTableCodecSuite) TestGoroutineLeak(c *C) {

sr = &selectResult{
resp: &mockResponse{},
results: make(chan PartialResult, 5),
done: make(chan error, 1),
results: make(chan resultWithErr, 5),
closed: make(chan struct{}),
}
go sr.Fetch()
Expand Down