From 4f97ed883e0f1f0c93fc6021d7589cc69d5d2be7 Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Tue, 19 Nov 2024 15:39:57 +0800 Subject: [PATCH 1/4] exit watch leader immediately Signed-off-by: Ryan Leung --- pkg/syncer/client.go | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/pkg/syncer/client.go b/pkg/syncer/client.go index dd6edbd8d73..a094ed0988d 100644 --- a/pkg/syncer/client.go +++ b/pkg/syncer/client.go @@ -43,6 +43,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. @@ -153,7 +154,11 @@ 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(): + 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())) @@ -165,7 +170,11 @@ 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(): + return + case <-time.After(retryInterval): + } break } if s.history.getNextIndex() != resp.GetStartIndex() { From 0903d7cb0f27a6769734c5e1235b34ba58b9be28 Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Wed, 20 Nov 2024 15:47:41 +0800 Subject: [PATCH 2/4] add debug log Signed-off-by: Ryan Leung --- pkg/syncer/client.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pkg/syncer/client.go b/pkg/syncer/client.go index a094ed0988d..c519da09fa4 100644 --- a/pkg/syncer/client.go +++ b/pkg/syncer/client.go @@ -156,6 +156,7 @@ 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)) select { case <-ctx.Done(): + log.Info("context is done, server stops to synchronize with leader", zap.String("server", s.server.Name()), zap.String("leader", s.server.GetLeader().GetName())) return case <-time.After(retryInterval): } @@ -172,6 +173,7 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) { } select { case <-ctx.Done(): + log.Info("context is done, server stops to synchronize with leader", zap.String("server", s.server.Name()), zap.String("leader", s.server.GetLeader().GetName())) return case <-time.After(retryInterval): } From 745ba97961d182cb8b773b0e598f1e581d38e81a Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Wed, 20 Nov 2024 16:18:24 +0800 Subject: [PATCH 3/4] remove log Signed-off-by: Ryan Leung --- pkg/syncer/client.go | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/pkg/syncer/client.go b/pkg/syncer/client.go index c519da09fa4..86a32e9c045 100644 --- a/pkg/syncer/client.go +++ b/pkg/syncer/client.go @@ -156,7 +156,6 @@ 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)) select { case <-ctx.Done(): - log.Info("context is done, server stops to synchronize with leader", zap.String("server", s.server.Name()), zap.String("leader", s.server.GetLeader().GetName())) return case <-time.After(retryInterval): } @@ -173,7 +172,6 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) { } select { case <-ctx.Done(): - log.Info("context is done, server stops to synchronize with leader", zap.String("server", s.server.Name()), zap.String("leader", s.server.GetLeader().GetName())) return case <-time.After(retryInterval): } @@ -219,13 +217,13 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) { log.Debug("region is stale", zap.Stringer("origin", origin.GetMeta()), errs.ZapError(err)) continue } - ctx := &core.MetaProcessContext{ + cctx := &core.MetaProcessContext{ Context: ctx, TaskRunner: ratelimit.NewSyncRunner(), Tracer: core.NewNoopHeartbeatProcessTracer(), // no limit for followers. } - saveKV, _, _, _ := regionGuide(ctx, region, origin) + saveKV, _, _, _ := regionGuide(cctx, region, origin) overlaps := bc.PutRegion(region) if hasBuckets { From 0e396cf80879762c480cfc4ab115976bb7bfec1f Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Wed, 20 Nov 2024 16:24:21 +0800 Subject: [PATCH 4/4] add log Signed-off-by: Ryan Leung --- pkg/syncer/client.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/pkg/syncer/client.go b/pkg/syncer/client.go index 86a32e9c045..3d7f36d9114 100644 --- a/pkg/syncer/client.go +++ b/pkg/syncer/client.go @@ -156,6 +156,7 @@ 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)) select { case <-ctx.Done(): + log.Info("stop synchronizing with leader due to context canceled") return case <-time.After(retryInterval): } @@ -172,6 +173,7 @@ func (s *RegionSyncer) StartSyncWithLeader(addr string) { } select { case <-ctx.Done(): + log.Info("stop synchronizing with leader due to context canceled") return case <-time.After(retryInterval): }