executor: implemented exclusive lock for temp storage dir per tidb instance (#15203)

config/config.go

@@ -16,6 +16,7 @@ package config
 import (
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
@@ -44,6 +45,10 @@ const (
 	DefMaxIndexLength = 3072
 	// DefMaxOfMaxIndexLength is the maximum index length(in bytes) for TiDB v3.0.7 and previous version.
 	DefMaxOfMaxIndexLength = 3072 * 4
+	// DefPort is the default port of TiDB
+	DefPort = 4000
+	// DefStatusPort is the default status port of TiBD
+	DefStatusPort = 10080
 )
 
 // Valid config maps
@@ -56,6 +61,8 @@ var (
 	CheckTableBeforeDrop = false
 	// checkBeforeDropLDFlag is a go build flag.
 	checkBeforeDropLDFlag = "None"
+	// tempStorageDirName is the default temporary storage dir name by base64 encoding a string `port/statusPort`
+	tempStorageDirName = encodeDefTempStorageDir(DefPort, DefStatusPort)
 )
 
 // Config contains configuration options.
@@ -123,6 +130,19 @@ type Config struct {
 	EnableDynamicConfig bool `toml:"enable-dynamic-config" json:"enable-dynamic-config"`
 }
 
+// UpdateTempStoragePath is to update the `TempStoragePath` if port/statusPort was changed
+// and the `tmp-storage-path` was not specified in the conf.toml or was specified the same as the default value.
+func (c *Config) UpdateTempStoragePath() {
+	if c.TempStoragePath == tempStorageDirName {
+		c.TempStoragePath = encodeDefTempStorageDir(c.Port, c.Status.StatusPort)
+	}
+}
+
+func encodeDefTempStorageDir(port, statusPort uint) string {
+	dirName := base64.URLEncoding.EncodeToString([]byte(fmt.Sprintf("%v/%v", port, statusPort)))
+	return filepath.Join(os.TempDir(), "tidb", dirName, "tmp-storage")
+}
+
 // nullableBool defaults unset bool options to unset instead of false, which enables us to know if the user has set 2
 // conflict options at the same time.
 type nullableBool struct {
@@ -503,7 +523,7 @@ type Experimental struct {
 var defaultConf = Config{
 	Host:                         "0.0.0.0",
 	AdvertiseAddress:             "",
-	Port:                         4000,
+	Port:                         DefPort,
 	Cors:                         "",
 	Store:                        "mocktikv",
 	Path:                         "/tmp/tidb",
@@ -512,7 +532,7 @@ var defaultConf = Config{
 	Lease:                        "45s",
 	TokenLimit:                   1000,
 	OOMUseTmpStorage:             true,
-	TempStoragePath:              filepath.Join(os.TempDir(), "tidb", "tmp-storage"),
+	TempStoragePath:              tempStorageDirName,
 	OOMAction:                    OOMActionCancel,
 	MemQuotaQuery:                1 << 30,
 	EnableStreaming:              false,
@@ -551,7 +571,7 @@ var defaultConf = Config{
 	Status: Status{
 		ReportStatus:    true,
 		StatusHost:      "0.0.0.0",
-		StatusPort:      10080,
+		StatusPort:      DefStatusPort,
 		MetricsInterval: 15,
 		RecordQPSbyDB:   false,
 	},

config/config.toml.example

@@ -38,9 +38,9 @@ mem-quota-query = 1073741824
 oom-use-tmp-storage = true
 
 # Specifies the temporary storage path for some operators when a single SQL statement exceeds the memory quota specified by mem-quota-query.
-# It defaults to `<TMPDIR>/tidb/tmp-storage` if it is unset.
+# It defaults to a generated directory in `<TMPDIR>/tidb/` if it is unset.
 # It only takes effect when `oom-use-tmp-storage` is `true`.
-# tmp-storage-path = "/tmp/tidb/tmp-storage"
+# tmp-storage-path = "/tmp/tidb/NDAwMC8xMDA4MA==/tmp-storage"
 
 # Specifies what operation TiDB performs when a single SQL statement exceeds the memory quota specified by mem-quota-query and cannot be spilled over to disk.
 # Valid options: ["log", "cancel"]

config/config_test.go

@@ -148,7 +148,7 @@ disable-error-stack = false
 
 func (s *testConfigSuite) TestConfig(c *C) {
 	conf := new(Config)
-	conf.TempStoragePath = filepath.Join(os.TempDir(), "tidb", "tmp-storage")
+	conf.TempStoragePath = tempStorageDirName
 	conf.Binlog.Enable = true
 	conf.Binlog.IgnoreError = true
 	conf.Binlog.Strategy = "hash"

executor/executor_test.go

@@ -90,7 +90,9 @@ func TestT(t *testing.T) {
 	new.Log.SlowThreshold = 30000 // 30s
 	new.Experimental.AllowsExpressionIndex = true
 	config.StoreGlobalConfig(&new)
-
+	tmpDir := config.GetGlobalConfig().TempStoragePath
+	_ = os.RemoveAll(tmpDir) // clean the uncleared temp file during the last run.
+	_ = os.MkdirAll(tmpDir, 0755)
 	testleak.BeforeTest()
 	TestingT(t)
 	testleak.AfterTestT(t)()

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f // indirect
 	github.com/cznic/mathutil v0.0.0-20181122101859-297441e03548
 	github.com/cznic/sortutil v0.0.0-20181122101858-f5f958428db8
+	github.com/danjacques/gofslock v0.0.0-20191023191349-0a45f885bc37
 	github.com/dgraph-io/ristretto v0.0.1
 	github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2
 	github.com/dustin/go-humanize v1.0.0 // indirect

go.sum

@@ -50,6 +50,8 @@ github.com/cznic/sortutil v0.0.0-20181122101858-f5f958428db8 h1:LpMLYGyy67BoAFGd
 github.com/cznic/sortutil v0.0.0-20181122101858-f5f958428db8/go.mod h1:q2w6Bg5jeox1B+QkJ6Wp/+Vn0G/bo3f1uY7Fn3vivIQ=
 github.com/cznic/strutil v0.0.0-20171016134553-529a34b1c186/go.mod h1:AHHPPPXTw0h6pVabbcbyGRK1DckRn7r/STdZEeIDzZc=
 github.com/cznic/y v0.0.0-20170802143616-045f81c6662a/go.mod h1:1rk5VM7oSnA4vjp+hrLQ3HWHa+Y4yPCa3/CsJrcNnvs=
+github.com/danjacques/gofslock v0.0.0-20191023191349-0a45f885bc37 h1:X6mKGhCFOxrKeeHAjv/3UvT6e5RRxW6wRdlqlV6/H4w=
+github.com/danjacques/gofslock v0.0.0-20191023191349-0a45f885bc37/go.mod h1:DC3JtzuG7kxMvJ6dZmf2ymjNyoXwgtklr7FN+Um2B0U=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=

tidb-server/main.go

@@ -19,12 +19,14 @@ import (
 	"fmt"
 	"io/ioutil"
 	"os"
+	"path/filepath"
 	"runtime"
 	"strconv"
 	"strings"
 	"sync/atomic"
 	"time"
 
+	"github.com/danjacques/gofslock/fslock"
 	"github.com/opentracing/opentracing-go"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
@@ -146,10 +148,11 @@ var (
 )
 
 var (
-	storage  kv.Storage
-	dom      *domain.Domain
-	svr      *server.Server
-	graceful bool
+	storage     kv.Storage
+	dom         *domain.Domain
+	svr         *server.Server
+	tempDirLock fslock.Handle
+	graceful    bool
 )
 
 func main() {
@@ -161,6 +164,10 @@ func main() {
 	registerStores()
 	registerMetrics()
 	config.InitializeConfig(*configPath, *configCheck, *configStrict, reloadConfig, overrideConfig)
+	if config.GetGlobalConfig().OOMUseTmpStorage {
+		config.GetGlobalConfig().UpdateTempStoragePath()
+		initializeTempDir()
+	}
 	setGlobalVars()
 	setCPUAffinity()
 	setupLog()
@@ -188,6 +195,42 @@ func syncLog() {
 	}
 }
 
+func initializeTempDir() {
+	tempDir := config.GetGlobalConfig().TempStoragePath
+	lockFile := "_dir.lock"
+	_, err := os.Stat(tempDir)
+	if err != nil && !os.IsExist(err) {
+		err = os.MkdirAll(tempDir, 0755)
+		terror.MustNil(err)
+	}
+	tempDirLock, err = fslock.Lock(filepath.Join(tempDir, lockFile))
+	if err != nil {
+		switch err {
+		case fslock.ErrLockHeld:
+			fmt.Fprintf(os.Stderr, "The current temporary storage dir(%s) has been occupied by another instance, "+
+				"check tmp-storage-path config and make sure they are different.", tempDir)
+		default:
+			fmt.Fprintf(os.Stderr, "Failed to acquire exclusive lock on the temporary storage dir(%s). Error detail: {%s}", tempDir, err)
+		}
+		os.Exit(1)
+	}
+
+	subDirs, err := ioutil.ReadDir(tempDir)
+	terror.MustNil(err)
+
+	for _, subDir := range subDirs {
+		// Do not remove the lock file.
+		if subDir.Name() == lockFile {
+			continue
+		}
+		err = os.RemoveAll(filepath.Join(tempDir, subDir.Name()))
+		if err != nil {
+			log.Warn("Remove temporary file error",
+				zap.String("tempStorageSubDir", filepath.Join(tempDir, subDir.Name())), zap.Error(err))
+		}
+	}
+}
+
 func setCPUAffinity() {
 	if affinityCPU == nil || len(*affinityCPU) == 0 {
 		return
@@ -630,6 +673,10 @@ func cleanup() {
 	}
 	plugin.Shutdown(context.Background())
 	closeDomainAndStorage()
+	if tempDirLock != nil {
+		err := tempDirLock.Unlock()
+		terror.Log(errors.Trace(err))
+	}
 }
 
 func stringToList(repairString string) []string {

util/chunk/chunk_test.go

@@ -17,6 +17,7 @@ import (
 	"bytes"
 	"fmt"
 	"math"
+	"os"
 	"strconv"
 	"strings"
 	"sync"
@@ -38,6 +39,8 @@ func TestT(t *testing.T) {
 	conf := *cfg
 	conf.TempStoragePath = "/tmp/tidb/test-temp-storage"
 	config.StoreGlobalConfig(&conf)
+	_ = os.RemoveAll(conf.TempStoragePath) // clean the uncleared temp file during the last run.
+	_ = os.MkdirAll(conf.TempStoragePath, 0755)
 	check.TestingT(t)
 }
 

util/chunk/disk.go

@@ -22,13 +22,11 @@ import (
 	"os"
 	"sync"
 
-	"github.com/pingcap/log"
 	"github.com/pingcap/parser/terror"
 	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util/disk"
 	"github.com/pingcap/tidb/util/stringutil"
-	"go.uber.org/zap"
 )
 
 const (
@@ -44,23 +42,6 @@ var bufReaderPool = sync.Pool{
 	New: func() interface{} { return bufio.NewReaderSize(nil, readBufSize) },
 }
 
-// initTempDirOnce is used for cleaning the temporary disk storage directory once and only once.
-var initTempDirOnce = &sync.Once{}
-
-func initializeTempDir() {
-	initTempDirOnce.Do(func() {
-		tmpDir := config.GetGlobalConfig().TempStoragePath
-		err := os.RemoveAll(tmpDir) // clean the uncleared temp file during the last run.
-		if err != nil {
-			log.Warn("Remove temporary file error", zap.String("tmpDir", tmpDir), zap.Error(err))
-		}
-		err = os.MkdirAll(tmpDir, 0755)
-		if err != nil {
-			log.Warn("Mkdir temporary file error", zap.String("tmpDir", tmpDir), zap.Error(err))
-		}
-	})
-}
-
 // ListInDisk represents a slice of chunks storing in temporary disk.
 type ListInDisk struct {
 	fieldTypes []*types.FieldType
@@ -90,7 +71,6 @@ func NewListInDisk(fieldTypes []*types.FieldType) *ListInDisk {
 }
 
 func (l *ListInDisk) initDiskFile() (err error) {
-	initializeTempDir()
 	l.disk, err = ioutil.TempFile(config.GetGlobalConfig().TempStoragePath, l.diskTracker.Label().String())
 	if err != nil {
 		return