gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(blobnode): add config to enable iopool feature

Browse Source 
with: #1000194242

Signed-off-by: mawei029 <mawei2@oppo.com>
This commit is contained in:
mawei029 2025-06-24 16:25:01 +08:00 committed by slasher
parent 04717ce338
commit c2eae16c72
4 changed files with 90 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
blobstore/blobnode/core/blobfile_test.go
View File

@ -30,32 +30,21 @@ import (
"github.com/cubefs/cubefs/blobstore/util/taskpool"
)
func TestBlobFile_Op(t *testing.T) {
testDir, err := os.MkdirTemp(os.TempDir(), "BlobFileOp")
require.NoError(t, err)
defer os.RemoveAll(testDir)
posixfilepath := filepath.Join(testDir, "PoxsixFile")
log.Info(posixfilepath)
temppath := filepath.Join(posixfilepath, "xxxtemp")
f, err := OpenFile(temppath, false)
require.Error(t, err)
require.Nil(t, f)
f, err = OpenFile(posixfilepath, true)
require.NoError(t, err)
require.NotNil(t, f)
func doBlobFileOp(t *testing.T, f *os.File, emptyIoPool bool) {
// create
syncWorker := mergetask.NewMergeTask(-1, func(interface{}) error { return nil })
ctr := gomock.NewController(t)
ioPool := mocks.NewMockIoPool(ctr)
ioPool.EXPECT().Submit(gomock.Any()).Do(func(args taskpool.IoPoolTaskArgs) {
args.TaskFn()
}).AnyTimes()
ioPool := taskpool.NewReadPool(0, 0, taskpool.IoPoolMetricConf{})
if !emptyIoPool {
ctr := gomock.NewController(t)
ioPool = mocks.NewMockIoPool(ctr)
ioPool.(*mocks.MockIoPool).EXPECT().Submit(gomock.Any()).Do(func(args taskpool.IoPoolTaskArgs) {
args.TaskFn()
}).AnyTimes()
// ioPool.EXPECT().IsEmpty().Return(emptyIoPool).AnyTimes()
}
ioPools := map[qos.IOTypeRW]taskpool.IoPool{
qos.IOTypeRead: ioPool,
qos.IOTypeWrite: ioPool,
@ -141,6 +130,35 @@ func TestBlobFile_Op(t *testing.T) {
require.Equal(t, int(stat.Blocks), 0)
}
func TestBlobFile_Op(t *testing.T) {
testDir, err := os.MkdirTemp(os.TempDir(), "BlobFileOp")
require.NoError(t, err)
defer os.RemoveAll(testDir)
posixfilepath := filepath.Join(testDir, "PoxsixFile")
log.Info(posixfilepath)
temppath := filepath.Join(posixfilepath, "xxxtemp")
f, err := OpenFile(temppath, false)
require.Error(t, err)
require.Nil(t, f)
f, err = OpenFile(posixfilepath, true)
require.NoError(t, err)
require.NotNil(t, f)
// enable io pools
doBlobFileOp(t, f, false)
// invalid/empty io pools
posixfilepath2 := filepath.Join(testDir, "PoxsixFile2")
f, err = OpenFile(posixfilepath2, true)
require.NoError(t, err)
require.NotNil(t, f)
doBlobFileOp(t, f, true)
}
func TestBlobFile_doTaskFnCtxCancel(t *testing.T) {
testDir, err := os.MkdirTemp(os.TempDir(), "BlobFileTaskCancel")
require.NoError(t, err)
@ -164,6 +182,7 @@ func TestBlobFile_doTaskFnCtxCancel(t *testing.T) {
ctr := gomock.NewController(t)
ioPool := mocks.NewMockIoPool(ctr)
// ioPool.EXPECT().IsEmpty().Return(false).AnyTimes()
ioPools := map[qos.IOTypeRW]taskpool.IoPool{
qos.IOTypeRead: ioPool,
qos.IOTypeWrite: ioPool,

6
blobstore/blobnode/core/config.go
View File

@ -153,9 +153,9 @@ func InitConfig(conf *Config) error {
conf.AllowCleanTrash = true
}
defaulter.LessOrEqual(&conf.WriteThreadCnt, defaultWriteThreadCnt)
defaulter.LessOrEqual(&conf.ReadThreadCnt, defaultReadThreadCnt)
defaulter.LessOrEqual(&conf.DeleteThreadCnt, defaultDeleteThreadCnt)
defaulter.Equal(&conf.WriteThreadCnt, defaultWriteThreadCnt)
defaulter.Equal(&conf.ReadThreadCnt, defaultReadThreadCnt)
defaulter.Equal(&conf.DeleteThreadCnt, defaultDeleteThreadCnt)
conf.DataQos.WriteChanQueCnt = int32(conf.WriteThreadCnt)
qos.InitAndFixQosConfig(&conf.DataQos)

15
blobstore/util/taskpool/iopool.go
View File

@ -95,6 +95,11 @@ func NewDeletePool(threadCnt, queueDepth int, conf IoPoolMetricConf) IoPool {
// $threadCnt: The number of read/write work goroutine, it must be greater than $chanCnt
// $queueDepth: The number of elements in the queue
func newCommonIoPool(chanCnt, threadCnt, queueDepth int, conf IoPoolMetricConf) *ioPoolSimple {
if threadCnt <= 0 || chanCnt <= 0 || queueDepth <= 0 {
// empty io pool, dont limit
return &ioPoolSimple{closed: make(chan struct{})}
}
iopoolMetric := prometheus.NewSummaryVec(
prometheus.SummaryOpts{
Namespace: conf.Namespace,
@ -175,6 +180,11 @@ func (p *ioPoolSimple) reportMetric(opStage string, tm time.Time) {
}
func (p *ioPoolSimple) Submit(args IoPoolTaskArgs) {
if p.notLimit() {
args.TaskFn()
return
}
idx, task := p.generateTask(args)
// if ctx has been cancelled, try to avoid enqueuing as much as possible;
@ -238,3 +248,8 @@ func (p *ioPoolSimple) Close() {
log.Info("close all io pool, exit")
}
func (p *ioPoolSimple) notLimit() bool {
// dont limit: queue is empty
return p.queue == nil
}

30
blobstore/util/taskpool/iopool_test.go
View File

@ -13,6 +13,7 @@ import (
)
func TestIoPoolSimple(t *testing.T) {
// normal
metricConf := IoPoolMetricConf{
ClusterID: 1,
IDC: "idc",
@ -241,4 +242,33 @@ func TestIoPoolSimple(t *testing.T) {
require.NoError(t, err)
require.Equal(t, 1, n) // long time task, not raw data
}
// empty, not limit write
emptyPool := newCommonIoPool(0, 0, 0, IoPoolMetricConf{})
require.True(t, emptyPool.notLimit())
defer emptyPool.Close()
n := 0
data := []byte(content)
task := IoPoolTaskArgs{
TaskFn: func() { n, err = file.WriteAt(data, 0) },
}
emptyPool.Submit(task)
require.NoError(t, err)
require.Equal(t, len(content), n)
// empty, not limit read
n = 0
data = make([]byte, len(content))
task = IoPoolTaskArgs{
TaskFn: func() { n, err = file.ReadAt(data, 0) },
}
require.Equal(t, uint8(0), data[0])
require.Equal(t, 0, n)
emptyPool.Submit(task)
require.NoError(t, err)
require.Equal(t, len(content), n)
require.Equal(t, content, string(data))
}