enhance: [2.4] Add go-deadlock as unittest only dependency (milvus-io…

…#33063) (milvus-io#34322) Cherry-pick from master pr: milvus-io#33063 See also milvus-io#33062 This PR: - Add lock.RWMutex & lock.Mutex alias to switch implementation based on build flags - When build flags has test in it, use go-deadlock to detect possible deadlocks - Replace all sync.RWMutex & sync.Mutex in datacoord pkg Signed-off-by: Congqi Xia <[email protected]>
wayblink · Jul 2, 2024 · 6b348e4 · 6b348e4
1 parent 9958606
commit 6b348e4
Show file tree

Hide file tree

Showing 26 changed files with 100 additions and 87 deletions.
diff --git a/go.mod b/go.mod
@@ -69,7 +69,6 @@ require (
 	github.com/milvus-io/milvus/pkg v0.0.0-00010101000000-000000000000
 	github.com/pkg/errors v0.9.1
 	github.com/valyala/fastjson v1.6.4
-	google.golang.org/protobuf v1.33.0
 	gopkg.in/yaml.v3 v3.0.1
 )
 
@@ -235,6 +234,7 @@ require (
 	google.golang.org/genproto v0.0.0-20230706204954-ccb25ca9f130 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20230629202037-9506855d4529 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240401170217-c3f982113cda // indirect
+	google.golang.org/protobuf v1.33.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect

diff --git a/internal/datacoord/analyze_meta.go b/internal/datacoord/analyze_meta.go
@@ -19,19 +19,19 @@ package datacoord
 import (
 	"context"
 	"fmt"
-	"sync"
 
 	"github.com/golang/protobuf/proto"
 	"go.uber.org/zap"
 
 	"github.com/milvus-io/milvus/internal/metastore"
 	"github.com/milvus-io/milvus/internal/proto/indexpb"
 	"github.com/milvus-io/milvus/pkg/log"
+	"github.com/milvus-io/milvus/pkg/util/lock"
 	"github.com/milvus-io/milvus/pkg/util/timerecord"
 )
 
 type analyzeMeta struct {
-	sync.RWMutex
+	lock.RWMutex
 
 	ctx     context.Context
 	catalog metastore.DataCoordCatalog

diff --git a/internal/datacoord/channel_manager.go b/internal/datacoord/channel_manager.go
@@ -19,7 +19,6 @@ package datacoord
 import (
 	"context"
 	"fmt"
-	"sync"
 	"time"
 
 	"github.com/samber/lo"
@@ -33,13 +32,14 @@ import (
 	"github.com/milvus-io/milvus/pkg/log"
 	"github.com/milvus-io/milvus/pkg/mq/msgstream"
 	"github.com/milvus-io/milvus/pkg/util/funcutil"
+	"github.com/milvus-io/milvus/pkg/util/lock"
 	"github.com/milvus-io/milvus/pkg/util/logutil"
 )
 
 // ChannelManagerImpl manages the allocation and the balance between channels and data nodes.
 type ChannelManagerImpl struct {
 	ctx              context.Context
-	mu               sync.RWMutex
+	mu               lock.RWMutex
 	h                Handler
 	store            RWChannelStore
 	factory          ChannelPolicyFactory

diff --git a/internal/datacoord/channel_manager_v2.go b/internal/datacoord/channel_manager_v2.go
@@ -31,6 +31,7 @@ import (
 	"github.com/milvus-io/milvus/pkg/kv"
 	"github.com/milvus-io/milvus/pkg/log"
 	"github.com/milvus-io/milvus/pkg/util/conc"
+	"github.com/milvus-io/milvus/pkg/util/lock"
 	"github.com/milvus-io/milvus/pkg/util/merr"
 	"github.com/milvus-io/milvus/pkg/util/typeutil"
 )
@@ -62,7 +63,7 @@ type SubCluster interface {
 
 type ChannelManagerImplV2 struct {
 	cancel context.CancelFunc
-	mu     sync.RWMutex
+	mu     lock.RWMutex
 	wg     sync.WaitGroup
 
 	h          Handler

diff --git a/internal/datacoord/cluster.go b/internal/datacoord/cluster.go
@@ -28,6 +28,7 @@ import (
 	"github.com/milvus-io/milvus/internal/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/log"
 	"github.com/milvus-io/milvus/pkg/util/commonpbutil"
+	"github.com/milvus-io/milvus/pkg/util/lock"
 	"github.com/milvus-io/milvus/pkg/util/paramtable"
 )
 
@@ -182,7 +183,7 @@ func (c *ClusterImpl) DropImport(nodeID int64, in *datapb.DropImportRequest) err
 func (c *ClusterImpl) QuerySlots() map[int64]int64 {
 	nodeIDs := c.sessionManager.GetSessionIDs()
 	nodeSlots := make(map[int64]int64)
-	mu := &sync.Mutex{}
+	mu := &lock.Mutex{}
 	wg := &sync.WaitGroup{}
 	for _, nodeID := range nodeIDs {
 		wg.Add(1)

diff --git a/internal/datacoord/compaction_task_meta.go b/internal/datacoord/compaction_task_meta.go
@@ -18,19 +18,19 @@ package datacoord
 
 import (
 	"context"
-	"sync"
 
 	"github.com/golang/protobuf/proto"
 	"go.uber.org/zap"
 
 	"github.com/milvus-io/milvus/internal/metastore"
 	"github.com/milvus-io/milvus/internal/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/log"
+	"github.com/milvus-io/milvus/pkg/util/lock"
 	"github.com/milvus-io/milvus/pkg/util/timerecord"
 )
 
 type compactionTaskMeta struct {
-	sync.RWMutex
+	lock.RWMutex
 	ctx     context.Context
 	catalog metastore.DataCoordCatalog
 	// currently only clustering compaction task is stored in persist meta
@@ -39,7 +39,7 @@ type compactionTaskMeta struct {
 
 func newCompactionTaskMeta(ctx context.Context, catalog metastore.DataCoordCatalog) (*compactionTaskMeta, error) {
 	csm := &compactionTaskMeta{
-		RWMutex:         sync.RWMutex{},
+		RWMutex:         lock.RWMutex{},
 		ctx:             ctx,
 		catalog:         catalog,
 		compactionTasks: make(map[int64]map[int64]*datapb.CompactionTask, 0),

diff --git a/internal/datacoord/compaction_trigger.go b/internal/datacoord/compaction_trigger.go
@@ -33,6 +33,7 @@ import (
 	"github.com/milvus-io/milvus/internal/proto/datapb"
 	"github.com/milvus-io/milvus/pkg/log"
 	"github.com/milvus-io/milvus/pkg/util/indexparamcheck"
+	"github.com/milvus-io/milvus/pkg/util/lock"
 	"github.com/milvus-io/milvus/pkg/util/logutil"
 	"github.com/milvus-io/milvus/pkg/util/paramtable"
 	"github.com/milvus-io/milvus/pkg/util/tsoutil"
@@ -75,7 +76,7 @@ type compactionTrigger struct {
 	signals           chan *compactionSignal
 	compactionHandler compactionPlanContext
 	globalTrigger     *time.Ticker
-	forceMu           sync.Mutex
+	forceMu           lock.Mutex
 	quit              chan struct{}
 	wg                sync.WaitGroup
 

diff --git a/internal/datacoord/garbage_collector_test.go b/internal/datacoord/garbage_collector_test.go
@@ -25,7 +25,6 @@ import (
 	"path"
 	"strconv"
 	"strings"
-	"sync"
 	"testing"
 	"time"
 
@@ -51,6 +50,7 @@ import (
 	"github.com/milvus-io/milvus/internal/storage"
 	"github.com/milvus-io/milvus/pkg/common"
 	"github.com/milvus-io/milvus/pkg/util/funcutil"
+	"github.com/milvus-io/milvus/pkg/util/lock"
 	"github.com/milvus-io/milvus/pkg/util/merr"
 	"github.com/milvus-io/milvus/pkg/util/paramtable"
 )
@@ -361,7 +361,7 @@ func createMetaForRecycleUnusedIndexes(catalog metastore.DataCoordCatalog) *meta
 		indexID = UniqueID(400)
 	)
 	return &meta{
-		RWMutex:      sync.RWMutex{},
+		RWMutex:      lock.RWMutex{},
 		ctx:          ctx,
 		catalog:      catalog,
 		collections:  nil,
@@ -476,7 +476,7 @@ func createMetaForRecycleUnusedSegIndexes(catalog metastore.DataCoordCatalog) *m
 		},
 	}
 	meta := &meta{
-		RWMutex:     sync.RWMutex{},
+		RWMutex:     lock.RWMutex{},
 		ctx:         ctx,
 		catalog:     catalog,
 		collections: nil,
@@ -641,7 +641,7 @@ func createMetaTableForRecycleUnusedIndexFiles(catalog *datacoord.Catalog) *meta
 		},
 	}
 	meta := &meta{
-		RWMutex:     sync.RWMutex{},
+		RWMutex:     lock.RWMutex{},
 		ctx:         ctx,
 		catalog:     catalog,
 		collections: nil,

diff --git a/internal/datacoord/import_meta.go b/internal/datacoord/import_meta.go
@@ -17,9 +17,8 @@
 package datacoord
 
 import (
-	"sync"
-
 	"github.com/milvus-io/milvus/internal/metastore"
+	"github.com/milvus-io/milvus/pkg/util/lock"
 )
 
 type ImportMeta interface {
@@ -37,7 +36,7 @@ type ImportMeta interface {
 }
 
 type importMeta struct {
-	mu    sync.RWMutex // guards jobs and tasks
+	mu    lock.RWMutex // guards jobs and tasks
 	jobs  map[int64]ImportJob
 	tasks map[int64]ImportTask
 

diff --git a/internal/datacoord/import_scheduler.go b/internal/datacoord/import_scheduler.go
@@ -31,6 +31,7 @@ import (
 	"github.com/milvus-io/milvus/internal/proto/internalpb"
 	"github.com/milvus-io/milvus/pkg/log"
 	"github.com/milvus-io/milvus/pkg/metrics"
+	"github.com/milvus-io/milvus/pkg/util/lock"
 )
 
 const (
@@ -147,7 +148,7 @@ func (s *importScheduler) peekSlots() map[int64]int64 {
 		return s.info.NodeID
 	})
 	nodeSlots := make(map[int64]int64)
-	mu := &sync.Mutex{}
+	mu := &lock.Mutex{}
 	wg := &sync.WaitGroup{}
 	for _, nodeID := range nodeIDs {
 		wg.Add(1)

diff --git a/internal/datacoord/index_engine_version_manager.go b/internal/datacoord/index_engine_version_manager.go
@@ -2,12 +2,12 @@ package datacoord
 
 import (
 	"math"
-	"sync"
 
 	"go.uber.org/zap"
 
 	"github.com/milvus-io/milvus/internal/util/sessionutil"
 	"github.com/milvus-io/milvus/pkg/log"
+	"github.com/milvus-io/milvus/pkg/util/lock"
 )
 
 type IndexEngineVersionManager interface {
@@ -21,7 +21,7 @@ type IndexEngineVersionManager interface {
 }
 
 type versionManagerImpl struct {
-	mu       sync.Mutex
+	mu       lock.Mutex
 	versions map[int64]sessionutil.IndexEngineVersion
 }
 

diff --git a/internal/datacoord/index_meta.go b/internal/datacoord/index_meta.go
@@ -21,7 +21,6 @@ import (
 	"context"
 	"fmt"
 	"strconv"
-	"sync"
 
 	"github.com/golang/protobuf/proto"
 	"github.com/prometheus/client_golang/prometheus"
@@ -35,12 +34,13 @@ import (
 	"github.com/milvus-io/milvus/pkg/common"
 	"github.com/milvus-io/milvus/pkg/log"
 	"github.com/milvus-io/milvus/pkg/metrics"
+	"github.com/milvus-io/milvus/pkg/util/lock"
 	"github.com/milvus-io/milvus/pkg/util/timerecord"
 	"github.com/milvus-io/milvus/pkg/util/typeutil"
 )
 
 type indexMeta struct {
-	sync.RWMutex
+	lock.RWMutex
 	ctx     context.Context
 	catalog metastore.DataCoordCatalog
 

diff --git a/internal/datacoord/index_meta_test.go b/internal/datacoord/index_meta_test.go
@@ -19,7 +19,6 @@ package datacoord
 
 import (
 	"context"
-	"sync"
 	"testing"
 
 	"github.com/cockroachdb/errors"
@@ -34,6 +33,7 @@ import (
 	"github.com/milvus-io/milvus/internal/metastore/model"
 	"github.com/milvus-io/milvus/internal/proto/indexpb"
 	"github.com/milvus-io/milvus/pkg/common"
+	"github.com/milvus-io/milvus/pkg/util/lock"
 )
 
 func TestReloadFromKV(t *testing.T) {
@@ -264,7 +264,7 @@ func TestMeta_HasSameReq(t *testing.T) {
 
 func newSegmentIndexMeta(catalog metastore.DataCoordCatalog) *indexMeta {
 	return &indexMeta{
-		RWMutex:              sync.RWMutex{},
+		RWMutex:              lock.RWMutex{},
 		ctx:                  context.Background(),
 		catalog:              catalog,
 		indexes:              make(map[UniqueID]map[UniqueID]*model.Index),

diff --git a/internal/datacoord/indexnode_manager.go b/internal/datacoord/indexnode_manager.go
@@ -28,6 +28,7 @@ import (
 	"github.com/milvus-io/milvus/internal/types"
 	"github.com/milvus-io/milvus/pkg/log"
 	"github.com/milvus-io/milvus/pkg/metrics"
+	"github.com/milvus-io/milvus/pkg/util/lock"
 	"github.com/milvus-io/milvus/pkg/util/merr"
 )
 
@@ -45,7 +46,7 @@ type WorkerManager interface {
 type IndexNodeManager struct {
 	nodeClients      map[UniqueID]types.IndexNodeClient
 	stoppingNodes    map[UniqueID]struct{}
-	lock             sync.RWMutex
+	lock             lock.RWMutex
 	ctx              context.Context
 	indexNodeCreator indexNodeCreatorFunc
 }
@@ -55,7 +56,7 @@ func NewNodeManager(ctx context.Context, indexNodeCreator indexNodeCreatorFunc)
 	return &IndexNodeManager{
 		nodeClients:      make(map[UniqueID]types.IndexNodeClient),
 		stoppingNodes:    make(map[UniqueID]struct{}),
-		lock:             sync.RWMutex{},
+		lock:             lock.RWMutex{},
 		ctx:              ctx,
 		indexNodeCreator: indexNodeCreator,
 	}
@@ -114,7 +115,7 @@ func (nm *IndexNodeManager) PickClient() (UniqueID, types.IndexNodeClient) {
 	ctx, cancel := context.WithCancel(nm.ctx)
 	var (
 		pickNodeID = UniqueID(0)
-		nodeMutex  = sync.Mutex{}
+		nodeMutex  = lock.Mutex{}
 		wg         = sync.WaitGroup{}
 	)
 
@@ -170,7 +171,7 @@ func (nm *IndexNodeManager) ClientSupportDisk() bool {
 	ctx, cancel := context.WithCancel(nm.ctx)
 	var (
 		enableDisk = false
-		nodeMutex  = sync.Mutex{}
+		nodeMutex  = lock.Mutex{}
 		wg         = sync.WaitGroup{}
 	)
 

diff --git a/internal/datacoord/indexnode_manager_test.go b/internal/datacoord/indexnode_manager_test.go
@@ -18,7 +18,6 @@ package datacoord
 
 import (
 	"context"
-	"sync"
 	"testing"
 
 	"github.com/cockroachdb/errors"
@@ -28,6 +27,7 @@ import (
 	"github.com/milvus-io/milvus/internal/mocks"
 	"github.com/milvus-io/milvus/internal/proto/indexpb"
 	"github.com/milvus-io/milvus/internal/types"
+	"github.com/milvus-io/milvus/pkg/util/lock"
 	"github.com/milvus-io/milvus/pkg/util/merr"
 )
 
@@ -108,7 +108,7 @@ func TestIndexNodeManager_ClientSupportDisk(t *testing.T) {
 	t.Run("support", func(t *testing.T) {
 		nm := &IndexNodeManager{
 			ctx:  context.Background(),
-			lock: sync.RWMutex{},
+			lock: lock.RWMutex{},
 			nodeClients: map[UniqueID]types.IndexNodeClient{
 				1: getMockedGetJobStatsClient(&indexpb.GetJobStatsResponse{
 					Status:     merr.Success(),
@@ -126,7 +126,7 @@ func TestIndexNodeManager_ClientSupportDisk(t *testing.T) {
 	t.Run("not support", func(t *testing.T) {
 		nm := &IndexNodeManager{
 			ctx:  context.Background(),
-			lock: sync.RWMutex{},
+			lock: lock.RWMutex{},
 			nodeClients: map[UniqueID]types.IndexNodeClient{
 				1: getMockedGetJobStatsClient(&indexpb.GetJobStatsResponse{
 					Status:     merr.Success(),
@@ -144,7 +144,7 @@ func TestIndexNodeManager_ClientSupportDisk(t *testing.T) {
 	t.Run("no indexnode", func(t *testing.T) {
 		nm := &IndexNodeManager{
 			ctx:         context.Background(),
-			lock:        sync.RWMutex{},
+			lock:        lock.RWMutex{},
 			nodeClients: map[UniqueID]types.IndexNodeClient{},
 		}
 
@@ -155,7 +155,7 @@ func TestIndexNodeManager_ClientSupportDisk(t *testing.T) {
 	t.Run("error", func(t *testing.T) {
 		nm := &IndexNodeManager{
 			ctx:  context.Background(),
-			lock: sync.RWMutex{},
+			lock: lock.RWMutex{},
 			nodeClients: map[UniqueID]types.IndexNodeClient{
 				1: getMockedGetJobStatsClient(nil, err),
 			},
@@ -168,7 +168,7 @@ func TestIndexNodeManager_ClientSupportDisk(t *testing.T) {
 	t.Run("fail reason", func(t *testing.T) {
 		nm := &IndexNodeManager{
 			ctx:  context.Background(),
-			lock: sync.RWMutex{},
+			lock: lock.RWMutex{},
 			nodeClients: map[UniqueID]types.IndexNodeClient{
 				1: getMockedGetJobStatsClient(&indexpb.GetJobStatsResponse{
 					Status:     merr.Status(err),