You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
1434 lines
42 KiB
1434 lines
42 KiB
// Copyright 2016 PingCAP, Inc.
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package tikv
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
"unsafe"
|
|
|
|
"github.com/cznic/mathutil"
|
|
"github.com/gogo/protobuf/proto"
|
|
"github.com/pingcap/errors"
|
|
"github.com/pingcap/failpoint"
|
|
"github.com/pingcap/kvproto/pkg/coprocessor"
|
|
"github.com/pingcap/kvproto/pkg/kvrpcpb"
|
|
"github.com/pingcap/parser/terror"
|
|
"github.com/pingcap/tidb/domain/infosync"
|
|
"github.com/pingcap/tidb/errno"
|
|
"github.com/pingcap/tidb/kv"
|
|
"github.com/pingcap/tidb/metrics"
|
|
"github.com/pingcap/tidb/store/tikv/tikvrpc"
|
|
"github.com/pingcap/tidb/util/execdetails"
|
|
"github.com/pingcap/tidb/util/logutil"
|
|
"github.com/pingcap/tidb/util/memory"
|
|
"github.com/pingcap/tipb/go-tipb"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
var (
|
|
tikvTxnRegionsNumHistogramWithCoprocessor = metrics.TiKVTxnRegionsNumHistogram.WithLabelValues("coprocessor")
|
|
tikvTxnRegionsNumHistogramWithBatchCoprocessor = metrics.TiKVTxnRegionsNumHistogram.WithLabelValues("batch_coprocessor")
|
|
coprCacheHistogramEvict = metrics.DistSQLCoprCacheHistogram.WithLabelValues("evict")
|
|
)
|
|
|
|
// CopClient is coprocessor client.
|
|
type CopClient struct {
|
|
kv.RequestTypeSupportedChecker
|
|
store *tikvStore
|
|
replicaReadSeed uint32
|
|
}
|
|
|
|
// Send builds the request and gets the coprocessor iterator response.
|
|
func (c *CopClient) Send(ctx context.Context, req *kv.Request, vars *kv.Variables, sessionMemTracker *memory.Tracker, enabledRateLimitAction bool) kv.Response {
|
|
if req.StoreType == kv.TiFlash && req.BatchCop {
|
|
logutil.BgLogger().Debug("send batch requests")
|
|
return c.sendBatch(ctx, req, vars)
|
|
}
|
|
ctx = context.WithValue(ctx, txnStartKey, req.StartTs)
|
|
bo := NewBackofferWithVars(ctx, copBuildTaskMaxBackoff, vars)
|
|
tasks, err := buildCopTasks(bo, c.store.regionCache, &copRanges{mid: req.KeyRanges}, req)
|
|
if err != nil {
|
|
return copErrorResponse{err}
|
|
}
|
|
it := &copIterator{
|
|
store: c.store,
|
|
req: req,
|
|
concurrency: req.Concurrency,
|
|
finishCh: make(chan struct{}),
|
|
vars: vars,
|
|
memTracker: req.MemTracker,
|
|
replicaReadSeed: c.replicaReadSeed,
|
|
rpcCancel: NewRPCanceller(),
|
|
}
|
|
it.minCommitTSPushed.data = make(map[uint64]struct{}, 5)
|
|
it.tasks = tasks
|
|
if it.concurrency > len(tasks) {
|
|
it.concurrency = len(tasks)
|
|
}
|
|
if it.concurrency < 1 {
|
|
// Make sure that there is at least one worker.
|
|
it.concurrency = 1
|
|
}
|
|
|
|
if it.req.KeepOrder {
|
|
it.sendRate = newRateLimit(2 * it.concurrency)
|
|
it.respChan = nil
|
|
} else {
|
|
capacity := it.concurrency
|
|
if enabledRateLimitAction {
|
|
// The count of cached response in memory is controlled by the capacity of the it.sendRate, not capacity of the respChan.
|
|
// As the worker will send finCopResponse after each task being handled, we make the capacity of the respCh equals to
|
|
// 2*it.concurrency to avoid deadlock in the unit test caused by the `MustExec` or `Exec`
|
|
capacity = it.concurrency * 2
|
|
}
|
|
it.respChan = make(chan *copResponse, capacity)
|
|
it.sendRate = newRateLimit(it.concurrency)
|
|
}
|
|
it.actionOnExceed = newRateLimitAction(uint(cap(it.sendRate.token)))
|
|
if sessionMemTracker != nil {
|
|
sessionMemTracker.FallbackOldAndSetNewAction(it.actionOnExceed)
|
|
}
|
|
|
|
if !it.req.Streaming {
|
|
ctx = context.WithValue(ctx, RPCCancellerCtxKey{}, it.rpcCancel)
|
|
}
|
|
it.open(ctx, enabledRateLimitAction)
|
|
return it
|
|
}
|
|
|
|
// copTask contains a related Region and KeyRange for a kv.Request.
|
|
type copTask struct {
|
|
region RegionVerID
|
|
ranges *copRanges
|
|
|
|
respChan chan *copResponse
|
|
storeAddr string
|
|
cmdType tikvrpc.CmdType
|
|
storeType kv.StoreType
|
|
}
|
|
|
|
func (r *copTask) String() string {
|
|
return fmt.Sprintf("region(%d %d %d) ranges(%d) store(%s)",
|
|
r.region.id, r.region.confVer, r.region.ver, r.ranges.len(), r.storeAddr)
|
|
}
|
|
|
|
// copRanges is like []kv.KeyRange, but may has extra elements at head/tail.
|
|
// It's for avoiding alloc big slice during build copTask.
|
|
type copRanges struct {
|
|
first *kv.KeyRange
|
|
mid []kv.KeyRange
|
|
last *kv.KeyRange
|
|
}
|
|
|
|
func (r *copRanges) String() string {
|
|
var s string
|
|
r.do(func(ran *kv.KeyRange) {
|
|
s += fmt.Sprintf("[%q, %q]", ran.StartKey, ran.EndKey)
|
|
})
|
|
return s
|
|
}
|
|
|
|
func (r *copRanges) len() int {
|
|
var l int
|
|
if r.first != nil {
|
|
l++
|
|
}
|
|
l += len(r.mid)
|
|
if r.last != nil {
|
|
l++
|
|
}
|
|
return l
|
|
}
|
|
|
|
func (r *copRanges) at(i int) kv.KeyRange {
|
|
if r.first != nil {
|
|
if i == 0 {
|
|
return *r.first
|
|
}
|
|
i--
|
|
}
|
|
if i < len(r.mid) {
|
|
return r.mid[i]
|
|
}
|
|
return *r.last
|
|
}
|
|
|
|
func (r *copRanges) slice(from, to int) *copRanges {
|
|
var ran copRanges
|
|
if r.first != nil {
|
|
if from == 0 && to > 0 {
|
|
ran.first = r.first
|
|
}
|
|
if from > 0 {
|
|
from--
|
|
}
|
|
if to > 0 {
|
|
to--
|
|
}
|
|
}
|
|
if to <= len(r.mid) {
|
|
ran.mid = r.mid[from:to]
|
|
} else {
|
|
if from <= len(r.mid) {
|
|
ran.mid = r.mid[from:]
|
|
}
|
|
if from < to {
|
|
ran.last = r.last
|
|
}
|
|
}
|
|
return &ran
|
|
}
|
|
|
|
func (r *copRanges) do(f func(ran *kv.KeyRange)) {
|
|
if r.first != nil {
|
|
f(r.first)
|
|
}
|
|
for _, ran := range r.mid {
|
|
f(&ran)
|
|
}
|
|
if r.last != nil {
|
|
f(r.last)
|
|
}
|
|
}
|
|
|
|
func (r *copRanges) toPBRanges() []*coprocessor.KeyRange {
|
|
ranges := make([]*coprocessor.KeyRange, 0, r.len())
|
|
r.do(func(ran *kv.KeyRange) {
|
|
ranges = append(ranges, &coprocessor.KeyRange{
|
|
Start: ran.StartKey,
|
|
End: ran.EndKey,
|
|
})
|
|
})
|
|
return ranges
|
|
}
|
|
|
|
// split ranges into (left, right) by key.
|
|
func (r *copRanges) split(key []byte) (*copRanges, *copRanges) {
|
|
n := sort.Search(r.len(), func(i int) bool {
|
|
cur := r.at(i)
|
|
return len(cur.EndKey) == 0 || bytes.Compare(cur.EndKey, key) > 0
|
|
})
|
|
// If a range p contains the key, it will split to 2 parts.
|
|
if n < r.len() {
|
|
p := r.at(n)
|
|
if bytes.Compare(key, p.StartKey) > 0 {
|
|
left := r.slice(0, n)
|
|
left.last = &kv.KeyRange{StartKey: p.StartKey, EndKey: key}
|
|
right := r.slice(n+1, r.len())
|
|
right.first = &kv.KeyRange{StartKey: key, EndKey: p.EndKey}
|
|
return left, right
|
|
}
|
|
}
|
|
return r.slice(0, n), r.slice(n, r.len())
|
|
}
|
|
|
|
// rangesPerTask limits the length of the ranges slice sent in one copTask.
|
|
const rangesPerTask = 25000
|
|
|
|
func buildCopTasks(bo *Backoffer, cache *RegionCache, ranges *copRanges, req *kv.Request) ([]*copTask, error) {
|
|
start := time.Now()
|
|
cmdType := tikvrpc.CmdCop
|
|
if req.Streaming {
|
|
cmdType = tikvrpc.CmdCopStream
|
|
}
|
|
|
|
if req.StoreType == kv.TiDB {
|
|
return buildTiDBMemCopTasks(ranges, req)
|
|
}
|
|
|
|
rangesLen := ranges.len()
|
|
var tasks []*copTask
|
|
appendTask := func(regionWithRangeInfo *KeyLocation, ranges *copRanges) {
|
|
// TiKV will return gRPC error if the message is too large. So we need to limit the length of the ranges slice
|
|
// to make sure the message can be sent successfully.
|
|
rLen := ranges.len()
|
|
for i := 0; i < rLen; {
|
|
nextI := mathutil.Min(i+rangesPerTask, rLen)
|
|
tasks = append(tasks, &copTask{
|
|
region: regionWithRangeInfo.Region,
|
|
ranges: ranges.slice(i, nextI),
|
|
// Channel buffer is 2 for handling region split.
|
|
// In a common case, two region split tasks will not be blocked.
|
|
respChan: make(chan *copResponse, 2),
|
|
cmdType: cmdType,
|
|
storeType: req.StoreType,
|
|
})
|
|
i = nextI
|
|
}
|
|
}
|
|
|
|
err := splitRanges(bo, cache, ranges, appendTask)
|
|
if err != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
|
|
if req.Desc {
|
|
reverseTasks(tasks)
|
|
}
|
|
if elapsed := time.Since(start); elapsed > time.Millisecond*500 {
|
|
logutil.BgLogger().Warn("buildCopTasks takes too much time",
|
|
zap.Duration("elapsed", elapsed),
|
|
zap.Int("range len", rangesLen),
|
|
zap.Int("task len", len(tasks)))
|
|
}
|
|
tikvTxnRegionsNumHistogramWithCoprocessor.Observe(float64(len(tasks)))
|
|
return tasks, nil
|
|
}
|
|
|
|
func buildTiDBMemCopTasks(ranges *copRanges, req *kv.Request) ([]*copTask, error) {
|
|
servers, err := infosync.GetAllServerInfo(context.Background())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
cmdType := tikvrpc.CmdCop
|
|
if req.Streaming {
|
|
cmdType = tikvrpc.CmdCopStream
|
|
}
|
|
tasks := make([]*copTask, 0, len(servers))
|
|
for _, ser := range servers {
|
|
addr := ser.IP + ":" + strconv.FormatUint(uint64(ser.StatusPort), 10)
|
|
tasks = append(tasks, &copTask{
|
|
ranges: ranges,
|
|
respChan: make(chan *copResponse, 2),
|
|
cmdType: cmdType,
|
|
storeType: req.StoreType,
|
|
storeAddr: addr,
|
|
})
|
|
}
|
|
return tasks, nil
|
|
}
|
|
|
|
func splitRanges(bo *Backoffer, cache *RegionCache, ranges *copRanges, fn func(regionWithRangeInfo *KeyLocation, ranges *copRanges)) error {
|
|
for ranges.len() > 0 {
|
|
loc, err := cache.LocateKey(bo, ranges.at(0).StartKey)
|
|
if err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
|
|
// Iterate to the first range that is not complete in the region.
|
|
var i int
|
|
for ; i < ranges.len(); i++ {
|
|
r := ranges.at(i)
|
|
if !(loc.Contains(r.EndKey) || bytes.Equal(loc.EndKey, r.EndKey)) {
|
|
break
|
|
}
|
|
}
|
|
// All rest ranges belong to the same region.
|
|
if i == ranges.len() {
|
|
fn(loc, ranges)
|
|
break
|
|
}
|
|
|
|
r := ranges.at(i)
|
|
if loc.Contains(r.StartKey) {
|
|
// Part of r is not in the region. We need to split it.
|
|
taskRanges := ranges.slice(0, i)
|
|
taskRanges.last = &kv.KeyRange{
|
|
StartKey: r.StartKey,
|
|
EndKey: loc.EndKey,
|
|
}
|
|
fn(loc, taskRanges)
|
|
|
|
ranges = ranges.slice(i+1, ranges.len())
|
|
ranges.first = &kv.KeyRange{
|
|
StartKey: loc.EndKey,
|
|
EndKey: r.EndKey,
|
|
}
|
|
} else {
|
|
// rs[i] is not in the region.
|
|
taskRanges := ranges.slice(0, i)
|
|
fn(loc, taskRanges)
|
|
ranges = ranges.slice(i, ranges.len())
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// SplitRegionRanges get the split ranges from pd region.
|
|
func SplitRegionRanges(bo *Backoffer, cache *RegionCache, keyRanges []kv.KeyRange) ([]kv.KeyRange, error) {
|
|
ranges := copRanges{mid: keyRanges}
|
|
|
|
var ret []kv.KeyRange
|
|
appendRange := func(regionWithRangeInfo *KeyLocation, ranges *copRanges) {
|
|
for i := 0; i < ranges.len(); i++ {
|
|
ret = append(ret, ranges.at(i))
|
|
}
|
|
}
|
|
|
|
err := splitRanges(bo, cache, &ranges, appendRange)
|
|
if err != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
return ret, nil
|
|
}
|
|
|
|
func reverseTasks(tasks []*copTask) {
|
|
for i := 0; i < len(tasks)/2; i++ {
|
|
j := len(tasks) - i - 1
|
|
tasks[i], tasks[j] = tasks[j], tasks[i]
|
|
}
|
|
}
|
|
|
|
type copIterator struct {
|
|
store *tikvStore
|
|
req *kv.Request
|
|
concurrency int
|
|
finishCh chan struct{}
|
|
|
|
// If keepOrder, results are stored in copTask.respChan, read them out one by one.
|
|
tasks []*copTask
|
|
// curr indicates the curr id of the finished copTask
|
|
curr int
|
|
|
|
// sendRate controls the sending rate of copIteratorTaskSender
|
|
sendRate *rateLimit
|
|
|
|
// Otherwise, results are stored in respChan.
|
|
respChan chan *copResponse
|
|
|
|
vars *kv.Variables
|
|
|
|
memTracker *memory.Tracker
|
|
|
|
replicaReadSeed uint32
|
|
|
|
rpcCancel *RPCCanceller
|
|
|
|
wg sync.WaitGroup
|
|
// closed represents when the Close is called.
|
|
// There are two cases we need to close the `finishCh` channel, one is when context is done, the other one is
|
|
// when the Close is called. we use atomic.CompareAndSwap `closed` to to make sure the channel is not closed twice.
|
|
closed uint32
|
|
|
|
minCommitTSPushed
|
|
|
|
actionOnExceed *rateLimitAction
|
|
}
|
|
|
|
// copIteratorWorker receives tasks from copIteratorTaskSender, handles tasks and sends the copResponse to respChan.
|
|
type copIteratorWorker struct {
|
|
taskCh <-chan *copTask
|
|
wg *sync.WaitGroup
|
|
store *tikvStore
|
|
req *kv.Request
|
|
respChan chan<- *copResponse
|
|
finishCh <-chan struct{}
|
|
vars *kv.Variables
|
|
clientHelper
|
|
|
|
memTracker *memory.Tracker
|
|
|
|
replicaReadSeed uint32
|
|
|
|
actionOnExceed *rateLimitAction
|
|
}
|
|
|
|
// copIteratorTaskSender sends tasks to taskCh then wait for the workers to exit.
|
|
type copIteratorTaskSender struct {
|
|
taskCh chan<- *copTask
|
|
wg *sync.WaitGroup
|
|
tasks []*copTask
|
|
finishCh <-chan struct{}
|
|
respChan chan<- *copResponse
|
|
sendRate *rateLimit
|
|
}
|
|
|
|
type copResponse struct {
|
|
pbResp *coprocessor.Response
|
|
detail *CopRuntimeStats
|
|
startKey kv.Key
|
|
err error
|
|
respSize int64
|
|
respTime time.Duration
|
|
}
|
|
|
|
const (
|
|
sizeofExecDetails = int(unsafe.Sizeof(execdetails.ExecDetails{}))
|
|
sizeofCommitDetails = int(unsafe.Sizeof(execdetails.CommitDetails{}))
|
|
)
|
|
|
|
// GetData implements the kv.ResultSubset GetData interface.
|
|
func (rs *copResponse) GetData() []byte {
|
|
return rs.pbResp.Data
|
|
}
|
|
|
|
// GetStartKey implements the kv.ResultSubset GetStartKey interface.
|
|
func (rs *copResponse) GetStartKey() kv.Key {
|
|
return rs.startKey
|
|
}
|
|
|
|
func (rs *copResponse) GetCopRuntimeStats() *CopRuntimeStats {
|
|
return rs.detail
|
|
}
|
|
|
|
// MemSize returns how many bytes of memory this response use
|
|
func (rs *copResponse) MemSize() int64 {
|
|
if rs.respSize != 0 {
|
|
return rs.respSize
|
|
}
|
|
if rs == finCopResp {
|
|
return 0
|
|
}
|
|
|
|
// ignore rs.err
|
|
rs.respSize += int64(cap(rs.startKey))
|
|
if rs.detail != nil {
|
|
rs.respSize += int64(sizeofExecDetails)
|
|
}
|
|
if rs.pbResp != nil {
|
|
// Using a approximate size since it's hard to get a accurate value.
|
|
rs.respSize += int64(rs.pbResp.Size())
|
|
}
|
|
return rs.respSize
|
|
}
|
|
|
|
func (rs *copResponse) RespTime() time.Duration {
|
|
return rs.respTime
|
|
}
|
|
|
|
const minLogCopTaskTime = 300 * time.Millisecond
|
|
|
|
// When the worker finished `handleTask`, we need to notify the copIterator that there is one task finished.
|
|
// For the non-keep-order case, we send a finCopResp into the respCh after `handleTask`. When copIterator recv
|
|
// finCopResp from the respCh, it will be aware that there is one task finished.
|
|
var finCopResp *copResponse
|
|
|
|
func init() {
|
|
finCopResp = &copResponse{}
|
|
}
|
|
|
|
// run is a worker function that get a copTask from channel, handle it and
|
|
// send the result back.
|
|
func (worker *copIteratorWorker) run(ctx context.Context) {
|
|
defer func() {
|
|
failpoint.Inject("ticase-4169", func(val failpoint.Value) {
|
|
if val.(bool) {
|
|
worker.memTracker.Consume(10 * MockResponseSizeForTest)
|
|
worker.memTracker.Consume(10 * MockResponseSizeForTest)
|
|
}
|
|
})
|
|
worker.wg.Done()
|
|
}()
|
|
for task := range worker.taskCh {
|
|
respCh := worker.respChan
|
|
if respCh == nil {
|
|
respCh = task.respChan
|
|
}
|
|
worker.handleTask(ctx, task, respCh)
|
|
if worker.respChan != nil {
|
|
// When a task is finished by the worker, send a finCopResp into channel to notify the copIterator that
|
|
// there is a task finished.
|
|
worker.sendToRespCh(finCopResp, worker.respChan, false)
|
|
}
|
|
close(task.respChan)
|
|
if worker.vars != nil && worker.vars.Killed != nil && atomic.LoadUint32(worker.vars.Killed) == 1 {
|
|
return
|
|
}
|
|
select {
|
|
case <-worker.finishCh:
|
|
return
|
|
default:
|
|
}
|
|
}
|
|
}
|
|
|
|
// open starts workers and sender goroutines.
|
|
func (it *copIterator) open(ctx context.Context, enabledRateLimitAction bool) {
|
|
taskCh := make(chan *copTask, 1)
|
|
it.wg.Add(it.concurrency)
|
|
// Start it.concurrency number of workers to handle cop requests.
|
|
for i := 0; i < it.concurrency; i++ {
|
|
worker := &copIteratorWorker{
|
|
taskCh: taskCh,
|
|
wg: &it.wg,
|
|
store: it.store,
|
|
req: it.req,
|
|
respChan: it.respChan,
|
|
finishCh: it.finishCh,
|
|
vars: it.vars,
|
|
clientHelper: clientHelper{
|
|
LockResolver: it.store.lockResolver,
|
|
RegionCache: it.store.regionCache,
|
|
minCommitTSPushed: &it.minCommitTSPushed,
|
|
Client: it.store.client,
|
|
},
|
|
memTracker: it.memTracker,
|
|
replicaReadSeed: it.replicaReadSeed,
|
|
actionOnExceed: it.actionOnExceed,
|
|
}
|
|
go worker.run(ctx)
|
|
}
|
|
taskSender := &copIteratorTaskSender{
|
|
taskCh: taskCh,
|
|
wg: &it.wg,
|
|
tasks: it.tasks,
|
|
finishCh: it.finishCh,
|
|
sendRate: it.sendRate,
|
|
}
|
|
taskSender.respChan = it.respChan
|
|
// enabledRateLimit decides whether enabled ratelimit action
|
|
it.actionOnExceed.setEnabled(enabledRateLimitAction)
|
|
failpoint.Inject("ticase-4171", func(val failpoint.Value) {
|
|
if val.(bool) {
|
|
it.memTracker.Consume(10 * MockResponseSizeForTest)
|
|
it.memTracker.Consume(10 * MockResponseSizeForTest)
|
|
}
|
|
})
|
|
go taskSender.run()
|
|
}
|
|
|
|
func (sender *copIteratorTaskSender) run() {
|
|
// Send tasks to feed the worker goroutines.
|
|
for _, t := range sender.tasks {
|
|
// we control the sending rate to prevent all tasks
|
|
// being done (aka. all of the responses are buffered) by copIteratorWorker.
|
|
// We keep the number of inflight tasks within the number of 2 * concurrency when Keep Order is true.
|
|
// If KeepOrder is false, the number equals the concurrency.
|
|
// It sends one more task if a task has been finished in copIterator.Next.
|
|
exit := sender.sendRate.getToken(sender.finishCh)
|
|
if exit {
|
|
break
|
|
}
|
|
exit = sender.sendToTaskCh(t)
|
|
if exit {
|
|
break
|
|
}
|
|
}
|
|
close(sender.taskCh)
|
|
|
|
// Wait for worker goroutines to exit.
|
|
sender.wg.Wait()
|
|
if sender.respChan != nil {
|
|
close(sender.respChan)
|
|
}
|
|
}
|
|
|
|
func (it *copIterator) recvFromRespCh(ctx context.Context, respCh <-chan *copResponse) (resp *copResponse, ok bool, exit bool) {
|
|
ticker := time.NewTicker(3 * time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case resp, ok = <-respCh:
|
|
if it.memTracker != nil && resp != nil {
|
|
consumed := resp.MemSize()
|
|
failpoint.Inject("testRateLimitActionMockConsumeAndAssert", func(val failpoint.Value) {
|
|
if val.(bool) {
|
|
if resp != finCopResp {
|
|
consumed = MockResponseSizeForTest
|
|
}
|
|
}
|
|
})
|
|
it.memTracker.Consume(-consumed)
|
|
}
|
|
return
|
|
case <-it.finishCh:
|
|
exit = true
|
|
return
|
|
case <-ticker.C:
|
|
if atomic.LoadUint32(it.vars.Killed) == 1 {
|
|
resp = &copResponse{err: ErrQueryInterrupted}
|
|
ok = true
|
|
return
|
|
}
|
|
case <-ctx.Done():
|
|
// We select the ctx.Done() in the thread of `Next` instead of in the worker to avoid the cost of `WithCancel`.
|
|
if atomic.CompareAndSwapUint32(&it.closed, 0, 1) {
|
|
close(it.finishCh)
|
|
}
|
|
exit = true
|
|
return
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
func (sender *copIteratorTaskSender) sendToTaskCh(t *copTask) (exit bool) {
|
|
select {
|
|
case sender.taskCh <- t:
|
|
case <-sender.finishCh:
|
|
exit = true
|
|
}
|
|
return
|
|
}
|
|
|
|
func (worker *copIteratorWorker) sendToRespCh(resp *copResponse, respCh chan<- *copResponse, checkOOM bool) (exit bool) {
|
|
if worker.memTracker != nil && checkOOM {
|
|
consumed := resp.MemSize()
|
|
failpoint.Inject("testRateLimitActionMockConsumeAndAssert", func(val failpoint.Value) {
|
|
if val.(bool) {
|
|
if resp != finCopResp {
|
|
consumed = MockResponseSizeForTest
|
|
}
|
|
}
|
|
})
|
|
worker.memTracker.Consume(consumed)
|
|
}
|
|
select {
|
|
case respCh <- resp:
|
|
case <-worker.finishCh:
|
|
exit = true
|
|
}
|
|
return
|
|
}
|
|
|
|
// MockResponseSizeForTest mock the response size
|
|
const MockResponseSizeForTest = 100 * 1024 * 1024
|
|
|
|
// Next returns next coprocessor result.
|
|
// NOTE: Use nil to indicate finish, so if the returned ResultSubset is not nil, reader should continue to call Next().
|
|
func (it *copIterator) Next(ctx context.Context) (kv.ResultSubset, error) {
|
|
var (
|
|
resp *copResponse
|
|
ok bool
|
|
closed bool
|
|
)
|
|
defer func() {
|
|
if resp == nil {
|
|
failpoint.Inject("ticase-4170", func(val failpoint.Value) {
|
|
if val.(bool) {
|
|
it.memTracker.Consume(10 * MockResponseSizeForTest)
|
|
it.memTracker.Consume(10 * MockResponseSizeForTest)
|
|
}
|
|
})
|
|
}
|
|
}()
|
|
// wait unit at least 5 copResponse received.
|
|
failpoint.Inject("testRateLimitActionMockWaitMax", func(val failpoint.Value) {
|
|
if val.(bool) {
|
|
// we only need to trigger oom at least once.
|
|
if len(it.tasks) > 9 {
|
|
for it.memTracker.MaxConsumed() < 5*MockResponseSizeForTest {
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
}
|
|
}
|
|
})
|
|
// If data order matters, response should be returned in the same order as copTask slice.
|
|
// Otherwise all responses are returned from a single channel.
|
|
if it.respChan != nil {
|
|
// Get next fetched resp from chan
|
|
resp, ok, closed = it.recvFromRespCh(ctx, it.respChan)
|
|
if !ok || closed {
|
|
it.actionOnExceed.close()
|
|
return nil, nil
|
|
}
|
|
if resp == finCopResp {
|
|
it.actionOnExceed.destroyTokenIfNeeded(func() {
|
|
it.sendRate.putToken()
|
|
})
|
|
return it.Next(ctx)
|
|
}
|
|
} else {
|
|
for {
|
|
if it.curr >= len(it.tasks) {
|
|
// Resp will be nil if iterator is finishCh.
|
|
it.actionOnExceed.close()
|
|
return nil, nil
|
|
}
|
|
task := it.tasks[it.curr]
|
|
resp, ok, closed = it.recvFromRespCh(ctx, task.respChan)
|
|
if closed {
|
|
// Close() is already called, so Next() is invalid.
|
|
return nil, nil
|
|
}
|
|
if ok {
|
|
break
|
|
}
|
|
it.actionOnExceed.destroyTokenIfNeeded(func() {
|
|
it.sendRate.putToken()
|
|
})
|
|
// Switch to next task.
|
|
it.tasks[it.curr] = nil
|
|
it.curr++
|
|
}
|
|
}
|
|
|
|
if resp.err != nil {
|
|
return nil, errors.Trace(resp.err)
|
|
}
|
|
|
|
err := it.store.CheckVisibility(it.req.StartTs)
|
|
if err != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
return resp, nil
|
|
}
|
|
|
|
// Associate each region with an independent backoffer. In this way, when multiple regions are
|
|
// unavailable, TiDB can execute very quickly without blocking
|
|
func chooseBackoffer(ctx context.Context, backoffermap map[uint64]*Backoffer, task *copTask, worker *copIteratorWorker) *Backoffer {
|
|
bo, ok := backoffermap[task.region.id]
|
|
if ok {
|
|
return bo
|
|
}
|
|
newbo := NewBackofferWithVars(ctx, copNextMaxBackoff, worker.vars)
|
|
backoffermap[task.region.id] = newbo
|
|
return newbo
|
|
}
|
|
|
|
// handleTask handles single copTask, sends the result to channel, retry automatically on error.
|
|
func (worker *copIteratorWorker) handleTask(ctx context.Context, task *copTask, respCh chan<- *copResponse) {
|
|
defer func() {
|
|
r := recover()
|
|
if r != nil {
|
|
logutil.BgLogger().Error("copIteratorWork meet panic",
|
|
zap.Reflect("r", r),
|
|
zap.Stack("stack trace"))
|
|
resp := &copResponse{err: errors.Errorf("%v", r)}
|
|
// if panic has happened, set checkOOM to false to avoid another panic.
|
|
worker.sendToRespCh(resp, respCh, false)
|
|
}
|
|
}()
|
|
remainTasks := []*copTask{task}
|
|
backoffermap := make(map[uint64]*Backoffer)
|
|
for len(remainTasks) > 0 {
|
|
curTask := remainTasks[0]
|
|
bo := chooseBackoffer(ctx, backoffermap, curTask, worker)
|
|
tasks, err := worker.handleTaskOnce(bo, curTask, respCh)
|
|
if err != nil {
|
|
resp := &copResponse{err: errors.Trace(err)}
|
|
worker.sendToRespCh(resp, respCh, true)
|
|
return
|
|
}
|
|
// test whether the ctx is cancelled
|
|
if bo.vars != nil && bo.vars.Killed != nil && atomic.LoadUint32(bo.vars.Killed) == 1 {
|
|
return
|
|
}
|
|
|
|
if len(tasks) > 0 {
|
|
remainTasks = append(tasks, remainTasks[1:]...)
|
|
} else {
|
|
remainTasks = remainTasks[1:]
|
|
}
|
|
}
|
|
if worker.store.coprCache != nil && worker.store.coprCache.cache.Metrics != nil {
|
|
coprCacheHistogramEvict.Observe(float64(worker.store.coprCache.cache.Metrics.KeysEvicted()))
|
|
}
|
|
}
|
|
|
|
// handleTaskOnce handles single copTask, successful results are send to channel.
|
|
// If error happened, returns error. If region split or meet lock, returns the remain tasks.
|
|
func (worker *copIteratorWorker) handleTaskOnce(bo *Backoffer, task *copTask, ch chan<- *copResponse) ([]*copTask, error) {
|
|
failpoint.Inject("handleTaskOnceError", func(val failpoint.Value) {
|
|
if val.(bool) {
|
|
failpoint.Return(nil, errors.New("mock handleTaskOnce error"))
|
|
}
|
|
})
|
|
|
|
copReq := coprocessor.Request{
|
|
Tp: worker.req.Tp,
|
|
StartTs: worker.req.StartTs,
|
|
Data: worker.req.Data,
|
|
Ranges: task.ranges.toPBRanges(),
|
|
SchemaVer: worker.req.SchemaVar,
|
|
}
|
|
|
|
var cacheKey []byte = nil
|
|
var cacheValue *coprCacheValue = nil
|
|
|
|
// If there are many ranges, it is very likely to be a TableLookupRequest. They are not worth to cache since
|
|
// computing is not the main cost. Ignore such requests directly to avoid slowly building the cache key.
|
|
if task.cmdType == tikvrpc.CmdCop && worker.store.coprCache != nil && worker.req.Cacheable && worker.store.coprCache.CheckRequestAdmission(len(copReq.Ranges)) {
|
|
cKey, err := coprCacheBuildKey(&copReq)
|
|
if err == nil {
|
|
cacheKey = cKey
|
|
cValue := worker.store.coprCache.Get(cKey)
|
|
copReq.IsCacheEnabled = true
|
|
if cValue != nil && cValue.RegionID == task.region.id && cValue.TimeStamp <= worker.req.StartTs {
|
|
// Append cache version to the request to skip Coprocessor computation if possible
|
|
// when request result is cached
|
|
copReq.CacheIfMatchVersion = cValue.RegionDataVersion
|
|
cacheValue = cValue
|
|
} else {
|
|
copReq.CacheIfMatchVersion = 0
|
|
}
|
|
} else {
|
|
logutil.BgLogger().Warn("Failed to build copr cache key", zap.Error(err))
|
|
}
|
|
}
|
|
|
|
req := tikvrpc.NewReplicaReadRequest(task.cmdType, &copReq, worker.req.ReplicaRead, &worker.replicaReadSeed, kvrpcpb.Context{
|
|
IsolationLevel: pbIsolationLevel(worker.req.IsolationLevel),
|
|
Priority: kvPriorityToCommandPri(worker.req.Priority),
|
|
NotFillCache: worker.req.NotFillCache,
|
|
HandleTime: true,
|
|
ScanDetail: true,
|
|
TaskId: worker.req.TaskID,
|
|
})
|
|
req.StoreTp = task.storeType
|
|
startTime := time.Now()
|
|
if worker.Stats == nil {
|
|
worker.Stats = make(map[tikvrpc.CmdType]*RPCRuntimeStats)
|
|
}
|
|
resp, rpcCtx, storeAddr, err := worker.SendReqCtx(bo, req, task.region, ReadTimeoutMedium, task.storeType, task.storeAddr)
|
|
if err != nil {
|
|
if task.storeType == kv.TiDB {
|
|
err = worker.handleTiDBSendReqErr(err, task, ch)
|
|
return nil, err
|
|
}
|
|
return nil, errors.Trace(err)
|
|
}
|
|
|
|
// Set task.storeAddr field so its task.String() method have the store address information.
|
|
task.storeAddr = storeAddr
|
|
costTime := time.Since(startTime)
|
|
if costTime > minLogCopTaskTime {
|
|
worker.logTimeCopTask(costTime, task, bo, resp)
|
|
}
|
|
metrics.TiKVCoprocessorHistogram.Observe(costTime.Seconds())
|
|
|
|
if task.cmdType == tikvrpc.CmdCopStream {
|
|
return worker.handleCopStreamResult(bo, rpcCtx, resp.Resp.(*tikvrpc.CopStreamResponse), task, ch, costTime)
|
|
}
|
|
|
|
// Handles the response for non-streaming copTask.
|
|
return worker.handleCopResponse(bo, rpcCtx, &copResponse{pbResp: resp.Resp.(*coprocessor.Response)}, cacheKey, cacheValue, task, ch, nil, costTime)
|
|
}
|
|
|
|
type minCommitTSPushed struct {
|
|
data map[uint64]struct{}
|
|
sync.RWMutex
|
|
}
|
|
|
|
func (m *minCommitTSPushed) Update(from []uint64) {
|
|
m.Lock()
|
|
for _, v := range from {
|
|
m.data[v] = struct{}{}
|
|
}
|
|
m.Unlock()
|
|
}
|
|
|
|
func (m *minCommitTSPushed) Get() []uint64 {
|
|
m.RLock()
|
|
defer m.RUnlock()
|
|
if len(m.data) == 0 {
|
|
return nil
|
|
}
|
|
|
|
ret := make([]uint64, 0, len(m.data))
|
|
for k := range m.data {
|
|
ret = append(ret, k)
|
|
}
|
|
return ret
|
|
}
|
|
|
|
// clientHelper wraps LockResolver and RegionRequestSender.
|
|
// It's introduced to support the new lock resolving pattern in the large transaction.
|
|
// In the large transaction protocol, sending requests and resolving locks are
|
|
// context-dependent. For example, when a send request meets a secondary lock, we'll
|
|
// call ResolveLock, and if the lock belongs to a large transaction, we may retry
|
|
// the request. If there is no context information about the resolved locks, we'll
|
|
// meet the secondary lock again and run into a deadloop.
|
|
type clientHelper struct {
|
|
*LockResolver
|
|
*RegionCache
|
|
*minCommitTSPushed
|
|
Client
|
|
resolveLite bool
|
|
RegionRequestRuntimeStats
|
|
}
|
|
|
|
// ResolveLocks wraps the ResolveLocks function and store the resolved result.
|
|
func (ch *clientHelper) ResolveLocks(bo *Backoffer, callerStartTS uint64, locks []*Lock) (int64, error) {
|
|
var err error
|
|
var resolvedLocks []uint64
|
|
var msBeforeTxnExpired int64
|
|
if ch.Stats != nil {
|
|
defer func(start time.Time) {
|
|
recordRegionRequestRuntimeStats(ch.Stats, tikvrpc.CmdResolveLock, time.Since(start))
|
|
}(time.Now())
|
|
}
|
|
if ch.resolveLite {
|
|
msBeforeTxnExpired, resolvedLocks, err = ch.LockResolver.resolveLocksLite(bo, callerStartTS, locks)
|
|
} else {
|
|
msBeforeTxnExpired, resolvedLocks, err = ch.LockResolver.ResolveLocks(bo, callerStartTS, locks)
|
|
}
|
|
if err != nil {
|
|
return msBeforeTxnExpired, err
|
|
}
|
|
if len(resolvedLocks) > 0 {
|
|
ch.minCommitTSPushed.Update(resolvedLocks)
|
|
return 0, nil
|
|
}
|
|
return msBeforeTxnExpired, nil
|
|
}
|
|
|
|
// SendReqCtx wraps the SendReqCtx function and use the resolved lock result in the kvrpcpb.Context.
|
|
func (ch *clientHelper) SendReqCtx(bo *Backoffer, req *tikvrpc.Request, regionID RegionVerID, timeout time.Duration, sType kv.StoreType, directStoreAddr string) (*tikvrpc.Response, *RPCContext, string, error) {
|
|
sender := NewRegionRequestSender(ch.RegionCache, ch.Client)
|
|
if len(directStoreAddr) > 0 {
|
|
sender.storeAddr = directStoreAddr
|
|
}
|
|
sender.Stats = ch.Stats
|
|
req.Context.ResolvedLocks = ch.minCommitTSPushed.Get()
|
|
resp, ctx, err := sender.SendReqCtx(bo, req, regionID, timeout, sType)
|
|
return resp, ctx, sender.storeAddr, err
|
|
}
|
|
|
|
const (
|
|
minLogBackoffTime = 100
|
|
minLogKVProcessTime = 100
|
|
minLogKVWaitTime = 200
|
|
)
|
|
|
|
func (worker *copIteratorWorker) logTimeCopTask(costTime time.Duration, task *copTask, bo *Backoffer, resp *tikvrpc.Response) {
|
|
logStr := fmt.Sprintf("[TIME_COP_PROCESS] resp_time:%s txnStartTS:%d region_id:%d store_addr:%s", costTime, worker.req.StartTs, task.region.id, task.storeAddr)
|
|
if bo.totalSleep > minLogBackoffTime {
|
|
backoffTypes := strings.Replace(fmt.Sprintf("%v", bo.types), " ", ",", -1)
|
|
logStr += fmt.Sprintf(" backoff_ms:%d backoff_types:%s", bo.totalSleep, backoffTypes)
|
|
}
|
|
var detail *kvrpcpb.ExecDetails
|
|
if resp.Resp != nil {
|
|
switch r := resp.Resp.(type) {
|
|
case *coprocessor.Response:
|
|
detail = r.ExecDetails
|
|
case *tikvrpc.CopStreamResponse:
|
|
// streaming request returns io.EOF, so the first CopStreamResponse.Response maybe nil.
|
|
if r.Response != nil {
|
|
detail = r.Response.ExecDetails
|
|
}
|
|
default:
|
|
panic("unreachable")
|
|
}
|
|
}
|
|
|
|
if detail != nil && detail.HandleTime != nil {
|
|
processMs := detail.HandleTime.ProcessMs
|
|
waitMs := detail.HandleTime.WaitMs
|
|
if processMs > minLogKVProcessTime {
|
|
logStr += fmt.Sprintf(" kv_process_ms:%d", processMs)
|
|
if detail.ScanDetail != nil {
|
|
logStr = appendScanDetail(logStr, "write", detail.ScanDetail.Write)
|
|
logStr = appendScanDetail(logStr, "data", detail.ScanDetail.Data)
|
|
logStr = appendScanDetail(logStr, "lock", detail.ScanDetail.Lock)
|
|
}
|
|
}
|
|
if waitMs > minLogKVWaitTime {
|
|
logStr += fmt.Sprintf(" kv_wait_ms:%d", waitMs)
|
|
if processMs <= minLogKVProcessTime {
|
|
logStr = strings.Replace(logStr, "TIME_COP_PROCESS", "TIME_COP_WAIT", 1)
|
|
}
|
|
}
|
|
}
|
|
logutil.Logger(bo.ctx).Info(logStr)
|
|
}
|
|
|
|
func appendScanDetail(logStr string, columnFamily string, scanInfo *kvrpcpb.ScanInfo) string {
|
|
if scanInfo != nil {
|
|
logStr += fmt.Sprintf(" scan_total_%s:%d", columnFamily, scanInfo.Total)
|
|
logStr += fmt.Sprintf(" scan_processed_%s:%d", columnFamily, scanInfo.Processed)
|
|
}
|
|
return logStr
|
|
}
|
|
|
|
func (worker *copIteratorWorker) handleCopStreamResult(bo *Backoffer, rpcCtx *RPCContext, stream *tikvrpc.CopStreamResponse, task *copTask, ch chan<- *copResponse, costTime time.Duration) ([]*copTask, error) {
|
|
defer stream.Close()
|
|
var resp *coprocessor.Response
|
|
var lastRange *coprocessor.KeyRange
|
|
resp = stream.Response
|
|
if resp == nil {
|
|
// streaming request returns io.EOF, so the first Response is nil.
|
|
return nil, nil
|
|
}
|
|
for {
|
|
remainedTasks, err := worker.handleCopResponse(bo, rpcCtx, &copResponse{pbResp: resp}, nil, nil, task, ch, lastRange, costTime)
|
|
if err != nil || len(remainedTasks) != 0 {
|
|
return remainedTasks, errors.Trace(err)
|
|
}
|
|
resp, err = stream.Recv()
|
|
if err != nil {
|
|
if errors.Cause(err) == io.EOF {
|
|
return nil, nil
|
|
}
|
|
|
|
if err1 := bo.Backoff(boTiKVRPC, errors.Errorf("recv stream response error: %v, task: %s", err, task)); err1 != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
|
|
// No coprocessor.Response for network error, rebuild task based on the last success one.
|
|
if errors.Cause(err) == context.Canceled {
|
|
logutil.BgLogger().Info("stream recv timeout", zap.Error(err))
|
|
} else {
|
|
logutil.BgLogger().Info("stream unknown error", zap.Error(err))
|
|
}
|
|
return worker.buildCopTasksFromRemain(bo, lastRange, task)
|
|
}
|
|
if resp.Range != nil {
|
|
lastRange = resp.Range
|
|
}
|
|
}
|
|
}
|
|
|
|
// handleCopResponse checks coprocessor Response for region split and lock,
|
|
// returns more tasks when that happens, or handles the response if no error.
|
|
// if we're handling streaming coprocessor response, lastRange is the range of last
|
|
// successful response, otherwise it's nil.
|
|
func (worker *copIteratorWorker) handleCopResponse(bo *Backoffer, rpcCtx *RPCContext, resp *copResponse, cacheKey []byte, cacheValue *coprCacheValue, task *copTask, ch chan<- *copResponse, lastRange *coprocessor.KeyRange, costTime time.Duration) ([]*copTask, error) {
|
|
if regionErr := resp.pbResp.GetRegionError(); regionErr != nil {
|
|
if rpcCtx != nil && task.storeType == kv.TiDB {
|
|
resp.err = errors.Errorf("error: %v", regionErr)
|
|
worker.sendToRespCh(resp, ch, true)
|
|
return nil, nil
|
|
}
|
|
errStr := fmt.Sprintf("region_id:%v, region_ver:%v, store_type:%s, peer_addr:%s, error:%s",
|
|
task.region.id, task.region.ver, task.storeType.Name(), task.storeAddr, regionErr.String())
|
|
if err := bo.Backoff(BoRegionMiss, errors.New(errStr)); err != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
// We may meet RegionError at the first packet, but not during visiting the stream.
|
|
return buildCopTasks(bo, worker.store.regionCache, task.ranges, worker.req)
|
|
}
|
|
if lockErr := resp.pbResp.GetLocked(); lockErr != nil {
|
|
logutil.BgLogger().Debug("coprocessor encounters",
|
|
zap.Stringer("lock", lockErr))
|
|
msBeforeExpired, err1 := worker.ResolveLocks(bo, worker.req.StartTs, []*Lock{NewLock(lockErr)})
|
|
if err1 != nil {
|
|
return nil, errors.Trace(err1)
|
|
}
|
|
if msBeforeExpired > 0 {
|
|
if err := bo.BackoffWithMaxSleep(boTxnLockFast, int(msBeforeExpired), errors.New(lockErr.String())); err != nil {
|
|
return nil, errors.Trace(err)
|
|
}
|
|
}
|
|
return worker.buildCopTasksFromRemain(bo, lastRange, task)
|
|
}
|
|
if otherErr := resp.pbResp.GetOtherError(); otherErr != "" {
|
|
err := errors.Errorf("other error: %s", otherErr)
|
|
logutil.BgLogger().Warn("other error",
|
|
zap.Uint64("txnStartTS", worker.req.StartTs),
|
|
zap.Uint64("regionID", task.region.id),
|
|
zap.String("storeAddr", task.storeAddr),
|
|
zap.Error(err))
|
|
return nil, errors.Trace(err)
|
|
}
|
|
// When the request is using streaming API, the `Range` is not nil.
|
|
if resp.pbResp.Range != nil {
|
|
resp.startKey = resp.pbResp.Range.Start
|
|
} else if task.ranges != nil && task.ranges.len() > 0 {
|
|
resp.startKey = task.ranges.at(0).StartKey
|
|
}
|
|
if resp.detail == nil {
|
|
resp.detail = new(CopRuntimeStats)
|
|
}
|
|
resp.detail.Stats = worker.Stats
|
|
worker.Stats = nil
|
|
resp.detail.BackoffTime = time.Duration(bo.totalSleep) * time.Millisecond
|
|
resp.detail.BackoffSleep = make(map[string]time.Duration, len(bo.backoffTimes))
|
|
resp.detail.BackoffTimes = make(map[string]int, len(bo.backoffTimes))
|
|
for backoff := range bo.backoffTimes {
|
|
backoffName := backoff.String()
|
|
resp.detail.BackoffTimes[backoffName] = bo.backoffTimes[backoff]
|
|
resp.detail.BackoffSleep[backoffName] = time.Duration(bo.backoffSleepMS[backoff]) * time.Millisecond
|
|
}
|
|
if rpcCtx != nil {
|
|
resp.detail.CalleeAddress = rpcCtx.Addr
|
|
}
|
|
resp.respTime = costTime
|
|
if pbDetails := resp.pbResp.ExecDetails; pbDetails != nil {
|
|
if handleTime := pbDetails.HandleTime; handleTime != nil {
|
|
resp.detail.WaitTime = time.Duration(handleTime.WaitMs) * time.Millisecond
|
|
resp.detail.ProcessTime = time.Duration(handleTime.ProcessMs) * time.Millisecond
|
|
}
|
|
if scanDetail := pbDetails.ScanDetail; scanDetail != nil {
|
|
if scanDetail.Write != nil {
|
|
resp.detail.TotalKeys += scanDetail.Write.Total
|
|
resp.detail.ProcessedKeys += scanDetail.Write.Processed
|
|
}
|
|
}
|
|
}
|
|
if resp.pbResp.IsCacheHit {
|
|
if cacheValue == nil {
|
|
return nil, errors.New("Internal error: received illegal TiKV response")
|
|
}
|
|
// Cache hit and is valid: use cached data as response data and we don't update the cache.
|
|
data := make([]byte, len(cacheValue.Data))
|
|
copy(data, cacheValue.Data)
|
|
resp.pbResp.Data = data
|
|
resp.detail.CoprCacheHit = true
|
|
} else {
|
|
// Cache not hit or cache hit but not valid: update the cache if the response can be cached.
|
|
if cacheKey != nil && resp.pbResp.CanBeCached && resp.pbResp.CacheLastVersion > 0 {
|
|
if worker.store.coprCache.CheckResponseAdmission(resp.pbResp.Data.Size(), resp.detail.ProcessTime) {
|
|
data := make([]byte, len(resp.pbResp.Data))
|
|
copy(data, resp.pbResp.Data)
|
|
|
|
newCacheValue := coprCacheValue{
|
|
Data: data,
|
|
TimeStamp: worker.req.StartTs,
|
|
RegionID: task.region.id,
|
|
RegionDataVersion: resp.pbResp.CacheLastVersion,
|
|
}
|
|
worker.store.coprCache.Set(cacheKey, &newCacheValue)
|
|
}
|
|
}
|
|
}
|
|
worker.sendToRespCh(resp, ch, true)
|
|
return nil, nil
|
|
}
|
|
|
|
// CopRuntimeStats contains execution detail information.
|
|
type CopRuntimeStats struct {
|
|
execdetails.ExecDetails
|
|
RegionRequestRuntimeStats
|
|
|
|
CoprCacheHit bool
|
|
}
|
|
|
|
func (worker *copIteratorWorker) handleTiDBSendReqErr(err error, task *copTask, ch chan<- *copResponse) error {
|
|
errCode := errno.ErrUnknown
|
|
errMsg := err.Error()
|
|
if terror.ErrorEqual(err, ErrTiKVServerTimeout) {
|
|
errCode = errno.ErrTiKVServerTimeout
|
|
errMsg = "TiDB server timeout, address is " + task.storeAddr
|
|
}
|
|
selResp := tipb.SelectResponse{
|
|
Warnings: []*tipb.Error{
|
|
{
|
|
Code: int32(errCode),
|
|
Msg: errMsg,
|
|
},
|
|
},
|
|
}
|
|
data, err := proto.Marshal(&selResp)
|
|
if err != nil {
|
|
return errors.Trace(err)
|
|
}
|
|
resp := &copResponse{
|
|
pbResp: &coprocessor.Response{
|
|
Data: data,
|
|
},
|
|
detail: &CopRuntimeStats{},
|
|
}
|
|
worker.sendToRespCh(resp, ch, true)
|
|
return nil
|
|
}
|
|
|
|
func (worker *copIteratorWorker) buildCopTasksFromRemain(bo *Backoffer, lastRange *coprocessor.KeyRange, task *copTask) ([]*copTask, error) {
|
|
remainedRanges := task.ranges
|
|
if worker.req.Streaming && lastRange != nil {
|
|
remainedRanges = worker.calculateRemain(task.ranges, lastRange, worker.req.Desc)
|
|
}
|
|
return buildCopTasks(bo, worker.store.regionCache, remainedRanges, worker.req)
|
|
}
|
|
|
|
// calculateRemain splits the input ranges into two, and take one of them according to desc flag.
|
|
// It's used in streaming API, to calculate which range is consumed and what needs to be retry.
|
|
// For example:
|
|
// ranges: [r1 --> r2) [r3 --> r4)
|
|
// split: [s1 --> s2)
|
|
// In normal scan order, all data before s1 is consumed, so the remain ranges should be [s1 --> r2) [r3 --> r4)
|
|
// In reverse scan order, all data after s2 is consumed, so the remain ranges should be [r1 --> r2) [r3 --> s2)
|
|
func (worker *copIteratorWorker) calculateRemain(ranges *copRanges, split *coprocessor.KeyRange, desc bool) *copRanges {
|
|
if desc {
|
|
left, _ := ranges.split(split.End)
|
|
return left
|
|
}
|
|
_, right := ranges.split(split.Start)
|
|
return right
|
|
}
|
|
|
|
func (it *copIterator) Close() error {
|
|
if atomic.CompareAndSwapUint32(&it.closed, 0, 1) {
|
|
close(it.finishCh)
|
|
}
|
|
it.rpcCancel.CancelAll()
|
|
it.actionOnExceed.close()
|
|
it.wg.Wait()
|
|
return nil
|
|
}
|
|
|
|
type rateLimit struct {
|
|
token chan struct{}
|
|
}
|
|
|
|
func newRateLimit(n int) *rateLimit {
|
|
return &rateLimit{
|
|
token: make(chan struct{}, n),
|
|
}
|
|
}
|
|
|
|
func (r *rateLimit) getToken(done <-chan struct{}) (exit bool) {
|
|
select {
|
|
case <-done:
|
|
return true
|
|
case r.token <- struct{}{}:
|
|
return false
|
|
}
|
|
}
|
|
|
|
func (r *rateLimit) putToken() {
|
|
select {
|
|
case <-r.token:
|
|
default:
|
|
panic("put a redundant token")
|
|
}
|
|
}
|
|
|
|
// copErrorResponse returns error when calling Next()
|
|
type copErrorResponse struct{ error }
|
|
|
|
func (it copErrorResponse) Next(ctx context.Context) (kv.ResultSubset, error) {
|
|
return nil, it.error
|
|
}
|
|
|
|
func (it copErrorResponse) Close() error {
|
|
return nil
|
|
}
|
|
|
|
// rateLimitAction an OOM Action which is used to control the token if OOM triggered. The token number should be
|
|
// set on initial. Each time the Action is triggered, one token would be destroyed. If the count of the token is less
|
|
// than 2, the action would be delegated to the fallback action.
|
|
type rateLimitAction struct {
|
|
memory.BaseOOMAction
|
|
// enabled indicates whether the rateLimitAction is permitted to Action. 1 means permitted, 0 denied.
|
|
enabled uint32
|
|
// totalTokenNum indicates the total token at initial
|
|
totalTokenNum uint
|
|
cond struct {
|
|
sync.Mutex
|
|
// exceeded indicates whether have encountered OOM situation.
|
|
exceeded bool
|
|
// remainingTokenNum indicates the count of tokens which still exists
|
|
remainingTokenNum uint
|
|
once sync.Once
|
|
// triggerCountForTest indicates the total count of the rateLimitAction's Action being executed
|
|
triggerCountForTest uint
|
|
}
|
|
}
|
|
|
|
func newRateLimitAction(totalTokenNumber uint) *rateLimitAction {
|
|
return &rateLimitAction{
|
|
totalTokenNum: totalTokenNumber,
|
|
cond: struct {
|
|
sync.Mutex
|
|
exceeded bool
|
|
remainingTokenNum uint
|
|
once sync.Once
|
|
triggerCountForTest uint
|
|
}{
|
|
Mutex: sync.Mutex{},
|
|
exceeded: false,
|
|
remainingTokenNum: totalTokenNumber,
|
|
once: sync.Once{},
|
|
},
|
|
}
|
|
}
|
|
|
|
// Action implements ActionOnExceed.Action
|
|
func (e *rateLimitAction) Action(t *memory.Tracker) {
|
|
if !e.isEnabled() {
|
|
if fallback := e.GetFallback(); fallback != nil {
|
|
fallback.Action(t)
|
|
}
|
|
return
|
|
}
|
|
e.conditionLock()
|
|
defer e.conditionUnlock()
|
|
e.cond.once.Do(func() {
|
|
if e.cond.remainingTokenNum < 2 {
|
|
e.setEnabled(false)
|
|
logutil.BgLogger().Info("memory exceeds quota, rateLimitAction delegate to fallback action",
|
|
zap.Uint("total token count", e.totalTokenNum))
|
|
if fallback := e.GetFallback(); fallback != nil {
|
|
fallback.Action(t)
|
|
}
|
|
return
|
|
}
|
|
failpoint.Inject("testRateLimitActionMockConsumeAndAssert", func(val failpoint.Value) {
|
|
if val.(bool) {
|
|
if e.cond.triggerCountForTest+e.cond.remainingTokenNum != e.totalTokenNum {
|
|
panic("triggerCount + remainingTokenNum not equal to totalTokenNum")
|
|
}
|
|
}
|
|
})
|
|
logutil.BgLogger().Info("memory exceeds quota, destroy one token now.",
|
|
zap.Int64("consumed", t.BytesConsumed()),
|
|
zap.Int64("quota", t.GetBytesLimit()),
|
|
zap.Uint("total token count", e.totalTokenNum),
|
|
zap.Uint("remaining token count", e.cond.remainingTokenNum))
|
|
e.cond.exceeded = true
|
|
e.cond.triggerCountForTest++
|
|
})
|
|
}
|
|
|
|
// SetLogHook implements ActionOnExceed.SetLogHook
|
|
func (e *rateLimitAction) SetLogHook(hook func(uint64)) {
|
|
|
|
}
|
|
|
|
// GetPriority get the priority of the Action.
|
|
func (e *rateLimitAction) GetPriority() int64 {
|
|
return memory.DefRateLimitPriority
|
|
}
|
|
|
|
// destroyTokenIfNeeded will check the `exceed` flag after copWorker finished one task.
|
|
// If the exceed flag is true and there is no token been destroyed before, one token will be destroyed,
|
|
// or the token would be return back.
|
|
func (e *rateLimitAction) destroyTokenIfNeeded(returnToken func()) {
|
|
if !e.isEnabled() {
|
|
returnToken()
|
|
return
|
|
}
|
|
e.conditionLock()
|
|
defer e.conditionUnlock()
|
|
if !e.cond.exceeded {
|
|
returnToken()
|
|
return
|
|
}
|
|
// If actionOnExceed has been triggered and there is no token have been destroyed before,
|
|
// destroy one token.
|
|
e.cond.remainingTokenNum = e.cond.remainingTokenNum - 1
|
|
e.cond.exceeded = false
|
|
e.cond.once = sync.Once{}
|
|
}
|
|
|
|
func (e *rateLimitAction) conditionLock() {
|
|
e.cond.Lock()
|
|
}
|
|
|
|
func (e *rateLimitAction) conditionUnlock() {
|
|
e.cond.Unlock()
|
|
}
|
|
|
|
func (e *rateLimitAction) close() {
|
|
if !e.isEnabled() {
|
|
return
|
|
}
|
|
e.setEnabled(false)
|
|
e.conditionLock()
|
|
defer e.conditionUnlock()
|
|
e.cond.exceeded = false
|
|
}
|
|
|
|
func (e *rateLimitAction) setEnabled(enabled bool) {
|
|
newValue := uint32(0)
|
|
if enabled {
|
|
newValue = uint32(1)
|
|
}
|
|
atomic.StoreUint32(&e.enabled, newValue)
|
|
}
|
|
|
|
func (e *rateLimitAction) isEnabled() bool {
|
|
return atomic.LoadUint32(&e.enabled) > 0
|
|
}
|
|
|