fix(data): rewrite disk selector

Signed-off-by: NaturalSelect <huangzhibin1@oppo.com>

datanode/space_manager.go

@@ -17,6 +17,7 @@ package datanode
 import (
 	"fmt"
 	"math"
+	"math/rand"
 	"os"
 	"path"
 	"sync"
@@ -25,6 +26,7 @@ import (
 
 	"github.com/cubefs/cubefs/proto"
 	"github.com/cubefs/cubefs/raftstore"
+	"github.com/cubefs/cubefs/util"
 	"github.com/cubefs/cubefs/util/atomicutil"
 	"github.com/cubefs/cubefs/util/loadutil"
 	"github.com/cubefs/cubefs/util/log"
@@ -33,20 +35,23 @@ import (
 
 // SpaceManager manages the disk space.
 type SpaceManager struct {
-	clusterID      string
-	disks          map[string]*Disk
-	partitions     map[uint64]*DataPartition
-	raftStore      raftstore.RaftStore
-	nodeID         uint64
-	diskMutex      sync.RWMutex
-	partitionMutex sync.RWMutex
-	stats          *Stats
-	stopC          chan bool
-	diskList       []string
-	dataNode       *DataNode
-	diskUtils      map[string]*atomicutil.Float64
-	samplerDone    chan struct{}
-	allDisksLoaded bool
+	clusterID            string
+	disks                map[string]*Disk
+	partitions           map[uint64]*DataPartition
+	raftStore            raftstore.RaftStore
+	nodeID               uint64
+	diskMutex            sync.RWMutex
+	partitionMutex       sync.RWMutex
+	stats                *Stats
+	stopC                chan bool
+	selectedIndex        int // TODO what is selected index
+	diskList             []string
+	dataNode             *DataNode
+	createPartitionMutex sync.RWMutex
+	rand                 *rand.Rand
+	diskUtils            map[string]*atomicutil.Float64
+	samplerDone          chan struct{}
+	allDisksLoaded       bool
 }
 
 const diskSampleDuration = 1 * time.Second
@@ -60,6 +65,7 @@ func NewSpaceManager(dataNode *DataNode) *SpaceManager {
 	space.stats = NewStats(dataNode.zoneName)
 	space.stopC = make(chan bool)
 	space.dataNode = dataNode
+	space.rand = rand.New(rand.NewSource(time.Now().Unix()))
 	space.diskUtils = make(map[string]*atomicutil.Float64)
 	go space.statUpdateScheduler()
 
@@ -325,39 +331,37 @@ func (manager *SpaceManager) updateMetrics() {
 		remainingCapacityToCreatePartition, maxCapacityToCreatePartition, partitionCnt)
 }
 
-func (manager *SpaceManager) minPartitionCnt(decommissionedDisks []string) (d *Disk) {
+const DiskSelectMaxStraw = 65536
+
+func (manager *SpaceManager) selectDisk(decommissionedDisks []string) (d *Disk) {
 	manager.diskMutex.Lock()
 	defer manager.diskMutex.Unlock()
-	var (
-		minWeight     float64
-		minWeightDisk *Disk
-	)
 	decommissionedDiskMap := make(map[string]struct{})
 	for _, disk := range decommissionedDisks {
 		decommissionedDiskMap[disk] = struct{}{}
 	}
-	minWeight = math.MaxFloat64
+	maxStraw := float64(0)
 	for _, disk := range manager.disks {
 		if _, ok := decommissionedDiskMap[disk.Path]; ok {
 			log.LogInfof("action[minPartitionCnt] exclude decommissioned disk[%v]", disk.Path)
 			continue
 		}
 		if disk.Status != proto.ReadWrite {
+			log.LogInfof("[minPartitionCnt] disk(%v) is not writable", disk.Path)
 			continue
 		}
-		diskWeight := disk.getSelectWeight()
-		if diskWeight < minWeight {
-			minWeight = diskWeight
-			minWeightDisk = disk
+
+		straw := float64(manager.rand.Intn(DiskSelectMaxStraw))
+		straw = math.Log(straw/float64(DiskSelectMaxStraw)) / (float64(atomic.LoadUint64(&disk.Available)) / util.GB)
+		if d == nil || straw > maxStraw {
+			maxStraw = straw
+			d = disk
 		}
 	}
-	if minWeightDisk == nil {
+	if d != nil && d.Status != proto.ReadWrite {
+		d = nil
 		return
 	}
-	if minWeightDisk.Status != proto.ReadWrite {
-		return
-	}
-	d = minWeightDisk
 	return d
 }
 
@@ -423,8 +427,9 @@ func (manager *SpaceManager) CreatePartition(request *proto.CreateDataPartitionR
 		}
 		return
 	}
-	disk := manager.minPartitionCnt(request.DecommissionedDisks)
+	disk := manager.selectDisk(request.DecommissionedDisks)
 	if disk == nil {
+		log.LogErrorf("[CreatePartition] dp(%v) failed to select disk", dpCfg.PartitionID)
 		return nil, ErrNoSpaceToCreatePartition
 	}
 	if dp, err = CreateDataPartition(dpCfg, disk, request); err != nil {

datanode/space_manager_test.go

@@ -0,0 +1,113 @@
+// Copyright 2024 The CubeFS Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+// implied. See the License for the specific language governing
+// permissions and limitations under the License.
+
+package datanode
+
+import (
+	"fmt"
+	"math/rand"
+	"sort"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/cubefs/cubefs/proto"
+	"github.com/cubefs/cubefs/util"
+	"github.com/stretchr/testify/require"
+)
+
+func prepareDisksForSelectDiskTest(t *testing.T, sm *SpaceManager, cnt int) {
+	for i := 0; i < cnt; i++ {
+		size := uint64(i+1) * 1000 * util.GB
+		diskPath := fmt.Sprintf("/cfs/disk_%v", strconv.FormatInt(int64(i), 10))
+		disk := &Disk{
+			Total:     size,
+			Available: size,
+			Used:      0,
+			Allocated: 0,
+			Status:    proto.ReadWrite,
+			Path:      diskPath,
+		}
+		sm.disks[diskPath] = disk
+		t.Logf("disk(%v) space(%v) GB", diskPath, size/util.GB)
+	}
+}
+
+func TestSelectDisk(t *testing.T) {
+	sm := &SpaceManager{
+		disks: make(map[string]*Disk),
+		rand:  rand.New(rand.NewSource(time.Now().Unix())),
+	}
+	// prepare disks
+	prepareDisksForSelectDiskTest(t, sm, 4)
+	const testCount = 1500
+	selectTimes := make(map[string]int)
+	for _, disk := range sm.disks {
+		selectTimes[disk.Path] = 0
+	}
+	decommsionDisk := []string{}
+	for i := 0; i < testCount; i++ {
+		disk := sm.selectDisk(decommsionDisk)
+		require.NotNil(t, disk)
+		selectTimes[disk.Path] += 1
+		used := rand.Float64() * util.GB * 10
+		disk.Allocated += uint64(used)
+		disk.Available -= uint64(used)
+		disk.Used += uint64(used)
+	}
+	for disk, times := range selectTimes {
+		t.Logf("disk(%v) select times(%v)", disk, times)
+	}
+	for _, disk := range sm.disks {
+		t.Logf("disk(%v) left space(%v) GB", disk.Path, disk.Available/util.GB)
+	}
+
+	// NOTE: check for space
+	ratio := make([]float64, 0)
+	for _, disk := range sm.disks {
+		r := float64(disk.Available) / float64(disk.Total)
+		ratio = append(ratio, r)
+		t.Logf("disk(%v) ratio(%v)", disk.Path, r)
+	}
+	sort.Slice(ratio, func(i, j int) bool {
+		return ratio[i] < ratio[j]
+	})
+	require.Less(t, ratio[len(ratio)-1]-ratio[0], 0.1)
+}
+
+func TestSelectDiskForSmallDp(t *testing.T) {
+	sm := &SpaceManager{
+		disks: make(map[string]*Disk),
+		rand:  rand.New(rand.NewSource(time.Now().Unix())),
+	}
+	// prepare disks
+	prepareDisksForSelectDiskTest(t, sm, 4)
+	const testCount = 1500
+	selectTimes := make(map[string]int)
+	for _, disk := range sm.disks {
+		selectTimes[disk.Path] = 0
+	}
+	decommsionDisk := []string{}
+	for i := 0; i < testCount; i++ {
+		disk := sm.selectDisk(decommsionDisk)
+		require.NotNil(t, disk)
+		selectTimes[disk.Path] += 1
+	}
+	for disk, times := range selectTimes {
+		t.Logf("disk(%v) select times(%v)", disk, times)
+	}
+	for _, disk := range sm.disks {
+		t.Logf("disk(%v) left space(%v) GB", disk.Path, disk.Available/util.GB)
+	}
+}

datanode/wrap_prepare.go

@@ -93,6 +93,7 @@ func (s *DataNode) checkPartition(p *repl.Packet) (err error) {
 	p.Object = dp
 	if p.IsNormalWriteOperation() || p.IsCreateExtentOperation() {
 		if dp.Available() <= 0 {
+			log.LogErrorf("[checkPartition] dp(%v) disk no space available(%v) can write(%v)", dp.partitionID, dp.Available(), dp.disk.CanWrite())
 			err = storage.NoSpaceError
 			return
 		}