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

store/tikv: coprocess streaming tiny fix #6186

Merged
merged 14 commits into from
Apr 4, 2018
Merged

store/tikv: coprocess streaming tiny fix #6186

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
0e1d1b6
[DNM] enable coprocess streaming for integrated testing
tiancaiamao Mar 29, 2018
0b4bad8
check range based on last response
tiancaiamao Mar 30, 2018
05cde22
handle the first response meet lock
tiancaiamao Mar 30, 2018
8a0bbd3
executor,store: make leak test more stable
tiancaiamao Mar 30, 2018
e67c48c
Merge branch 'test-leak' into testing
tiancaiamao Mar 30, 2018
5a594f9
Merge branch 'master' into testing
tiancaiamao Apr 2, 2018
47e0ed5
disable
tiancaiamao Apr 2, 2018
76813d8
Merge branch 'master' into testing
tiancaiamao Apr 2, 2018
6bac9eb
address comment
tiancaiamao Apr 3, 2018
877a944
address comment
tiancaiamao Apr 3, 2018
0f4cf65
address comment
tiancaiamao Apr 4, 2018
79f1c37
simplify code and test
tiancaiamao Apr 4, 2018
93448ba
Merge branch 'master' into testing
tiancaiamao Apr 4, 2018
d5eb49e
all-test-passed
tiancaiamao Apr 4, 2018
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
7 changes: 6 additions & 1 deletion store/tikv/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@
package tikv

import (
"io"
"strconv"
"sync"
"sync/atomic"
Expand All @@ -30,6 +31,7 @@ import (
"github.com/pingcap/tidb/metrics"
"github.com/pingcap/tidb/store/tikv/tikvrpc"
"github.com/pingcap/tidb/terror"
log "github.com/sirupsen/logrus"
"golang.org/x/net/context"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials"
Expand Down Expand Up @@ -254,7 +256,10 @@ func (c *rpcClient) SendRequest(ctx context.Context, addr string, req *tikvrpc.R
var first *coprocessor.Response
first, err = copStream.Recv()
if err != nil {
return nil, errors.Trace(err)
if errors.Cause(err) != io.EOF {
return nil, errors.Trace(err)
}
log.Debug("copstream returns nothing for the request.")
}
copStream.Response = first
return resp, nil
Expand Down
14 changes: 9 additions & 5 deletions store/tikv/coprocessor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -612,15 +612,19 @@ func (it *copIterator) handleTaskOnce(bo *Backoffer, task *copTask, ch chan copR
}

// Handles the response for non-streaming copTask.
return it.handleCopResponse(bo, resp.Cop, task, ch)
return it.handleCopResponse(bo, resp.Cop, task, ch, resp.Cop)
}

func (it *copIterator) handleCopStreamResult(bo *Backoffer, stream *tikvrpc.CopStreamResponse, task *copTask, ch chan copResponse) ([]*copTask, error) {
defer stream.Close()
var resp, lastResp *coprocessor.Response
resp = stream.Response
if resp == nil {
// streaming request returns io.EOF, so the first Response is nil.
return nil, nil
}
for {
remainedTasks, err := it.handleCopResponse(bo, resp, task, ch)
remainedTasks, err := it.handleCopResponse(bo, resp, task, ch, lastResp)
if err != nil || len(remainedTasks) != 0 {
return remainedTasks, errors.Trace(err)
}
Expand Down Expand Up @@ -652,7 +656,7 @@ func (it *copIterator) handleCopStreamResult(bo *Backoffer, stream *tikvrpc.CopS

// handleCopResponse checks coprocessor Response for region split and lock,
// returns more tasks when that happens, or handles the response if no error.
func (it *copIterator) handleCopResponse(bo *Backoffer, resp *coprocessor.Response, task *copTask, ch chan copResponse) ([]*copTask, error) {
func (it *copIterator) handleCopResponse(bo *Backoffer, resp *coprocessor.Response, task *copTask, ch chan copResponse, lastResp *coprocessor.Response) ([]*copTask, error) {
Copy link
Contributor

Choose a reason for hiding this comment

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

Could you add the comment for lastResp?

Copy link
Member

@coocood coocood Apr 3, 2018 •
edited
Loading

Choose a reason for hiding this comment

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

How about using lastRange *KeyRange instead of lastResp *coprocessor.Response?
It's easier to understand the purpose of the argument.

if regionErr := resp.GetRegionError(); regionErr != nil {
if err := bo.Backoff(BoRegionMiss, errors.New(regionErr.String())); err != nil {
return nil, errors.Trace(err)
Expand All @@ -672,7 +676,7 @@ func (it *copIterator) handleCopResponse(bo *Backoffer, resp *coprocessor.Respon
return nil, errors.Trace(err)
}
}
return buildCopTasksFromRemain(bo, it.store.regionCache, resp, task, it.req.Desc, it.req.Streaming)
return buildCopTasksFromRemain(bo, it.store.regionCache, lastResp, task, it.req.Desc, it.req.Streaming)
}
if otherErr := resp.GetOtherError(); otherErr != "" {
err := errors.Errorf("other error: %s", otherErr)
Expand All @@ -692,7 +696,7 @@ func (it *copIterator) handleCopResponse(bo *Backoffer, resp *coprocessor.Respon

func buildCopTasksFromRemain(bo *Backoffer, cache *RegionCache, resp *coprocessor.Response, task *copTask, desc bool, streaming bool) ([]*copTask, error) {
remainedRanges := task.ranges
if streaming {
if streaming && resp != nil {
remainedRanges = calculateRemain(task.ranges, resp.Range, desc)
}
return buildCopTasks(bo, cache, remainedRanges, desc, streaming)
Expand Down