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

store/tikv: pipe the split and scatter task #23357

Closed
wants to merge 3 commits into from
+84 −18
Closed

store/tikv: pipe the split and scatter task #23357

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
33a3846
store/tikv: pipe the split and scatter task
nolouch Mar 16, 2021
4c4140e
address comments and add more log
nolouch Mar 17, 2021
07f0eb1
tiny clean
nolouch Mar 17, 2021
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
3 changes: 3 additions & 0 deletions ddl/split_region.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ package ddl

import (
"context"
"time"

"github.com/pingcap/errors"
"github.com/pingcap/parser/model"
Expand Down Expand Up @@ -133,6 +134,7 @@ func splitIndexRegion(store kv.SplittableStore, tblInfo *model.TableInfo, scatte
}

func waitScatterRegionFinish(ctx context.Context, store kv.SplittableStore, regionIDs ...uint64) {
start := time.Now()
for _, regionID := range regionIDs {
err := store.WaitScatterRegionFinish(ctx, regionID, 0)
if err != nil {
Expand All @@ -143,4 +145,5 @@ func waitScatterRegionFinish(ctx context.Context, store kv.SplittableStore, regi
}
}
}
logutil.BgLogger().Info("wait scatter region finished", zap.Duration("cost-time", time.Since(start)), zap.String("action", "ddl"), zap.Int("region-numbers", len(regionIDs)))
}
5 changes: 3 additions & 2 deletions store/tikv/rawkv.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -646,6 +646,7 @@ type batch struct {
}

type singleBatchResp struct {
resp *tikvrpc.Response
err error
regionID RegionVerID
resp *tikvrpc.Response
err error
}
94 changes: 78 additions & 16 deletions store/tikv/split_region.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import (
"fmt"
"math"
"sync/atomic"
"time"

"github.com/pingcap/errors"
"github.com/pingcap/kvproto/pkg/kvrpcpb"
Expand Down Expand Up @@ -56,7 +57,7 @@ func (s *KVStore) splitBatchRegionsReq(bo *Backoffer, keys [][]byte, scatter boo
}
// The first time it enters this function.
if bo.totalSleep == 0 {
logutil.BgLogger().Info("split batch regions request",
logutil.BgLogger().Info("batch split regions request",
zap.Int("split key count", len(keys)),
zap.Int("batch count", len(batches)),
zap.Uint64("first batch, region ID", batches[0].regionID.id),
Expand All @@ -67,11 +68,11 @@ func (s *KVStore) splitBatchRegionsReq(bo *Backoffer, keys [][]byte, scatter boo
return resp.resp, errors.Trace(resp.err)
}
ch := make(chan singleBatchResp, len(batches))
// step 1: split all regions in each batch
for _, batch1 := range batches {
go func(b batch) {
backoffer, cancel := bo.Fork()
defer cancel()

util.WithRecovery(func() {
select {
case ch <- s.batchSendSingleRegion(backoffer, b, scatter, tableID):
Expand All @@ -86,6 +87,38 @@ func (s *KVStore) splitBatchRegionsReq(bo *Backoffer, keys [][]byte, scatter boo
}(batch1)
}

// step 2: scatter all regions in each batch
if scatter {
splitResps := make([]singleBatchResp, 0, len(batches))
for i := 0; i < len(batches); i++ {
r := <-ch
splitResps = append(splitResps, r)
}

for _, r := range splitResps {
resp := r
if resp.err != nil {
ch <- resp
continue
}
go func() {
backoffer, cancel := bo.Fork()
defer cancel()
util.WithRecovery(func() {
select {
case ch <- s.scatterRegionsInBatchSplitResp(backoffer, tableID, resp):
case <-bo.ctx.Done():
ch <- singleBatchResp{err: bo.ctx.Err()}
}
}, func(r interface{}) {
if r != nil {
ch <- singleBatchResp{err: errors.Errorf("%v", r)}
}
})
}()
}
}

srResp := &kvrpcpb.SplitRegionResponse{Regions: make([]*metapb.Region, 0, len(keys)*2)}
for i := 0; i < len(batches); i++ {
batchResp := <-ch
Expand Down Expand Up @@ -114,7 +147,11 @@ func (s *KVStore) batchSendSingleRegion(bo *Backoffer, batch batch, scatter bool
}
}
}

start := time.Now()
tid := int64(0)
if tableID != nil {
tid = *tableID
}
req := tikvrpc.NewRequest(tikvrpc.CmdSplitRegion, &kvrpcpb.SplitRegionRequest{
SplitKeys: batch.keys,
}, kvrpcpb.Context{
Expand All @@ -124,7 +161,7 @@ func (s *KVStore) batchSendSingleRegion(bo *Backoffer, batch batch, scatter bool
sender := NewRegionRequestSender(s.regionCache, s.client)
resp, err := sender.SendReq(bo, req, batch.regionID, readTimeoutShort)

batchResp := singleBatchResp{resp: resp}
batchResp := singleBatchResp{regionID: batch.regionID, resp: resp}
if err != nil {
batchResp.err = errors.Trace(err)
return batchResp
Expand All @@ -138,6 +175,11 @@ func (s *KVStore) batchSendSingleRegion(bo *Backoffer, batch batch, scatter bool
err := bo.Backoff(BoRegionMiss, errors.New(regionErr.String()))
if err != nil {
batchResp.err = errors.Trace(err)
logutil.BgLogger().Error("batch split regions request failed",
zap.Uint64("batch region ID", batch.regionID.id),
zap.Int64("table ID", tid),
zap.Duration("cost-time", time.Since(start)),
zap.Error(err))
return batchResp
}
resp, err = s.splitBatchRegionsReq(bo, batch.keys, scatter, tableID)
Expand All @@ -157,28 +199,41 @@ func (s *KVStore) batchSendSingleRegion(bo *Backoffer, batch batch, scatter bool
if len(spResp.Regions) > 0 {
newRegionLeft = logutil.Hex(spResp.Regions[0]).String()
}
logutil.BgLogger().Info("batch split regions complete",
logutil.BgLogger().Info("batch split regions, split regions request complete",
zap.Uint64("batch region ID", batch.regionID.id),
zap.Int64("table ID", tid),
zap.Duration("cost-time", time.Since(start)),
zap.Stringer("first at", kv.Key(batch.keys[0])),
zap.String("first new region left", newRegionLeft),
zap.Int("new region count", len(spResp.Regions)))

if !scatter {
return batchResp
}
return batchResp
}

for i, r := range spResp.Regions {
func (s *KVStore) scatterRegionsInBatchSplitResp(bo *Backoffer, tableID *int64, batchResp singleBatchResp) singleBatchResp {
var err error
start := time.Now()
tid := int64(0)
if tableID != nil {
tid = *tableID
}
spResp := batchResp.resp.Resp.(*kvrpcpb.SplitRegionResponse)
for _, r := range spResp.Regions {
if err = s.scatterRegion(bo, r.Id, tableID); err == nil {
Copy link
Member

Choose a reason for hiding this comment

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

So it's still using

tidb/store/tikv/split_region.go

Line 217 in 579421f

_, err := s.pdClient.ScatterRegions(bo.ctx, []uint64{regionID}, opts...)
the batch API to scatter regions one by one. Which may make the scatter not work?

Copy link
Member

Choose a reason for hiding this comment

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

I mean #22788 .
OK, I'm off topic.

Copy link
Contributor

Choose a reason for hiding this comment

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

The current region scatter may not work in a specific case which described in tikv/pd#3422 , I think it's not related to the api.

logutil.BgLogger().Info("batch split regions, scatter region complete",
zap.Uint64("batch region ID", batch.regionID.id),
zap.Stringer("at", kv.Key(batch.keys[i])),
logutil.BgLogger().Info("batch split regions, scatter single region complete",
zap.Uint64("batch region ID", batchResp.regionID.id),
zap.Uint64("scatter region ID", r.GetId()),
zap.Int64("table ID", tid),
zap.Stringer("at", kv.Key(r.EndKey)),
zap.Stringer("new region left", logutil.Hex(r)))
continue
}

logutil.BgLogger().Info("batch split regions, scatter region failed",
zap.Uint64("batch region ID", batch.regionID.id),
zap.Stringer("at", kv.Key(batch.keys[i])),
logutil.BgLogger().Warn("batch split regions, scatter region failed",
zap.Uint64("batch region ID", batchResp.regionID.id),
zap.Uint64("scatter region ID", r.GetId()),
zap.Int64("table ID", tid),
zap.Stringer("at", kv.Key(r.EndKey)),
zap.Stringer("new region left", logutil.Hex(r)),
zap.Error(err))
if batchResp.err == nil {
Expand All @@ -188,11 +243,18 @@ func (s *KVStore) batchSendSingleRegion(bo *Backoffer, batch batch, scatter bool
break
}
}

logutil.BgLogger().Info("batch split regions, scatter all region complete",
zap.Uint64("batch region ID", batchResp.regionID.id),
zap.Duration("cost-time", time.Since(start)),
zap.Int64("table ID", tid))

return batchResp
}

// SplitRegions splits regions by splitKeys.
func (s *KVStore) SplitRegions(ctx context.Context, splitKeys [][]byte, scatter bool, tableID *int64) (regionIDs []uint64, err error) {
start := time.Now()
bo := NewBackofferWithVars(ctx, int(math.Min(float64(len(splitKeys))*splitRegionBackoff, maxSplitRegionsBackoff)), nil)
resp, err := s.splitBatchRegionsReq(bo, splitKeys, scatter, tableID)
regionIDs = make([]uint64, 0, len(splitKeys))
Expand All @@ -201,8 +263,8 @@ func (s *KVStore) SplitRegions(ctx context.Context, splitKeys [][]byte, scatter
for _, r := range spResp.Regions {
regionIDs = append(regionIDs, r.Id)
}
logutil.BgLogger().Info("split regions complete", zap.Int("region count", len(regionIDs)), zap.Uint64s("region IDs", regionIDs))
}
logutil.BgLogger().Info("split regions request complete", zap.Int("region count", len(regionIDs)), zap.Duration("cost-time", time.Since(start)), zap.Uint64s("region IDs", regionIDs))
return regionIDs, errors.Trace(err)
}

Expand Down