executor: support tidb memory debug mode

executor/explain.go

@@ -16,11 +16,21 @@ package executor
 
 import (
 	"context"
+	"go.uber.org/zap"
+	"os"
+	"path/filepath"
+	"runtime"
+	rpprof "runtime/pprof"
+	"strconv"
+	"time"
 
+	"github.com/cznic/mathutil"
 	"github.com/pingcap/errors"
+	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/planner/core"
 	"github.com/pingcap/tidb/util/chunk"
-	"github.com/pingcap/tidb/util/mathutil"
+	"github.com/pingcap/tidb/util/logutil"
+	"github.com/pingcap/tidb/util/memory"
 )
 
 // ExplainExec represents an explain executor.
@@ -89,6 +99,16 @@ func (e *ExplainExec) executeAnalyzeExec(ctx context.Context) (err error) {
 				}
 			}
 		}()
+		if e.ctx.GetSessionVars().MemoryDebugMode > 0 {
+			exit := make(chan bool)
+			e.runMemoryDebugGoroutine(exit)
+			defer func() {
+				select {
+				case <-exit:
+				}
+			}()
+			defer func() { exit <- false }()
+		}
 		e.executed = true
 		chk := newFirstChunk(e.analyzeExec)
 		for {
@@ -123,3 +143,86 @@ func (e *ExplainExec) getAnalyzeExecToExecutedNoDelay() Executor {
 	}
 	return nil
 }
+
+func (e *ExplainExec) runMemoryDebugGoroutine(exit chan bool) {
+	debugMode := e.ctx.GetSessionVars().MemoryDebugMode
+	ticker := time.NewTicker(5 * time.Second)
+	tracker := e.ctx.GetSessionVars().StmtCtx.MemTracker
+	instanceStats := &runtime.MemStats{}
+	var heapInUse, trackedMem uint64
+	const GB = 1024 * 1024 * 1024
+	go func() {
+		defer func() {
+			runtime.GC()
+			runtime.ReadMemStats(instanceStats)
+			heapInUse = instanceStats.HeapInuse
+			trackedMem = uint64(tracker.BytesConsumed())
+			logutil.BgLogger().Warn("Memory Debug Mode",
+				zap.String("sql", "finished"),
+				zap.String("heap in use", memory.FormatBytes(int64(heapInUse))),
+				zap.String("tracked memory", memory.FormatBytes(int64(trackedMem))),
+				zap.String("heap profile", e.getHeapProfile(false)))
+			close(exit)
+		}()
+		times := 0
+		for {
+			select {
+			case <-exit:
+				return
+			case <-ticker.C:
+				if debugMode == 2 {
+					runtime.GC()
+				}
+				runtime.ReadMemStats(instanceStats)
+				heapInUse = instanceStats.HeapInuse
+				trackedMem = uint64(tracker.BytesConsumed())
+
+				times++
+				if times%6 == 0 {
+					logutil.BgLogger().Warn("Memory Debug Mode",
+						zap.String("sql", "running"),
+						zap.String("heap in use", memory.FormatBytes(int64(heapInUse))),
+						zap.String("tracked memory", memory.FormatBytes(int64(trackedMem))))
+				}
+
+				if debugMode == 1 {
+					if heapInUse > 10*GB && trackedMem/10*15 < heapInUse {
+						logutil.BgLogger().Warn("Memory Debug Mode",
+							zap.String("debug mode 1 alarm", "trackedMem * 150% < heapInUse, maybe some allocation and free frequently"),
+							zap.String("heap in use", memory.FormatBytes(int64(heapInUse))),
+							zap.String("tracked memory", memory.FormatBytes(int64(trackedMem))),
+							zap.String("heap profile", e.getHeapProfile(true)))
+					}
+				} else {
+					if heapInUse > 10*GB && trackedMem/10*11 < heapInUse {
+						logutil.BgLogger().Warn("Memory Debug Mode",
+							zap.String("debug mode 2 alarm", "trackedMem * 110% < heapInUse after GC, maybe some memory not be tracked"),
+							zap.String("heap in use", memory.FormatBytes(int64(heapInUse))),
+							zap.String("tracked memory", memory.FormatBytes(int64(trackedMem))),
+							zap.String("heap profile", e.getHeapProfile(false)))
+						ts := tracker.SearchTrackerConsumedMoreThanNBytes(GB)
+						logs := make([]zap.Field, 0, len(ts))
+						for _, t := range ts {
+							logs = append(logs, zap.String("Executor_"+strconv.Itoa(t.Label()), memory.FormatBytes(t.BytesConsumed())))
+						}
+						logutil.BgLogger().Warn("Memory Debug Mode, Log all trackers that consumes more than 1GB", logs...)
+					}
+				}
+			}
+		}
+	}()
+}
+
+func (e *ExplainExec) getHeapProfile(gc bool) (fileName string) {
+	tempDir := filepath.Join(config.GetGlobalConfig().TempStoragePath, "record")
+	timeString := time.Now().Format(time.RFC3339)
+	if gc {
+		runtime.GC()
+	}
+	fileName = filepath.Join(tempDir, "heapGC"+timeString)
+	f, _ := os.Create(fileName)
+	p := rpprof.Lookup("heap")
+	_ = p.WriteTo(f, 0)
+	_ = f.Close()
+	return fileName
+}

