Fix/datarace when concurrent save the same file (#836)

Makefile

@@ -10,7 +10,7 @@ GOARCH  := $(if $(GOARCH),$(GOARCH),amd64)
 GOENV   := GO111MODULE=on CGO_ENABLED=0 GOOS=$(GOOS) GOARCH=$(GOARCH)
 GO      := $(GOENV) go
 GOBUILD := $(GO) build $(BUILD_FLAG)
-GOTEST  := GO111MODULE=on CGO_ENABLED=1 $(GO) test -p 3
+GOTEST  := GO111MODULE=on CGO_ENABLED=1 go test -p 3
 SHELL   := /usr/bin/env bash
 
 COMMIT    := $(shell git describe --no-match --always --dirty)
@@ -93,6 +93,10 @@ unit-test:
 	mkdir -p cover
 	TIUP_HOME=$(shell pwd)/tests/tiup $(GOTEST) ./... -covermode=count -coverprofile cover/cov.unit-test.out
 
+race: failpoint-enable
+	TIUP_HOME=$(shell pwd)/tests/tiup $(GOTEST) -race ./...  || { $(FAILPOINT_DISABLE); exit 1; }
+	@$(FAILPOINT_DISABLE)
+
 failpoint-enable: tools/bin/failpoint-ctl
 	@$(FAILPOINT_ENABLE)
 

pkg/cluster/task/builder.go

@@ -384,7 +384,7 @@ func (b *Builder) DeploySpark(inst spec.Instance, sparkVersion, srcPath, deployD
 	)
 }
 
-// TLSCert geenrates certificate for instance and transfer it to the server
+// TLSCert generates certificate for instance and transfers it to the server
 func (b *Builder) TLSCert(inst spec.Instance, ca *crypto.CertificateAuthority, paths meta.DirPaths) *Builder {
 	b.tasks = append(b.tasks, &TLSCert{
 		ca:    ca,

pkg/file/file.go

@@ -5,11 +5,17 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/pingcap/errors"
 )
 
+var (
+	fileLocks = make(map[string]*sync.Mutex)
+	filesLock = sync.Mutex{}
+)
+
 // SaveFileWithBackup will backup the file before save it.
 // e.g., backup meta.yaml as meta-2006-01-02T15:04:05Z07:00.yaml
 // backup the files in the same dir of path if backupDir is empty.
@@ -25,6 +31,15 @@ func SaveFileWithBackup(path string, data []byte, backupDir string) error {
 		return errors.Errorf("%s is directory", path)
 	}
 
+	filesLock.Lock()
+	defer filesLock.Unlock()
+	if _, ok := fileLocks[path]; !ok {
+		fileLocks[path] = &sync.Mutex{}
+	}
+
+	fileLocks[path].Lock()
+	defer fileLocks[path].Unlock()
+
 	// backup file
 	if !os.IsNotExist(err) {
 		base := filepath.Base(path)

pkg/file/file_test.go

@@ -1,13 +1,17 @@
 package file
 
 import (
+	"bytes"
 	"io/ioutil"
+	"math/rand"
 	"os"
 	"path/filepath"
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/pingcap/check"
 )
@@ -78,3 +82,42 @@ func (s *fileSuite) TestSaveFileWithBackup(c *check.C) {
 	c.Assert(err, check.IsNil)
 	c.Assert(string(data), check.Equals, "9")
 }
+
+func (s *fileSuite) TestConcurrentSaveFileWithBackup(c *check.C) {
+	dir := c.MkDir()
+	name := "meta.yaml"
+	data := []byte("concurrent-save-file-with-backup")
+
+	rand.Seed(time.Now().UnixNano())
+	var wg sync.WaitGroup
+	for i := 0; i < 10; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			time.Sleep(time.Duration(rand.Intn(100)+4) * time.Millisecond)
+			err := SaveFileWithBackup(filepath.Join(dir, name), data, "")
+			c.Assert(err, check.IsNil)
+		}()
+	}
+
+	wg.Wait()
+
+	// Verify the saved files.
+	var paths []string
+	err := filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
+		// simply filter the not relate files.
+		if strings.Contains(path, "meta") {
+			paths = append(paths, path)
+		}
+		return nil
+	})
+
+	c.Assert(err, check.IsNil)
+	c.Assert(len(paths), check.Equals, 10)
+	for _, path := range paths {
+		body, err := ioutil.ReadFile(path)
+		c.Assert(err, check.IsNil)
+		c.Assert(len(body), check.Equals, len(data))
+		c.Assert(bytes.Equal(body, data), check.IsTrue)
+	}
+}