fix: avoid rw datarace on CompactionTask

Signed-off-by: zhenshan.cao <[email protected]>
czs007 · Jul 16, 2024 · 169a855 · 169a855
1 parent 804dd54
commit 169a855
Show file tree

Hide file tree

Showing 10 changed files with 449 additions and 398 deletions.
diff --git a/internal/datacoord/compaction.go b/internal/datacoord/compaction.go
@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/cockroachdb/errors"
+	"github.com/golang/protobuf/proto"
 	"github.com/samber/lo"
 	"go.opentelemetry.io/otel"
 	"go.uber.org/atomic"
@@ -551,7 +552,6 @@ func (c *compactionPlanHandler) enqueueCompaction(task *datapb.CompactionTask) e
 	if err != nil {
 		return err
 	}
-	t.SetTask(t.ShadowClone(setStartTime(time.Now().Unix())))
 	err = t.SaveTaskMeta()
 	if err != nil {
 		c.meta.SetSegmentsCompacting(t.GetInputSegments(), false)
@@ -563,24 +563,26 @@ func (c *compactionPlanHandler) enqueueCompaction(task *datapb.CompactionTask) e
 }
 
 // set segments compacting, one segment can only participate one compactionTask
-func (c *compactionPlanHandler) createCompactTask(t *datapb.CompactionTask) (CompactionTask, error) {
+func (c *compactionPlanHandler) createCompactTask(pb *datapb.CompactionTask) (CompactionTask, error) {
 	var task CompactionTask
-	switch t.GetType() {
+	pbCloned := proto.Clone(pb).(*datapb.CompactionTask)
+	pbCloned.StartTime = time.Now().Unix()
+	switch pb.GetType() {
 	case datapb.CompactionType_MixCompaction:
 		task = &mixCompactionTask{
-			CompactionTask: t,
-			meta:           c.meta,
-			sessions:       c.sessions,
+			pb:       pbCloned,
+			meta:     c.meta,
+			sessions: c.sessions,
 		}
 	case datapb.CompactionType_Level0DeleteCompaction:
 		task = &l0CompactionTask{
-			CompactionTask: t,
-			meta:           c.meta,
-			sessions:       c.sessions,
+			pb:       pbCloned,
+			meta:     c.meta,
+			sessions: c.sessions,
 		}
 	case datapb.CompactionType_ClusteringCompaction:
 		task = &clusteringCompactionTask{
-			CompactionTask:   t,
+			pb:               pbCloned,
 			meta:             c.meta,
 			sessions:         c.sessions,
 			handler:          c.handler,
@@ -589,7 +591,7 @@ func (c *compactionPlanHandler) createCompactTask(t *datapb.CompactionTask) (Com
 	default:
 		return nil, merr.WrapErrIllegalCompactionPlan("illegal compaction type")
 	}
-	exist, succeed := c.meta.CheckAndSetSegmentsCompacting(t.GetInputSegments())
+	exist, succeed := c.meta.CheckAndSetSegmentsCompacting(pbCloned.GetInputSegments())
 	if !exist {
 		return nil, merr.WrapErrIllegalCompactionPlan("segment not exist")
 	}

diff --git a/internal/datacoord/compaction_task.go b/internal/datacoord/compaction_task.go
@@ -46,13 +46,12 @@ type CompactionTask interface {
 
 	GetNodeID() UniqueID
 	GetSpan() trace.Span
-	ShadowClone(opts ...compactionTaskOpt) *datapb.CompactionTask
+	CloneTaskPB(opts ...compactionTaskOpt) *datapb.CompactionTask
+	GetTaskPB() *datapb.CompactionTask
 	SetNodeID(UniqueID) error
-	SetTask(*datapb.CompactionTask)
 	SetSpan(trace.Span)
 	SetResult(*datapb.CompactionPlanResult)
 	EndSpan()
-	CleanLogPath()
 	NeedReAssignNodeID() bool
 	SaveTaskMeta() error
 }
@@ -89,6 +88,12 @@ func setResultSegments(segments []int64) compactionTaskOpt {
 	}
 }
 
+func setAnalyzeVersion(version int64) compactionTaskOpt {
+	return func(task *datapb.CompactionTask) {
+		task.AnalyzeVersion = version
+	}
+}
+
 func setState(state datapb.CompactionTaskState) compactionTaskOpt {
 	return func(task *datapb.CompactionTask) {
 		task.State = state