statistics: fix wrong singleflight implementation for stats' syncload

Signed-off-by: Weizhen Wang <wangweizhen@pingcap.com>
pingcap · Apr 2, 2024 · 92241bd · 92241bd
1 parent ac22710
commit 92241bd
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 13 deletions.
diff --git a/pkg/parser/model/model.go b/pkg/parser/model/model.go
@@ -1786,6 +1786,7 @@ type TableItemID struct {
 	IsSyncLoadFailed bool
 }
 
+// Key is used to generate unique key for TableItemID to use in the syncload
 func (t TableItemID) Key() string {
 	return fmt.Sprintf("%d%d%t%t", t.ID, t.TableID, t.IsIndex, t.IsSyncLoadFailed)
 }

diff --git a/pkg/statistics/handle/syncload/stats_syncload.go b/pkg/statistics/handle/syncload/stats_syncload.go
@@ -48,7 +48,6 @@ func NewStatsSyncLoad(statsHandle statstypes.StatsHandle) statstypes.StatsSyncLo
 	s.StatsLoad.SubCtxs = make([]sessionctx.Context, cfg.Performance.StatsLoadConcurrency)
 	s.StatsLoad.NeededItemsCh = make(chan *statstypes.NeededItemTask, cfg.Performance.StatsLoadQueueSize)
 	s.StatsLoad.TimeoutItemsCh = make(chan *statstypes.NeededItemTask, cfg.Performance.StatsLoadQueueSize)
-	//s.StatsLoad.WorkingColMap = map[model.TableItemID][]chan stmtctx.StatsLoadResult{}
 	return s
 }
 
@@ -231,13 +230,13 @@ func (s *statsSyncLoad) HandleOneTask(sctx sessionctx.Context, lastTask *statsty
 	} else {
 		task = lastTask
 	}
-	resultChan := s.StatsLoad.Singleflight.DoChan(task.Item.Key(), func() (interface{}, error) {
+	resultChan := s.StatsLoad.Singleflight.DoChan(task.Item.Key(), func() (any, error) {
 		return s.handleOneItemTask(sctx, task)
 	})
-	timeout := task.ToTimeout.Sub(time.Now())
+	timeout := time.Until(task.ToTimeout)
 	select {
 	case result := <-resultChan:
-		if result.Err != nil {
+		if result.Err == nil {
 			slr := *(result.Val.(*stmtctx.StatsLoadResult))
 			if slr.Error != nil {
 				return task, slr.Error
@@ -251,14 +250,20 @@ func (s *statsSyncLoad) HandleOneTask(sctx sessionctx.Context, lastTask *statsty
 	}
 }
 
-func (s *statsSyncLoad) handleOneItemTask(sctx sessionctx.Context, task *statstypes.NeededItemTask) (*stmtctx.StatsLoadResult, error) {
-	result := &stmtctx.StatsLoadResult{Item: task.Item.TableItemID}
+func (s *statsSyncLoad) handleOneItemTask(sctx sessionctx.Context, task *statstypes.NeededItemTask) (result *stmtctx.StatsLoadResult, err error) {
+	defer func() {
+		// recover for each task, worker keeps working
+		if r := recover(); r != nil {
+			logutil.BgLogger().Error("handleOneItemTask panicked", zap.Any("recover", r), zap.Stack("stack"))
+			err = errors.Errorf("stats loading panicked: %v", r)
+		}
+	}()
+	result = &stmtctx.StatsLoadResult{Item: task.Item.TableItemID}
 	item := result.Item
 	tbl, ok := s.statsHandle.Get(item.TableID)
 	if !ok {
 		return result, nil
 	}
-	var err error
 	wrapper := &statsWrapper{}
 	if item.IsIndex {
 		index, loadNeeded := tbl.IndexIsLoadNeeded(item.ID)

diff --git a/pkg/statistics/handle/syncload/stats_syncload_test.go b/pkg/statistics/handle/syncload/stats_syncload_test.go
@@ -207,17 +207,14 @@ func TestConcurrentLoadHistWithPanicAndFail(t *testing.T) {
 		require.Error(t, err1)
 		require.NotNil(t, task1)
 
-		task2, err2 := h.HandleOneTask(testKit.Session().(sessionctx.Context), nil, exitCh)
-		require.Nil(t, err2)
-		require.Nil(t, task2)
-
 		require.NoError(t, failpoint.Disable(fp.failPath))
 		task3, err3 := h.HandleOneTask(testKit.Session().(sessionctx.Context), task1, exitCh)
 		require.NoError(t, err3)
 		require.Nil(t, task3)
 
-		require.Len(t, stmtCtx1.StatsLoad.ResultCh, 1)
-		require.Len(t, stmtCtx2.StatsLoad.ResultCh, 1)
+		task, err3 := h.HandleOneTask(testKit.Session().(sessionctx.Context), nil, exitCh)
+		require.NoError(t, err3)
+		require.Nil(t, task)
 
 		rs1, ok1 := <-stmtCtx1.StatsLoad.ResultCh
 		require.True(t, ok1)

diff --git a/pkg/statistics/handle/types/BUILD.bazel b/pkg/statistics/handle/types/BUILD.bazel
@@ -17,5 +17,6 @@ go_library(
         "//pkg/types",
         "//pkg/util",
         "//pkg/util/sqlexec",
+        "@org_golang_x_sync//singleflight",
     ],
 )