test/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2024-11-30 02:48:45 +08:00
Code Issues Actions 11 Packages Projects Releases Wiki Activity

Move task merge cap to config file (#19387)

Browse Source 
Signed-off-by: sunby <bingyi.sun@zilliz.com>

Signed-off-by: sunby <bingyi.sun@zilliz.com>
Co-authored-by: sunby <bingyi.sun@zilliz.com>
This commit is contained in:
Bingyi Sun 2022-09-24 17:56:51 +08:00 committed by GitHub
parent 9e18645363
commit 29d2c0bd7b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 14 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
configs/milvus.yaml
View File

@ -185,6 +185,7 @@ queryCoord:
distPullInterval: 500
loadTimeoutSeconds: 600
checkHandoffInterval: 5000
taskMergeCap: 2

10
internal/querycoordv2/task/merger.go
View File

@ -6,15 +6,13 @@ import (
"time"
"github.com/milvus-io/milvus/internal/log"
. "github.com/milvus-io/milvus/internal/querycoordv2/params"
"github.com/milvus-io/milvus/internal/util/typeutil"
"go.uber.org/zap"
)
// Merger merges tasks with the same mergeID.
const (
taskQueueCap = 16
waitQueueCap = 128
)
const waitQueueCap = 128
type Merger[K comparable, R any] struct {
stopCh chan struct{}
@ -32,7 +30,7 @@ func NewMerger[K comparable, R any]() *Merger[K, R] {
processors: typeutil.NewConcurrentSet[K](),
queues: make(map[K]chan MergeableTask[K, R]),
waitQueue: make(chan MergeableTask[K, R], waitQueueCap),
outCh: make(chan MergeableTask[K, R], taskQueueCap),
outCh: make(chan MergeableTask[K, R], Params.QueryCoordCfg.TaskMergeCap),
}
}
@ -68,7 +66,7 @@ func (merger *Merger[K, R]) schedule(ctx context.Context) {
case task := <-merger.waitQueue:
queue, ok := merger.queues[task.ID()]
if !ok {
queue = make(chan MergeableTask[K, R], taskQueueCap)
queue = make(chan MergeableTask[K, R], Params.QueryCoordCfg.TaskMergeCap)
merger.queues[task.ID()] = queue
}
outer:

3
internal/querycoordv2/task/merger_test.go
View File

@ -7,6 +7,7 @@ import (
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/proto/querypb"
. "github.com/milvus-io/milvus/internal/querycoordv2/params"
"github.com/stretchr/testify/suite"
)
@ -22,6 +23,8 @@ type MergerSuite struct {
}
func (suite *MergerSuite) SetupSuite() {
Params.Init()
Params.QueryCoordCfg.TaskMergeCap = 3
suite.collectionID = 1000
suite.replicaID = 100
suite.nodeID = 1

6
internal/util/paramtable/component_param.go
View File

@ -617,6 +617,7 @@ type queryCoordConfig struct {
//---- Task ---
RetryNum int32
RetryInterval int64
TaskMergeCap int32
//---- Handoff ---
AutoHandoff bool
@ -641,6 +642,7 @@ func (p *queryCoordConfig) init(base *BaseTable) {
//---- Task ---
p.initTaskRetryNum()
p.initTaskRetryInterval()
p.initTaskMergeCap()
//---- Handoff ---
p.initAutoHandoff()
@ -666,6 +668,10 @@ func (p *queryCoordConfig) initTaskRetryInterval() {
p.RetryInterval = p.Base.ParseInt64WithDefault("queryCoord.task.retryinterval", int64(10*time.Second))
}
func (p *queryCoordConfig) initTaskMergeCap() {
p.TaskMergeCap = p.Base.ParseInt32WithDefault("queryCoord.taskMergeCap", 2)
}
func (p *queryCoordConfig) initAutoHandoff() {
handoff, err := p.Base.Load("queryCoord.autoHandoff")
if err != nil {