Skip to content

Commit

Permalink
*: optimize heartbeat process with concurrent runner - part 1 (#8053)
Browse files Browse the repository at this point in the history
ref #7897

*: Optimize the heartbeat process with the concurrent runner 
- add a switch about the runner
- move some tasks out, not block the heartbeat process

Signed-off-by: nolouch <nolouch@gmail.com>
  • Loading branch information
nolouch authored Apr 15, 2024
1 parent 340c58c commit 0214778
Show file tree
Hide file tree
Showing 16 changed files with 424 additions and 154 deletions.
14 changes: 12 additions & 2 deletions pkg/cluster/cluster.go
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ type Cluster interface {
GetLabelStats() *statistics.LabelStatistics
GetCoordinator() *schedule.Coordinator
GetRuleManager() *placement.RuleManager
GetBasicCluster() *core.BasicCluster
}

// HandleStatsAsync handles the flow asynchronously.
Expand Down Expand Up @@ -55,8 +56,17 @@ func HandleOverlaps(c Cluster, overlaps []*core.RegionInfo) {
}

// Collect collects the cluster information.
func Collect(c Cluster, region *core.RegionInfo, stores []*core.StoreInfo, hasRegionStats bool) {
func Collect(c Cluster, region *core.RegionInfo, hasRegionStats bool) {
if hasRegionStats {
c.GetRegionStats().Observe(region, stores)
// get region again from root tree. make sure the observed region is the latest.
bc := c.GetBasicCluster()
if bc == nil {
return
}
region = bc.GetRegion(region.GetID())
if region == nil {
return
}
c.GetRegionStats().Observe(region, c.GetBasicCluster().GetRegionStores(region))
}
}
40 changes: 40 additions & 0 deletions pkg/core/context.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
// Copyright 2024 TiKV Project Authors.
//
// 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,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package core

import (
"context"

"github.com/tikv/pd/pkg/ratelimit"
)

// MetaProcessContext is a context for meta process.
type MetaProcessContext struct {
context.Context
Tracer RegionHeartbeatProcessTracer
TaskRunner ratelimit.Runner
Limiter *ratelimit.ConcurrencyLimiter
}

// NewMetaProcessContext creates a new MetaProcessContext.
// used in tests, can be changed if no need to test concurrency.
func ContextTODO() *MetaProcessContext {
return &MetaProcessContext{
Context: context.TODO(),
Tracer: NewNoopHeartbeatProcessTracer(),
TaskRunner: ratelimit.NewSyncRunner(),
// Limit default is nil
}
}
82 changes: 62 additions & 20 deletions pkg/core/region.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ package core

import (
"bytes"
"context"
"encoding/hex"
"fmt"
"math"
Expand All @@ -35,6 +36,7 @@ import (
"github.com/pingcap/kvproto/pkg/replication_modepb"
"github.com/pingcap/log"
"github.com/tikv/pd/pkg/errs"
"github.com/tikv/pd/pkg/ratelimit"
"github.com/tikv/pd/pkg/utils/logutil"
"github.com/tikv/pd/pkg/utils/syncutil"
"github.com/tikv/pd/pkg/utils/typeutil"
Expand Down Expand Up @@ -711,20 +713,50 @@ func (r *RegionInfo) isRegionRecreated() bool {

// RegionGuideFunc is a function that determines which follow-up operations need to be performed based on the origin
// and new region information.
type RegionGuideFunc func(region, origin *RegionInfo) (saveKV, saveCache, needSync bool)
type RegionGuideFunc func(ctx *MetaProcessContext, region, origin *RegionInfo) (saveKV, saveCache, needSync bool)

// GenerateRegionGuideFunc is used to generate a RegionGuideFunc. Control the log output by specifying the log function.
// nil means do not print the log.
func GenerateRegionGuideFunc(enableLog bool) RegionGuideFunc {
noLog := func(string, ...zap.Field) {}
debug, info := noLog, noLog
d, i := noLog, noLog
if enableLog {
debug = log.Debug
info = log.Info
d = log.Debug
i = log.Info
}
// Save to storage if meta is updated.
// Save to cache if meta or leader is updated, or contains any down/pending peer.
return func(region, origin *RegionInfo) (saveKV, saveCache, needSync bool) {
return func(ctx *MetaProcessContext, region, origin *RegionInfo) (saveKV, saveCache, needSync bool) {
taskRunner := ctx.TaskRunner
limiter := ctx.Limiter
// print log asynchronously
debug, info := d, i
if taskRunner != nil {
debug = func(msg string, fields ...zap.Field) {
taskRunner.RunTask(
ctx.Context,
ratelimit.TaskOpts{
TaskName: "Log",
Limit: limiter,
},
func(_ context.Context) {
d(msg, fields...)
},
)
}
info = func(msg string, fields ...zap.Field) {
taskRunner.RunTask(
ctx.Context,
ratelimit.TaskOpts{
TaskName: "Log",
Limit: limiter,
},
func(_ context.Context) {
i(msg, fields...)
},
)
}
}
if origin == nil {
if log.GetLevel() <= zap.DebugLevel {
debug("insert new region",
Expand Down Expand Up @@ -789,7 +821,7 @@ func GenerateRegionGuideFunc(enableLog bool) RegionGuideFunc {
}
if !SortedPeersStatsEqual(region.GetDownPeers(), origin.GetDownPeers()) {
if log.GetLevel() <= zap.DebugLevel {
debug("down-peers changed", zap.Uint64("region-id", region.GetID()))
debug("down-peers changed", zap.Uint64("region-id", region.GetID()), zap.Reflect("before", origin.GetDownPeers()), zap.Reflect("after", region.GetDownPeers()))
}
saveCache, needSync = true, true
return
Expand Down Expand Up @@ -912,7 +944,7 @@ func (r *RegionsInfo) CheckAndPutRegion(region *RegionInfo) []*RegionInfo {
if origin == nil || !bytes.Equal(origin.GetStartKey(), region.GetStartKey()) || !bytes.Equal(origin.GetEndKey(), region.GetEndKey()) {
ols = r.tree.overlaps(&regionItem{RegionInfo: region})
}
err := check(region, origin, ols)
err := check(region, origin, convertItemsToRegions(ols))
if err != nil {
log.Debug("region is stale", zap.Stringer("origin", origin.GetMeta()), errs.ZapError(err))
// return the state region to delete.
Expand All @@ -933,48 +965,59 @@ func (r *RegionsInfo) PutRegion(region *RegionInfo) []*RegionInfo {
}

// PreCheckPutRegion checks if the region is valid to put.
func (r *RegionsInfo) PreCheckPutRegion(region *RegionInfo, trace RegionHeartbeatProcessTracer) (*RegionInfo, []*regionItem, error) {
origin, overlaps := r.GetRelevantRegions(region, trace)
func (r *RegionsInfo) PreCheckPutRegion(region *RegionInfo) (*RegionInfo, []*RegionInfo, error) {
origin, overlaps := r.GetRelevantRegions(region)
err := check(region, origin, overlaps)
return origin, overlaps, err
}

func convertItemsToRegions(items []*regionItem) []*RegionInfo {
regions := make([]*RegionInfo, 0, len(items))
for _, item := range items {
regions = append(regions, item.RegionInfo)
}
return regions
}

// AtomicCheckAndPutRegion checks if the region is valid to put, if valid then put.
func (r *RegionsInfo) AtomicCheckAndPutRegion(region *RegionInfo, trace RegionHeartbeatProcessTracer) ([]*RegionInfo, error) {
func (r *RegionsInfo) AtomicCheckAndPutRegion(ctx *MetaProcessContext, region *RegionInfo) ([]*RegionInfo, error) {
tracer := ctx.Tracer
r.t.Lock()
var ols []*regionItem
origin := r.getRegionLocked(region.GetID())
if origin == nil || !bytes.Equal(origin.GetStartKey(), region.GetStartKey()) || !bytes.Equal(origin.GetEndKey(), region.GetEndKey()) {
ols = r.tree.overlaps(&regionItem{RegionInfo: region})
}
trace.OnCheckOverlapsFinished()
err := check(region, origin, ols)
tracer.OnCheckOverlapsFinished()
err := check(region, origin, convertItemsToRegions(ols))
if err != nil {
r.t.Unlock()
trace.OnValidateRegionFinished()
tracer.OnValidateRegionFinished()
return nil, err
}
trace.OnValidateRegionFinished()
tracer.OnValidateRegionFinished()
origin, overlaps, rangeChanged := r.setRegionLocked(region, true, ols...)
r.t.Unlock()
trace.OnSetRegionFinished()
tracer.OnSetRegionFinished()
r.UpdateSubTree(region, origin, overlaps, rangeChanged)
trace.OnUpdateSubTreeFinished()
tracer.OnUpdateSubTreeFinished()
return overlaps, nil
}

// GetRelevantRegions returns the relevant regions for a given region.
func (r *RegionsInfo) GetRelevantRegions(region *RegionInfo, _ RegionHeartbeatProcessTracer) (origin *RegionInfo, overlaps []*regionItem) {
func (r *RegionsInfo) GetRelevantRegions(region *RegionInfo) (origin *RegionInfo, overlaps []*RegionInfo) {
r.t.RLock()
defer r.t.RUnlock()
origin = r.getRegionLocked(region.GetID())
if origin == nil || !bytes.Equal(origin.GetStartKey(), region.GetStartKey()) || !bytes.Equal(origin.GetEndKey(), region.GetEndKey()) {
overlaps = r.tree.overlaps(&regionItem{RegionInfo: region})
for _, item := range r.tree.overlaps(&regionItem{RegionInfo: region}) {
overlaps = append(overlaps, item.RegionInfo)
}
}
return
}

func check(region, origin *RegionInfo, overlaps []*regionItem) error {
func check(region, origin *RegionInfo, overlaps []*RegionInfo) error {
for _, item := range overlaps {
// PD ignores stale regions' heartbeats, unless it is recreated recently by unsafe recover operation.
if region.GetRegionEpoch().GetVersion() < item.GetRegionEpoch().GetVersion() && !region.isRegionRecreated() {
Expand Down Expand Up @@ -1043,7 +1086,6 @@ func (r *RegionsInfo) setRegionLocked(region *RegionInfo, withOverlaps bool, ol
item = &regionItem{RegionInfo: region}
r.regions[region.GetID()] = item
}

var overlaps []*RegionInfo
if rangeChanged {
overlaps = r.tree.update(item, withOverlaps, ol...)
Expand Down
6 changes: 3 additions & 3 deletions pkg/core/region_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -363,7 +363,7 @@ func TestNeedSync(t *testing.T) {
for _, testCase := range testCases {
regionA := region.Clone(testCase.optionsA...)
regionB := region.Clone(testCase.optionsB...)
_, _, needSync := RegionGuide(regionA, regionB)
_, _, needSync := RegionGuide(ContextTODO(), regionA, regionB)
re.Equal(testCase.needSync, needSync)
}
}
Expand Down Expand Up @@ -459,9 +459,9 @@ func TestSetRegionConcurrence(t *testing.T) {
regions := NewRegionsInfo()
region := NewTestRegionInfo(1, 1, []byte("a"), []byte("b"))
go func() {
regions.AtomicCheckAndPutRegion(region, NewNoopHeartbeatProcessTracer())
regions.AtomicCheckAndPutRegion(ContextTODO(), region)
}()
regions.AtomicCheckAndPutRegion(region, NewNoopHeartbeatProcessTracer())
regions.AtomicCheckAndPutRegion(ContextTODO(), region)
re.NoError(failpoint.Disable("github.com/tikv/pd/pkg/core/UpdateSubTree"))
}

Expand Down
5 changes: 5 additions & 0 deletions pkg/core/region_tree.go
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,11 @@ func (r *regionItem) GetStartKey() []byte {
return r.meta.StartKey
}

// GetID returns the ID of the region.
func (r *regionItem) GetID() uint64 {
return r.meta.GetId()
}

// GetEndKey returns the end key of the region.
func (r *regionItem) GetEndKey() []byte {
return r.meta.EndKey
Expand Down
2 changes: 2 additions & 0 deletions pkg/core/region_tree_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -158,6 +158,8 @@ func TestRegionTree(t *testing.T) {

updateNewItem(tree, regionA)
updateNewItem(tree, regionC)
re.Nil(tree.overlaps(newRegionItem([]byte("b"), []byte("c"))))
re.Equal(regionC, tree.overlaps(newRegionItem([]byte("a"), []byte("cc")))[1].RegionInfo)
re.Nil(tree.search([]byte{}))
re.Equal(regionA, tree.search([]byte("a")))
re.Nil(tree.search([]byte("b")))
Expand Down
Loading

0 comments on commit 0214778

Please sign in to comment.