planner/core/common_plans.go

@@ -1354,6 +1354,11 @@ func (e *Explain) RenderResult() error {
 			if err != nil {
 				return err
 			}
+			if e.Analyze {
+				row := e.Rows[0]
+				tracker := e.SCtx().GetSessionVars().StmtCtx.MemTracker
+				row[7] = row[7] + "(Total: " + tracker.FormatBytes(tracker.MaxConsumed()) + ")"
+			}
 		}
 	case types.ExplainFormatDOT:
 		if physicalPlan, ok := e.TargetPlan.(PhysicalPlan); ok {

sessionctx/variable/session.go

@@ -1155,6 +1155,9 @@ type SessionVars struct {
 
 	// MaxAllowedPacket indicates the maximum size of a packet for the MySQL protocol.
 	MaxAllowedPacket uint64
+
+	// MemoryDebugMode indicated the memory debug mode.
+	MemoryDebugMode int64
 }
 
 // InitStatementContext initializes a StatementContext, the object is reused to reduce allocation.

sessionctx/variable/sysvar.go

@@ -1641,6 +1641,10 @@ var defaultSysVars = []*SysVar{
 			metrics.ToggleSimplifiedMode(TiDBOptOn(s))
 			return nil
 		}},
+	{Scope: ScopeSession, Name: TiDBMemoryDebugMode, Value: strconv.Itoa(0), Type: TypeInt, MinValue: 0, MaxValue: 2, SetSession: func(s *SessionVars, val string) error {
+		s.MemoryDebugMode = TidbOptInt64(val, 0)
+		return nil
+	}},
 }
 
 // FeedbackProbability points to the FeedbackProbability in statistics package.

sessionctx/variable/tidb_vars.go

@@ -677,6 +677,9 @@ const (
 
 	// TiDBSimplifiedMetrics controls whether to unregister some unused metrics.
 	TiDBSimplifiedMetrics = "tidb_simplified_metrics"
+
+	// TiDBMemoryDebugMode is used to set tidb memory debug mode.
+	TiDBMemoryDebugMode = "tidb_memory_debug_mode"
 )
 
 // TiDB vars that have only global scope

util/memory/tracker.go

@@ -416,6 +416,20 @@ func (t *Tracker) SearchTrackerWithoutLock(label int) *Tracker {
 	return nil
 }
 
+// SearchTrackerConsumedMoreThanNBytes searches the specific tracker that consumes more than NBytes.
+func (t *Tracker) SearchTrackerConsumedMoreThanNBytes(limit int64) (res []*Tracker) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+	for _, sli := range t.mu.children {
+		for _, tracker := range sli {
+			if tracker.BytesConsumed() > limit {
+				res = append(res, tracker)
+			}
+		}
+	}
+	return
+}
+
 // String returns the string representation of this Tracker tree.
 func (t *Tracker) String() string {
 	buffer := bytes.NewBufferString("\n")