Skip to content

Commit

Permalink
Merge branch 'release-8.1' into cherry-pick-8539-to-release-8.1
Browse files Browse the repository at this point in the history
  • Loading branch information
ti-chi-bot[bot] authored Dec 12, 2024
2 parents efbef2a + ec3892a commit a0e2e18
Show file tree
Hide file tree
Showing 14 changed files with 203 additions and 95 deletions.
9 changes: 5 additions & 4 deletions client/resource_group/controller/controller.go
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ import (
"github.com/prometheus/client_golang/prometheus"
pd "github.com/tikv/pd/client"
"github.com/tikv/pd/client/errs"
"github.com/tikv/pd/client/timerutil"
atomicutil "go.uber.org/atomic"
"go.uber.org/zap"
"golang.org/x/exp/slices"
Expand Down Expand Up @@ -291,7 +292,7 @@ func (c *ResourceGroupsController) Start(ctx context.Context) {
watchMetaChannel, err = c.provider.Watch(ctx, pd.GroupSettingsPathPrefixBytes, pd.WithRev(metaRevision), pd.WithPrefix(), pd.WithPrevKV())
if err != nil {
log.Warn("watch resource group meta failed", zap.Error(err))
watchRetryTimer.Reset(watchRetryInterval)
timerutil.SafeResetTimer(watchRetryTimer, watchRetryInterval)
failpoint.Inject("watchStreamError", func() {
watchRetryTimer.Reset(20 * time.Millisecond)
})
Expand All @@ -301,7 +302,7 @@ func (c *ResourceGroupsController) Start(ctx context.Context) {
watchConfigChannel, err = c.provider.Watch(ctx, pd.ControllerConfigPathPrefixBytes, pd.WithRev(cfgRevision), pd.WithPrefix())
if err != nil {
log.Warn("watch resource group config failed", zap.Error(err))
watchRetryTimer.Reset(watchRetryInterval)
timerutil.SafeResetTimer(watchRetryTimer, watchRetryInterval)
}
}
case <-emergencyTokenAcquisitionTicker.C:
Expand Down Expand Up @@ -335,7 +336,7 @@ func (c *ResourceGroupsController) Start(ctx context.Context) {
})
if !ok {
watchMetaChannel = nil
watchRetryTimer.Reset(watchRetryInterval)
timerutil.SafeResetTimer(watchRetryTimer, watchRetryInterval)
failpoint.Inject("watchStreamError", func() {
watchRetryTimer.Reset(20 * time.Millisecond)
})
Expand Down Expand Up @@ -371,7 +372,7 @@ func (c *ResourceGroupsController) Start(ctx context.Context) {
case resp, ok := <-watchConfigChannel:
if !ok {
watchConfigChannel = nil
watchRetryTimer.Reset(watchRetryInterval)
timerutil.SafeResetTimer(watchRetryTimer, watchRetryInterval)
failpoint.Inject("watchStreamError", func() {
watchRetryTimer.Reset(20 * time.Millisecond)
})
Expand Down
2 changes: 1 addition & 1 deletion client/timerpool/pool.go → client/timerutil/pool.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

// Note: This file is copied from https://go-review.googlesource.com/c/go/+/276133

package timerpool
package timerutil

import (
"sync"
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

// Note: This file is copied from https://go-review.googlesource.com/c/go/+/276133

package timerpool
package timerutil

import (
"testing"
Expand Down
32 changes: 32 additions & 0 deletions client/timerutil/util.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,32 @@
// 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 timerutil

import "time"

// SafeResetTimer is used to reset timer safely.
// Before Go 1.23, the only safe way to use Reset was to call Timer.Stop and explicitly drain the timer first.
// We need be careful here, see more details in the comments of Timer.Reset.
// https://pkg.go.dev/time@master#Timer.Reset
func SafeResetTimer(t *time.Timer, d time.Duration) {
// Stop the timer if it's not stopped.
if !t.Stop() {
select {
case <-t.C: // try to drain from the channel
default:
}
}
t.Reset(d)
}
21 changes: 6 additions & 15 deletions client/tso_dispatcher.go
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ import (
"github.com/tikv/pd/client/errs"
"github.com/tikv/pd/client/grpcutil"
"github.com/tikv/pd/client/retry"
"github.com/tikv/pd/client/timerpool"
"github.com/tikv/pd/client/timerutil"
"github.com/tikv/pd/client/tsoutil"
"go.uber.org/zap"
"google.golang.org/grpc"
Expand Down Expand Up @@ -151,7 +151,7 @@ func newTSDeadline(
done chan struct{},
cancel context.CancelFunc,
) *deadline {
timer := timerpool.GlobalTimerPool.Get(timeout)
timer := timerutil.GlobalTimerPool.Get(timeout)
return &deadline{
timer: timer,
done: done,
Expand Down Expand Up @@ -197,11 +197,11 @@ func (c *tsoClient) watchTSDeadline(ctx context.Context, dcLocation string) {
case <-d.timer.C:
log.Error("[tso] tso request is canceled due to timeout", zap.String("dc-location", dc), errs.ZapError(errs.ErrClientGetTSOTimeout))
d.cancel()
timerpool.GlobalTimerPool.Put(d.timer)
timerutil.GlobalTimerPool.Put(d.timer)
case <-d.done:
timerpool.GlobalTimerPool.Put(d.timer)
timerutil.GlobalTimerPool.Put(d.timer)
case <-ctx.Done():
timerpool.GlobalTimerPool.Put(d.timer)
timerutil.GlobalTimerPool.Put(d.timer)
return
}
case <-ctx.Done():
Expand Down Expand Up @@ -432,16 +432,7 @@ tsoBatchLoop:
if maxBatchWaitInterval >= 0 {
tbc.adjustBestBatchSize()
}
// Stop the timer if it's not stopped.
if !streamLoopTimer.Stop() {
select {
case <-streamLoopTimer.C: // try to drain from the channel
default:
}
}
// We need be careful here, see more details in the comments of Timer.Reset.
// https://pkg.go.dev/time@master#Timer.Reset
streamLoopTimer.Reset(c.option.timeout)
timerutil.SafeResetTimer(streamLoopTimer, c.option.timeout)
// Choose a stream to send the TSO gRPC request.
streamChoosingLoop:
for {
Expand Down
12 changes: 2 additions & 10 deletions pkg/election/lease.go
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ import (
"github.com/tikv/pd/pkg/errs"
"github.com/tikv/pd/pkg/utils/etcdutil"
"github.com/tikv/pd/pkg/utils/logutil"
"github.com/tikv/pd/pkg/utils/timerutil"
"github.com/tikv/pd/pkg/utils/typeutil"
"go.etcd.io/etcd/clientv3"
"go.uber.org/zap"
Expand Down Expand Up @@ -124,16 +125,7 @@ func (l *lease) KeepAlive(ctx context.Context) {
l.expireTime.Store(t)
}
}
// Stop the timer if it's not stopped.
if !timer.Stop() {
select {
case <-timer.C: // try to drain from the channel
default:
}
}
// We need be careful here, see more details in the comments of Timer.Reset.
// https://pkg.go.dev/time@master#Timer.Reset
timer.Reset(l.leaseTimeout)
timerutil.SafeResetTimer(timer, l.leaseTimeout)
case <-timer.C:
log.Info("keep alive lease too slow", zap.Duration("timeout-duration", l.leaseTimeout), zap.Time("actual-expire", l.expireTime.Load().(time.Time)), zap.String("purpose", l.Purpose))
return
Expand Down
88 changes: 88 additions & 0 deletions pkg/mock/mockserver/mockserver.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
// 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 mockserver

import (
"context"

"github.com/pingcap/kvproto/pkg/pdpb"
"github.com/tikv/pd/pkg/core"
"github.com/tikv/pd/pkg/storage"
"github.com/tikv/pd/pkg/utils/grpcutil"
)

// MockServer is used to mock Server for test use.
type MockServer struct {
ctx context.Context
member, leader *pdpb.Member
storage storage.Storage
bc *core.BasicCluster
}

// NewMockServer creates a new MockServer.
func NewMockServer(ctx context.Context, member, leader *pdpb.Member, storage storage.Storage, bc *core.BasicCluster) *MockServer {
return &MockServer{
ctx: ctx,
member: member,
leader: leader,
storage: storage,
bc: bc,
}
}

// LoopContext returns the context of the server.
func (s *MockServer) LoopContext() context.Context {
return s.ctx
}

// ClusterID returns the cluster ID of the server.
func (*MockServer) ClusterID() uint64 {
return 1
}

// GetMemberInfo returns the member info of the server.
func (s *MockServer) GetMemberInfo() *pdpb.Member {
return s.member
}

// GetLeader returns the leader of the server.
func (s *MockServer) GetLeader() *pdpb.Member {
return s.leader
}

// GetStorage returns the storage of the server.
func (s *MockServer) GetStorage() storage.Storage {
return s.storage
}

// Name returns the name of the server.
func (*MockServer) Name() string {
return "mock-server"
}

// GetRegions returns the regions of the server.
func (s *MockServer) GetRegions() []*core.RegionInfo {
return s.bc.GetRegions()
}

// GetTLSConfig returns the TLS config of the server.
func (*MockServer) GetTLSConfig() *grpcutil.TLSConfig {
return &grpcutil.TLSConfig{}
}

// GetBasicCluster returns the basic cluster of the server.
func (s *MockServer) GetBasicCluster() *core.BasicCluster {
return s.bc
}
15 changes: 13 additions & 2 deletions pkg/syncer/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ const (
keepaliveTime = 10 * time.Second
keepaliveTimeout = 3 * time.Second
msgSize = 8 * units.MiB
retryInterval = time.Second
)

// StopSyncWithLeader stop to sync the region with leader.
Expand Down Expand Up @@ -150,7 +151,12 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) {
}
}
log.Error("server failed to establish sync stream with leader", zap.String("server", s.server.Name()), zap.String("leader", s.server.GetLeader().GetName()), errs.ZapError(err))
time.Sleep(time.Second)
select {
case <-ctx.Done():
log.Info("stop synchronizing with leader due to context canceled")
return
case <-time.After(retryInterval):
}
continue
}
log.Info("server starts to synchronize with leader", zap.String("server", s.server.Name()), zap.String("leader", s.server.GetLeader().GetName()), zap.Uint64("request-index", s.history.GetNextIndex()))
Expand All @@ -162,7 +168,12 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) {
if err = stream.CloseSend(); err != nil {
log.Error("failed to terminate client stream", errs.ZapError(errs.ErrGRPCCloseSend, err))
}
time.Sleep(time.Second)
select {
case <-ctx.Done():
log.Info("stop synchronizing with leader due to context canceled")
return
case <-time.After(retryInterval):
}
break
}
if s.history.GetNextIndex() != resp.GetStartIndex() {
Expand Down
69 changes: 15 additions & 54 deletions pkg/syncer/client_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,9 +21,9 @@ import (

"github.com/pingcap/failpoint"
"github.com/pingcap/kvproto/pkg/metapb"
"github.com/pingcap/kvproto/pkg/pdpb"
"github.com/stretchr/testify/require"
"github.com/tikv/pd/pkg/core"
"github.com/tikv/pd/pkg/mock/mockserver"
"github.com/tikv/pd/pkg/storage"
"github.com/tikv/pd/pkg/utils/grpcutil"
"google.golang.org/grpc/codes"
Expand All @@ -37,11 +37,13 @@ func TestLoadRegion(t *testing.T) {
rs, err := storage.NewRegionStorageWithLevelDBBackend(context.Background(), tempDir, nil)
re.NoError(err)

server := &mockServer{
ctx: context.Background(),
storage: storage.NewCoreStorage(storage.NewStorageWithMemoryBackend(), rs),
bc: core.NewBasicCluster(),
}
server := mockserver.NewMockServer(
context.Background(),
nil,
nil,
storage.NewCoreStorage(storage.NewStorageWithMemoryBackend(), rs),
core.NewBasicCluster(),
)
for i := 0; i < 30; i++ {
rs.SaveRegion(&metapb.Region{Id: uint64(i) + 1})
}
Expand All @@ -64,11 +66,13 @@ func TestErrorCode(t *testing.T) {
tempDir := t.TempDir()
rs, err := storage.NewRegionStorageWithLevelDBBackend(context.Background(), tempDir, nil)
re.NoError(err)
server := &mockServer{
ctx: context.Background(),
storage: storage.NewCoreStorage(storage.NewStorageWithMemoryBackend(), rs),
bc: core.NewBasicCluster(),
}
server := mockserver.NewMockServer(
context.Background(),
nil,
nil,
storage.NewCoreStorage(storage.NewStorageWithMemoryBackend(), rs),
core.NewBasicCluster(),
)
ctx, cancel := context.WithCancel(context.TODO())
rc := NewRegionSyncer(server)
conn, err := grpcutil.GetClientConn(ctx, "http://127.0.0.1", nil)
Expand All @@ -79,46 +83,3 @@ func TestErrorCode(t *testing.T) {
re.True(ok)
re.Equal(codes.Canceled, ev.Code())
}

type mockServer struct {
ctx context.Context
member, leader *pdpb.Member
storage storage.Storage
bc *core.BasicCluster
}

func (s *mockServer) LoopContext() context.Context {
return s.ctx
}

func (s *mockServer) ClusterID() uint64 {
return 1
}

func (s *mockServer) GetMemberInfo() *pdpb.Member {
return s.member
}

func (s *mockServer) GetLeader() *pdpb.Member {
return s.leader
}

func (s *mockServer) GetStorage() storage.Storage {
return s.storage
}

func (s *mockServer) Name() string {
return "mock-server"
}

func (s *mockServer) GetRegions() []*core.RegionInfo {
return s.bc.GetRegions()
}

func (s *mockServer) GetTLSConfig() *grpcutil.TLSConfig {
return &grpcutil.TLSConfig{}
}

func (s *mockServer) GetBasicCluster() *core.BasicCluster {
return s.bc
}
Loading

0 comments on commit a0e2e18

Please sign in to comment.