Fix data race in processlist map (#1884)

Author: jycor

enginetest/engine_only_test.go

@@ -333,6 +333,10 @@ func TestTrackProcess(t *testing.T) {
 	require.Error(ctx.Err())
 }
 
+func TestConcurrentProcessList(t *testing.T) {
+	enginetest.TestConcurrentProcessList(t, enginetest.NewDefaultMemoryHarness())
+}
+
 func getRuleFrom(rules []analyzer.Rule, id analyzer.RuleId) *analyzer.Rule {
 	for _, rule := range rules {
 		if rule.Id == id {

enginetest/enginetests.go

@@ -20,6 +20,7 @@ import (
 	"io"
 	"net"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -4945,6 +4946,69 @@ func TestTransactionScripts(t *testing.T, harness Harness) {
 	}
 }
 
+func TestConcurrentProcessList(t *testing.T, harness Harness) {
+	require := require.New(t)
+	pl := sqle.NewProcessList()
+	numSessions := 2
+
+	for i := 0; i < numSessions; i++ {
+		pl.AddConnection(uint32(i), "foo")
+		sess := sql.NewBaseSessionWithClientServer("0.0.0.0:3306", sql.Client{Address: "", User: ""}, uint32(i))
+		pl.ConnectionReady(sess)
+
+		var err error
+		ctx := sql.NewContext(context.Background(), sql.WithPid(uint64(i)), sql.WithSession(sess), sql.WithProcessList(pl))
+		_, err = pl.BeginQuery(ctx, "foo")
+		require.NoError(err)
+	}
+
+	var wg sync.WaitGroup
+
+	// Read concurrently
+	for i := 0; i < numSessions; i++ {
+		wg.Add(1)
+		go func(x int) {
+			defer wg.Done()
+			procs := pl.Processes()
+			for _, proc := range procs {
+				for prog, part := range proc.Progress {
+					if prog == "" {
+					}
+					for p, pp := range part.PartitionsProgress {
+						if p == "" {
+						}
+						if pp.Name == "" {
+						}
+					}
+				}
+			}
+		}(i)
+	}
+
+	// Writes concurrently
+	for i := 0; i < numSessions; i++ {
+		wg.Add(4)
+		go func(x int) {
+			defer wg.Done()
+			pl.AddTableProgress(uint64(x), "foo", 100)
+		}(i)
+		go func(x int) {
+			defer wg.Done()
+			pl.AddPartitionProgress(uint64(x), "foo", "bar", 100)
+		}(i)
+		go func(x int) {
+			defer wg.Done()
+			pl.UpdateTableProgress(uint64(x), "foo", 100)
+		}(i)
+		go func(x int) {
+			defer wg.Done()
+			pl.UpdatePartitionProgress(uint64(x), "foo", "bar", 100)
+		}(i)
+	}
+
+	wg.Wait()
+}
+
 func TestNoDatabaseSelected(t *testing.T, harness Harness) {
 	harness.Setup(setup.MydbData)
 	e := mustNewEngine(t, harness)

processlist.go

@@ -47,12 +47,21 @@ func (pl *ProcessList) Processes() []sql.Process {
 	defer pl.mu.RUnlock()
 	var result = make([]sql.Process, 0, len(pl.procs))
 
+	// Make a deep copy of all maps to avoid race
 	for _, proc := range pl.procs {
 		p := *proc
-		var progress = make(map[string]sql.TableProgress, len(p.Progress))
-		for n, p := range p.Progress {
-			progress[n] = p
+		var progMap = make(map[string]sql.TableProgress, len(p.Progress))
+		for progName, prog := range p.Progress {
+			newProg := sql.TableProgress{
+				Progress:           prog.Progress,
+				PartitionsProgress: make(map[string]sql.PartitionProgress, len(prog.PartitionsProgress)),
+			}
+			for partName, partProg := range prog.PartitionsProgress {
+				newProg.PartitionsProgress[partName] = partProg
+			}
+			progMap[progName] = newProg
 		}
+		p.Progress = progMap
 		result = append(result, p)
 	}