From a50765f1c98cf69d420edd944525a1aa20e2d4fa Mon Sep 17 00:00:00 2001 From: Bingyi Sun Date: Tue, 14 Jul 2026 17:48:36 +0800 Subject: [PATCH] enhance: reduce coordinator metadata lock contention (#51255) issue: #51333 ## Summary - serialize DataCoord checkpoint mutations per channel so unrelated channel catalog writes can proceed concurrently - keep overlapping single, batch, mark, and drop operations serialized with deterministic multi-channel lock ordering - use a read lock for the read-only RootCoord `DescribeAlias` path - exclude all newly added metrics and profiling instrumentation --------- Signed-off-by: sunby Co-authored-by: Claude Opus 4.6 --- internal/datacoord/meta.go | 65 +++++++--- internal/datacoord/meta_test.go | 169 ++++++++++++++++++++++++++ internal/rootcoord/meta_table.go | 4 +- internal/rootcoord/meta_table_test.go | 41 +++++++ 4 files changed, 263 insertions(+), 16 deletions(-) diff --git a/internal/datacoord/meta.go b/internal/datacoord/meta.go index 3c55987c69..294df99ddc 100644 --- a/internal/datacoord/meta.go +++ b/internal/datacoord/meta.go @@ -23,6 +23,7 @@ import ( "fmt" "math" "path" + "sort" "strconv" "strings" "time" @@ -138,13 +139,15 @@ func (m *meta) GetSnapshotMeta() *snapshotMeta { type channelCPs struct { lock.RWMutex - checkpoints map[string]*msgpb.MsgPosition - cond *syncutil.ContextCond + checkpoints map[string]*msgpb.MsgPosition + channelLocks *lock.KeyLock[string] + cond *syncutil.ContextCond } func newChannelCps() *channelCPs { cp := &channelCPs{ - checkpoints: make(map[string]*msgpb.MsgPosition), + checkpoints: make(map[string]*msgpb.MsgPosition), + channelLocks: lock.NewKeyLock[string](), } // use the same lock as channelCPs cp.cond = syncutil.NewContextCond(&cp.RWMutex) @@ -2722,16 +2725,20 @@ func (m *meta) UpdateChannelCheckpoint(ctx context.Context, vChannel string, pos pos = minGrowingCP } - m.channelCPs.Lock() - defer m.channelCPs.Unlock() + m.channelCPs.channelLocks.Lock(vChannel) + defer m.channelCPs.channelLocks.Unlock(vChannel) + m.channelCPs.RLock() oldPosition, ok := m.channelCPs.checkpoints[vChannel] + m.channelCPs.RUnlock() if !ok || oldPosition.Timestamp < pos.Timestamp || (oldPosition.Timestamp == pos.Timestamp && !bytes.Equal(oldPosition.MsgID, pos.MsgID)) { err := m.catalog.SaveChannelCheckpoint(ctx, vChannel, pos) if err != nil { return err } + m.channelCPs.Lock() m.channelCPs.checkpoints[vChannel] = pos + m.channelCPs.Unlock() ts, _ := tsoutil.ParseTS(pos.Timestamp) mlog.Info(context.TODO(), "UpdateChannelCheckpoint done", mlog.String("vChannel", vChannel), @@ -2750,8 +2757,8 @@ func (m *meta) UpdateChannelCheckpoint(ctx context.Context, vChannel string, pos // UpdateChannelCheckpoint can overwrite it, and removes the channel-checkpoint // lag metric. func (m *meta) MarkChannelCheckpointDropped(ctx context.Context, channel string) error { - m.channelCPs.Lock() - defer m.channelCPs.Unlock() + m.channelCPs.channelLocks.Lock(channel) + defer m.channelCPs.channelLocks.Unlock(channel) cp := &msgpb.MsgPosition{ ChannelName: channel, @@ -2763,7 +2770,9 @@ func (m *meta) MarkChannelCheckpointDropped(ctx context.Context, channel string) return err } + m.channelCPs.Lock() m.channelCPs.checkpoints[channel] = cp + m.channelCPs.Unlock() metrics.DataCoordCheckpointUnixSeconds.DeleteLabelValues(paramtable.GetStringNodeID(), channel) return nil @@ -2786,24 +2795,52 @@ func (m *meta) UpdateChannelCheckpoints(ctx context.Context, positions []*msgpb. } } - m.channelCPs.Lock() - defer m.channelCPs.Unlock() - toUpdates := lo.Filter(positions, func(pos *msgpb.MsgPosition, _ int) bool { + validPositions := lo.Filter(positions, func(pos *msgpb.MsgPosition, _ int) bool { if pos == nil || (pos.GetMsgID() == nil && pos.GetWALName() != commonpb.WALName_WoodPecker) || pos.GetChannelName() == "" { mlog.Warn(context.TODO(), "illegal channel cp", mlog.Any("pos", pos)) return false } + return true + }) + channelSet := make(map[string]struct{}, len(validPositions)) + for _, pos := range validPositions { + channelSet[pos.GetChannelName()] = struct{}{} + } + channels := make([]string, 0, len(channelSet)) + for channel := range channelSet { + channels = append(channels, channel) + } + sort.Strings(channels) + for _, channel := range channels { + m.channelCPs.channelLocks.Lock(channel) + } + defer func() { + for i := len(channels) - 1; i >= 0; i-- { + m.channelCPs.channelLocks.Unlock(channels[i]) + } + }() + + m.channelCPs.RLock() + toUpdates := lo.Filter(validPositions, func(pos *msgpb.MsgPosition, _ int) bool { vChannel := pos.GetChannelName() oldPosition, ok := m.channelCPs.checkpoints[vChannel] return !ok || oldPosition.Timestamp < pos.Timestamp || (oldPosition.Timestamp == pos.Timestamp && !bytes.Equal(oldPosition.MsgID, pos.MsgID)) }) + m.channelCPs.RUnlock() err := m.catalog.SaveChannelCheckpoints(ctx, toUpdates) if err != nil { return err } + m.channelCPs.Lock() for _, pos := range toUpdates { channel := pos.GetChannelName() m.channelCPs.checkpoints[channel] = pos + } + // broadcast the change of channel checkpoint for TruncateCollection op to drop segments + m.channelCPs.cond.UnsafeBroadcast() + m.channelCPs.Unlock() + for _, pos := range toUpdates { + channel := pos.GetChannelName() mlog.Info(context.TODO(), "UpdateChannelCheckpoint done", mlog.String("channel", channel), mlog.Stringer("walName", pos.WALName), mlog.Uint64("ts", pos.GetTimestamp()), @@ -2811,8 +2848,6 @@ func (m *meta) UpdateChannelCheckpoints(ctx context.Context, positions []*msgpb. ts, _ := tsoutil.ParseTS(pos.Timestamp) metrics.DataCoordCheckpointUnixSeconds.WithLabelValues(paramtable.GetStringNodeID(), channel).Set(float64(ts.Unix())) } - // broadcast the change of channel checkpoint for TruncateCollection op to drop segments - m.channelCPs.cond.UnsafeBroadcast() return nil } @@ -2827,13 +2862,15 @@ func (m *meta) GetChannelCheckpoint(vChannel string) *msgpb.MsgPosition { } func (m *meta) DropChannelCheckpoint(vChannel string) error { - m.channelCPs.Lock() - defer m.channelCPs.Unlock() + m.channelCPs.channelLocks.Lock(vChannel) + defer m.channelCPs.channelLocks.Unlock(vChannel) err := m.catalog.DropChannelCheckpoint(m.ctx, vChannel) if err != nil { return err } + m.channelCPs.Lock() delete(m.channelCPs.checkpoints, vChannel) + m.channelCPs.Unlock() metrics.DataCoordCheckpointUnixSeconds.DeleteLabelValues(paramtable.GetStringNodeID(), vChannel) mlog.Info(context.TODO(), "DropChannelCheckpoint done", mlog.String("vChannel", vChannel)) return nil diff --git a/internal/datacoord/meta_test.go b/internal/datacoord/meta_test.go index 3dcdede24d..7b66c54cee 100644 --- a/internal/datacoord/meta_test.go +++ b/internal/datacoord/meta_test.go @@ -4746,6 +4746,175 @@ func equalCollectionInfo(t *testing.T, a *collectionInfo, b *collectionInfo) { assert.Equal(t, a.StartPositions, b.StartPositions) } +func TestUpdateChannelCheckpoint_DifferentChannelsPersistConcurrently(t *testing.T) { + const ( + channel1 = "channel-1" + channel2 = "channel-2" + ) + + catalog := mocks2.NewDataCoordCatalog(t) + channel1Entered := make(chan struct{}) + releaseChannel1 := make(chan struct{}) + channel2Entered := make(chan struct{}) + catalog.EXPECT().SaveChannelCheckpoint(mock.Anything, channel1, mock.Anything). + RunAndReturn(func(context.Context, string, *msgpb.MsgPosition) error { + close(channel1Entered) + <-releaseChannel1 + return nil + }).Once() + catalog.EXPECT().SaveChannelCheckpoint(mock.Anything, channel2, mock.Anything). + RunAndReturn(func(context.Context, string, *msgpb.MsgPosition) error { + close(channel2Entered) + return nil + }).Once() + + meta := &meta{ + ctx: context.Background(), + catalog: catalog, + collections: typeutil.NewConcurrentMap[UniqueID, *collectionInfo](), + segments: NewSegmentsInfo(), + channelCPs: newChannelCps(), + } + channel1Done := make(chan error, 1) + go func() { + channel1Done <- meta.UpdateChannelCheckpoint(context.Background(), channel1, &msgpb.MsgPosition{ + ChannelName: channel1, + MsgID: []byte{1}, + Timestamp: 1, + }) + }() + + select { + case <-channel1Entered: + case <-time.After(3 * time.Second): + close(releaseChannel1) + select { + case <-channel1Done: + case <-time.After(3 * time.Second): + } + t.Fatal("channel-1 did not enter SaveChannelCheckpoint") + } + + channel2Done := make(chan error, 1) + go func() { + channel2Done <- meta.UpdateChannelCheckpoint(context.Background(), channel2, &msgpb.MsgPosition{ + ChannelName: channel2, + MsgID: []byte{2}, + Timestamp: 2, + }) + }() + + channel2PersistedConcurrently := false + select { + case <-channel2Entered: + channel2PersistedConcurrently = true + case <-time.After(3 * time.Second): + } + + close(releaseChannel1) + errs := make(map[string]error, 2) + for channel, done := range map[string]<-chan error{ + channel1: channel1Done, + channel2: channel2Done, + } { + select { + case err := <-done: + errs[channel] = err + case <-time.After(3 * time.Second): + assert.Failf(t, "checkpoint update did not finish", "channel: %s", channel) + } + } + for _, err := range errs { + assert.NoError(t, err) + } + require.True(t, channel2PersistedConcurrently, "channel-2 did not enter catalog while channel-1 was blocked") +} + +func TestUpdateChannelCheckpoints_SerializesWithSingleUpdateOnSameChannel(t *testing.T) { + const channel = "channel-1" + + catalog := mocks2.NewDataCoordCatalog(t) + batchEntered := make(chan struct{}) + releaseBatch := make(chan struct{}) + singleEntered := make(chan struct{}) + catalog.EXPECT().SaveChannelCheckpoints(mock.Anything, mock.Anything). + RunAndReturn(func(context.Context, []*msgpb.MsgPosition) error { + close(batchEntered) + <-releaseBatch + return nil + }).Once() + catalog.EXPECT().SaveChannelCheckpoint(mock.Anything, channel, mock.Anything). + RunAndReturn(func(context.Context, string, *msgpb.MsgPosition) error { + close(singleEntered) + return nil + }).Once() + + meta := &meta{ + ctx: context.Background(), + catalog: catalog, + collections: typeutil.NewConcurrentMap[UniqueID, *collectionInfo](), + segments: NewSegmentsInfo(), + channelCPs: newChannelCps(), + } + batchDone := make(chan error, 1) + go func() { + batchDone <- meta.UpdateChannelCheckpoints(context.Background(), []*msgpb.MsgPosition{{ + ChannelName: channel, + MsgID: []byte{1}, + Timestamp: 1, + }}) + }() + + select { + case <-batchEntered: + case <-time.After(3 * time.Second): + close(releaseBatch) + select { + case <-batchDone: + case <-time.After(3 * time.Second): + } + t.Fatal("batch update did not enter SaveChannelCheckpoints") + } + + singleStarted := make(chan struct{}) + singleDone := make(chan error, 1) + go func() { + close(singleStarted) + singleDone <- meta.UpdateChannelCheckpoint(context.Background(), channel, &msgpb.MsgPosition{ + ChannelName: channel, + MsgID: []byte{2}, + Timestamp: 2, + }) + }() + <-singleStarted + + singleEnteredBeforeBatchRelease := false + select { + case <-singleEntered: + singleEnteredBeforeBatchRelease = true + case <-time.After(2 * time.Second): + } + + close(releaseBatch) + errs := make([]error, 0, 2) + for operation, done := range map[string]<-chan error{ + "batch": batchDone, + "single": singleDone, + } { + select { + case err := <-done: + errs = append(errs, err) + case <-time.After(3 * time.Second): + assert.Failf(t, "checkpoint update did not finish", "operation: %s", operation) + } + } + for _, err := range errs { + assert.NoError(t, err) + } + assert.False(t, singleEnteredBeforeBatchRelease, "single update entered catalog before the overlapping batch completed") + require.Equal(t, uint64(2), meta.GetChannelCheckpoint(channel).GetTimestamp()) +} + func TestChannelCP(t *testing.T) { mockVChannel := "fake-by-dev-rootcoord-dml-1-testchannelcp-v0" mockPChannel := "fake-by-dev-rootcoord-dml-1" diff --git a/internal/rootcoord/meta_table.go b/internal/rootcoord/meta_table.go index a651e3f030..fdfa66a35b 100644 --- a/internal/rootcoord/meta_table.go +++ b/internal/rootcoord/meta_table.go @@ -1592,8 +1592,8 @@ func (mt *MetaTable) CheckIfAliasAlterable(ctx context.Context, dbName string, a } func (mt *MetaTable) DescribeAlias(ctx context.Context, dbName string, alias string, ts Timestamp) (string, error) { - mt.ddLock.Lock() - defer mt.ddLock.Unlock() + mt.ddLock.RLock() + defer mt.ddLock.RUnlock() if dbName == "" { mlog.Warn(ctx, "db name is empty", mlog.String("alias", alias)) diff --git a/internal/rootcoord/meta_table_test.go b/internal/rootcoord/meta_table_test.go index 6b6680f96a..5e556c4ae8 100644 --- a/internal/rootcoord/meta_table_test.go +++ b/internal/rootcoord/meta_table_test.go @@ -20,6 +20,7 @@ import ( "context" "math/rand" "testing" + "time" "github.com/cockroachdb/errors" "github.com/stretchr/testify/assert" @@ -51,6 +52,46 @@ import ( "github.com/milvus-io/milvus/pkg/v3/util/typeutil" ) +func TestMetaTable_DescribeAliasAllowsConcurrentReaders(t *testing.T) { + const ( + collectionID = int64(100) + collectionName = "test_metatable_describe_alias" + aliasName = "a_alias" + ) + meta := &MetaTable{ + collID2Meta: map[typeutil.UniqueID]*model.Collection{ + collectionID: { + CollectionID: collectionID, + Name: collectionName, + }, + }, + aliases: newNameDb(), + } + meta.aliases.insert(util.DefaultDBName, aliasName, collectionID) + + type result struct { + collectionName string + err error + } + resultCh := make(chan result, 1) + meta.ddLock.RLock() + go func() { + collectionName, err := meta.DescribeAlias(context.Background(), util.DefaultDBName, aliasName, 0) + resultCh <- result{collectionName: collectionName, err: err} + }() + + select { + case result := <-resultCh: + meta.ddLock.RUnlock() + require.NoError(t, result.err) + assert.Equal(t, collectionName, result.collectionName) + case <-time.After(3 * time.Second): + meta.ddLock.RUnlock() + <-resultCh + t.Fatal("DescribeAlias blocked behind another reader") + } +} + func generateMetaTable(_ *testing.T) *MetaTable { kv, _ := kvfactory.GetEtcdAndPath() path := funcutil.RandomString(10)