diff --git a/blobstore/clustermgr/catalog/catalog.go b/blobstore/clustermgr/catalog/catalog.go index 9acdd1383..b73e5f644 100644 --- a/blobstore/clustermgr/catalog/catalog.go +++ b/blobstore/clustermgr/catalog/catalog.go @@ -61,12 +61,13 @@ var ( ) type CatalogMgr struct { - module string - allShards *concurrentShards - allSpaces *concurrentSpaces - dirty atomic.Value - initShardDone int32 - applyTaskPool *base.TaskDistribution + module string + allShards *concurrentShards + allSpaces *concurrentSpaces + dirty atomic.Value + initShardDone int32 + actualInitShardNum int // actual range count generated by InitShardingRange, may exceed InitShardNum due to power-of-2 rounding + applyTaskPool *base.TaskDistribution catalogTbl *catalogdb.CatalogTable transitedTbl *catalogdb.TransitedTable diff --git a/blobstore/clustermgr/catalog/createshard.go b/blobstore/clustermgr/catalog/createshard.go index 8e43e72bc..753b7fcbf 100644 --- a/blobstore/clustermgr/catalog/createshard.go +++ b/blobstore/clustermgr/catalog/createshard.go @@ -43,12 +43,15 @@ func (c *CatalogMgr) createShard(ctx context.Context) error { span := trace.SpanFromContextSafe(ctx) // alloc shardID and shard range, and save initial shard and unit info into transited tbl - createShardCtxs := make([]createShardCtx, 0, c.InitShardNum) - minShardID, _, err := c.scopeMgr.Alloc(ctx, shardIDScopeName, c.InitShardNum) + ranges := sharding.InitShardingRange(sharding.RangeType_RangeTypeHash, 2, c.InitShardNum) + if len(ranges) != c.InitShardNum { + span.Warnf("InitShardNum %d expanded to %d due to sharding algorithm power-of-2 rounding", c.InitShardNum, len(ranges)) + } + createShardCtxs := make([]createShardCtx, 0, len(ranges)) + minShardID, _, err := c.scopeMgr.Alloc(ctx, shardIDScopeName, len(ranges)) if err != nil { return errors.Info(err, "scope alloc shardID failed") } - ranges := sharding.InitShardingRange(sharding.RangeType_RangeTypeHash, 2, c.InitShardNum) unitCount := c.CodeMode.GetShardNum() for i := 0; i < len(ranges); i++ { shardID := proto.ShardID(minShardID + uint64(i)) @@ -234,7 +237,7 @@ func (c *CatalogMgr) applyCreateShard(ctx context.Context, shard *shardItem) err c.routeMgr.InsertRouteItems(ctx, []*base.RouteItem{route}) shard.info.RouteVersion = proto.RouteVersion(routeVersion) - if c.allShards.getShardNum() == c.InitShardNum { + if c.allShards.getShardNum() == c.actualInitShardNum { err := c.kvMgr.Set(proto.ShardInitDoneKey, []byte("1")) if err != nil { return errors.Info(err, "put shard init done key to kv failed") diff --git a/blobstore/clustermgr/catalog/createshard_test.go b/blobstore/clustermgr/catalog/createshard_test.go index cc16e7ce6..8c9b9b756 100644 --- a/blobstore/clustermgr/catalog/createshard_test.go +++ b/blobstore/clustermgr/catalog/createshard_test.go @@ -17,7 +17,12 @@ package catalog import ( "context" "encoding/json" + "fmt" + "math/rand" + "os" + "path" "testing" + "time" "github.com/golang/mock/gomock" "github.com/stretchr/testify/require" @@ -25,9 +30,11 @@ import ( "github.com/cubefs/cubefs/blobstore/api/clustermgr" "github.com/cubefs/cubefs/blobstore/clustermgr/base" "github.com/cubefs/cubefs/blobstore/clustermgr/cluster" + "github.com/cubefs/cubefs/blobstore/clustermgr/persistence/catalogdb" "github.com/cubefs/cubefs/blobstore/common/codemode" "github.com/cubefs/cubefs/blobstore/common/proto" "github.com/cubefs/cubefs/blobstore/common/raftserver" + "github.com/cubefs/cubefs/blobstore/common/sharding" "github.com/cubefs/cubefs/blobstore/common/trace" mock "github.com/cubefs/cubefs/blobstore/testing/mockclustermgr" "github.com/cubefs/cubefs/blobstore/testing/mocks" @@ -192,3 +199,91 @@ func TestCatalogMgr_applyCreateShard(t *testing.T) { require.NoError(t, err) require.Equal(t, expectedRouteVersion, shardRecord.RouteVersion) } + +// TestCatalogMgr_applyCreateShard_InitShardDone verifies that initShardDone is set only +// when the number of applied shards reaches actualInitShardNum (the real range count produced +// by InitShardingRange), not InitShardNum which may be smaller due to power-of-2 rounding. +func TestCatalogMgr_applyCreateShard_InitShardDone(t *testing.T) { + // Use InitShardNum=3 (odd): InitShardingRange rounds it up to 4 (next power of 2). + // Before the fix, initShardDone would trigger after 3 shards instead of 4. + conf := testConfig + conf.InitShardNum = 3 + + dir := path.Join(os.TempDir(), fmt.Sprintf("catalogmgr-initdone-%d-%010d", time.Now().Unix(), rand.Intn(100000000))) + catalogDBPath := path.Join(dir, "sharddb") + defer os.RemoveAll(dir) + + catalogDB, err := catalogdb.Open(catalogDBPath) + require.NoError(t, err) + + ctr := gomock.NewController(t) + mockDiskMgr := cluster.NewMockShardNodeManagerAPI(ctr) + mockScopeMgr := mock.NewMockScopeMgrAPI(ctr) + mockKvMgr := mock.NewMockKvMgrAPI(ctr) + + mockDiskMgr.EXPECT().GetDiskInfo(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(mockGetDiskInfo) + // ShardInitDoneKey not found: initShardDone must NOT be set on startup + mockKvMgr.EXPECT().Get(gomock.Any()).Return(nil, fmt.Errorf("key not found")) + // Set is called exactly once when actualInitShardNum shards have been applied + mockKvMgr.EXPECT().Set(gomock.Any(), gomock.Any()).Times(1).Return(nil) + + catalogMgr, err := NewCatalogMgr(conf, mockDiskMgr, mockScopeMgr, mockKvMgr, catalogDB) + require.NoError(t, err) + defer catalogMgr.Close() + + _, ctx := trace.StartSpanFromContext(context.Background(), "") + + // Initially, initShardDone must NOT be set + require.False(t, catalogMgr.IsShardInitDone(ctx)) + + // actualInitShardNum equals the real range count, which is greater than InitShardNum for odd values + actualNum := catalogMgr.actualInitShardNum + ranges := sharding.InitShardingRange(sharding.RangeType_RangeTypeHash, 2, conf.InitShardNum) + require.Equal(t, actualNum, len(ranges)) + require.Greater(t, actualNum, conf.InitShardNum, "InitShardingRange should expand odd shardCount") + + unitCount := conf.CodeMode.GetShardNum() + + // Apply actualNum-1 shards: initShardDone must NOT be set at any intermediate step + for i := 0; i < actualNum-1; i++ { + shard := newTestShardItem(proto.ShardID(i+1), ranges[i], unitCount) + require.NoError(t, catalogMgr.applyCreateShard(ctx, shard)) + require.False(t, catalogMgr.IsShardInitDone(ctx), + "initShardDone must not be set after only %d shards (need %d)", i+1, actualNum) + } + + // Apply the last shard: initShardDone must now be set + lastShard := newTestShardItem(proto.ShardID(actualNum), ranges[actualNum-1], unitCount) + require.NoError(t, catalogMgr.applyCreateShard(ctx, lastShard)) + require.True(t, catalogMgr.IsShardInitDone(ctx), + "initShardDone must be set after all %d shards are applied", actualNum) +} + +// newTestShardItem constructs a minimal valid shardItem for use in tests. +func newTestShardItem(shardID proto.ShardID, rng *sharding.Range, unitCount int) *shardItem { + unitEpochs := make([]*shardUnitEpoch, unitCount) + units := make([]clustermgr.ShardUnit, unitCount) + for j := 0; j < unitCount; j++ { + suid := proto.EncodeSuid(shardID, uint8(j), proto.MinEpoch) + unitEpochs[j] = &shardUnitEpoch{ + suidPrefix: suid.SuidPrefix(), + epoch: proto.MinEpoch, + nextEpoch: proto.MinEpoch, + } + units[j] = clustermgr.ShardUnit{ + Suid: suid, + DiskID: proto.DiskID(j + 1), + } + } + return &shardItem{ + shardID: shardID, + unitEpochs: unitEpochs, + info: shardInfoBase{ + Shard: clustermgr.Shard{ + ShardID: shardID, + Range: *rng, + Units: units, + }, + }, + } +} diff --git a/blobstore/clustermgr/catalog/startup.go b/blobstore/clustermgr/catalog/startup.go index a751cb7ad..fb2f70c56 100644 --- a/blobstore/clustermgr/catalog/startup.go +++ b/blobstore/clustermgr/catalog/startup.go @@ -28,6 +28,7 @@ import ( "github.com/cubefs/cubefs/blobstore/common/kvstore" "github.com/cubefs/cubefs/blobstore/common/proto" "github.com/cubefs/cubefs/blobstore/common/raftserver" + "github.com/cubefs/cubefs/blobstore/common/sharding" "github.com/cubefs/cubefs/blobstore/common/trace" "github.com/cubefs/cubefs/blobstore/util/defaulter" "github.com/cubefs/cubefs/blobstore/util/errors" @@ -75,12 +76,16 @@ func NewCatalogMgr(conf Config, diskMgr cluster.ShardNodeManagerAPI, scopeMgr sc return nil, errors.Info(err, "open transited table failed").Detail(err) } + // Compute actual range count; InitShardingRange may round up to the next power of 2. + actualInitShardNum := len(sharding.InitShardingRange(sharding.RangeType_RangeTypeHash, 2, conf.InitShardNum)) + // initial catalogMgr catalogMgr := &CatalogMgr{ - allShards: newConcurrentShards(conf.ShardConcurrentMapNum), - allSpaces: newConcurrentSpaces(conf.SpaceConcurrentMapNum), - catalogTbl: catalogTable, - transitedTbl: transitedTable, + allShards: newConcurrentShards(conf.ShardConcurrentMapNum), + allSpaces: newConcurrentSpaces(conf.SpaceConcurrentMapNum), + catalogTbl: catalogTable, + transitedTbl: transitedTable, + actualInitShardNum: actualInitShardNum, applyTaskPool: base.NewTaskDistribution(int(conf.ApplyConcurrency), 1), scopeMgr: scopeMgr,