From 3bd392e7cc080ac9cb9d667cd8165d5e4e9a5e49 Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Wed, 20 Nov 2024 17:20:02 +0800 Subject: [PATCH 1/3] This is an automated cherry-pick of #8824 close tikv/pd#8823 Signed-off-by: ti-chi-bot --- server/region_syncer/client.go | 29 +++++++++++++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-) diff --git a/server/region_syncer/client.go b/server/region_syncer/client.go index b7be4a27061..151e7558d2d 100644 --- a/server/region_syncer/client.go +++ b/server/region_syncer/client.go @@ -37,6 +37,11 @@ import ( const ( keepaliveTime = 10 * time.Second keepaliveTimeout = 3 * time.Second +<<<<<<< HEAD:server/region_syncer/client.go +======= + msgSize = 8 * units.MiB + retryInterval = time.Second +>>>>>>> 41ec8dced (syncer: exit watch leader immediately (#8824)):pkg/syncer/client.go ) // StopSyncWithLeader stop to sync the region with leader. @@ -162,7 +167,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 } @@ -174,7 +184,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() { @@ -217,7 +232,17 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) { log.Debug("region is stale", zap.Stringer("origin", origin.GetMeta()), errs.ZapError(err)) continue } +<<<<<<< HEAD:server/region_syncer/client.go _, saveKV, _, _ := regionGuide(region, origin) +======= + cctx := &core.MetaProcessContext{ + Context: ctx, + TaskRunner: ratelimit.NewSyncRunner(), + Tracer: core.NewNoopHeartbeatProcessTracer(), + // no limit for followers. + } + saveKV, _, _, _ := regionGuide(cctx, region, origin) +>>>>>>> 41ec8dced (syncer: exit watch leader immediately (#8824)):pkg/syncer/client.go overlaps := bc.PutRegion(region) if hasBuckets { From 2db28132b0841a86ff1c6309acec30396e48a4fa Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Thu, 21 Nov 2024 16:22:22 +0800 Subject: [PATCH 2/3] fix Signed-off-by: Ryan Leung --- server/region_syncer/client.go | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/server/region_syncer/client.go b/server/region_syncer/client.go index 151e7558d2d..251e7375bcb 100644 --- a/server/region_syncer/client.go +++ b/server/region_syncer/client.go @@ -37,11 +37,7 @@ import ( const ( keepaliveTime = 10 * time.Second keepaliveTimeout = 3 * time.Second -<<<<<<< HEAD:server/region_syncer/client.go -======= - msgSize = 8 * units.MiB retryInterval = time.Second ->>>>>>> 41ec8dced (syncer: exit watch leader immediately (#8824)):pkg/syncer/client.go ) // StopSyncWithLeader stop to sync the region with leader. @@ -232,17 +228,7 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) { log.Debug("region is stale", zap.Stringer("origin", origin.GetMeta()), errs.ZapError(err)) continue } -<<<<<<< HEAD:server/region_syncer/client.go _, saveKV, _, _ := regionGuide(region, origin) -======= - cctx := &core.MetaProcessContext{ - Context: ctx, - TaskRunner: ratelimit.NewSyncRunner(), - Tracer: core.NewNoopHeartbeatProcessTracer(), - // no limit for followers. - } - saveKV, _, _, _ := regionGuide(cctx, region, origin) ->>>>>>> 41ec8dced (syncer: exit watch leader immediately (#8824)):pkg/syncer/client.go overlaps := bc.PutRegion(region) if hasBuckets { From e563284b34b05e90a06090ec2ae7a42e9b1b76c0 Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Mon, 6 Jan 2025 14:02:36 +0800 Subject: [PATCH 3/3] fix Signed-off-by: Ryan Leung --- server/region_syncer/client.go | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/server/region_syncer/client.go b/server/region_syncer/client.go index 251e7375bcb..43ec59dd582 100644 --- a/server/region_syncer/client.go +++ b/server/region_syncer/client.go @@ -163,11 +163,14 @@ 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)) + timer := time.NewTimer(retryInterval) select { case <-ctx.Done(): log.Info("stop synchronizing with leader due to context canceled") + timer.Stop() return - case <-time.After(retryInterval): + case <-timer.C: + timer.Stop() } continue } @@ -180,11 +183,14 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) { if err = stream.CloseSend(); err != nil { log.Error("failed to terminate client stream", errs.ZapError(errs.ErrGRPCCloseSend, err)) } + timer := time.NewTimer(retryInterval) select { case <-ctx.Done(): log.Info("stop synchronizing with leader due to context canceled") + timer.Stop() return - case <-time.After(retryInterval): + case <-timer.C: + timer.Stop() } break }