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

Set triggerCondition when queryCoord reload loadbalanceTask meta (#15380)

Browse Source 
Signed-off-by: xige-16 <xi.ge@zilliz.com>
This commit is contained in:
xige-16 2022-01-25 17:26:13 +08:00 committed by GitHub
parent 8ab9d769de
commit 6336e2326c
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 17 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
internal/querycoord/impl.go
View File

@ -847,6 +847,7 @@ func (qc *QueryCoord) LoadBalance(ctx context.Context, req *querypb.LoadBalanceR
}
baseTask := newBaseTask(qc.loopCtx, querypb.TriggerCondition_LoadBalance)
req.BalanceReason = querypb.TriggerCondition_LoadBalance
loadBalanceTask := &loadBalanceTask{
baseTask: baseTask,
LoadBalanceRequest: req,

1
internal/querycoord/query_coord.go
View File

@ -343,6 +343,7 @@ func (qc *QueryCoord) watchNodeLoop() {
MsgType: commonpb.MsgType_LoadBalanceSegments,
SourceID: qc.session.ServerID,
},
BalanceReason: querypb.TriggerCondition_NodeDown,
SourceNodeIDs: offlineNodeIDs,
}

12
internal/querycoord/task.go
View File

@ -72,6 +72,7 @@ type task interface {
msgType() commonpb.MsgType
timestamp() Timestamp
getTriggerCondition() querypb.TriggerCondition
setTriggerCondition(trigger querypb.TriggerCondition)
preExecute(ctx context.Context) error
execute(ctx context.Context) error
postExecute(ctx context.Context) error
@ -109,6 +110,7 @@ type baseTask struct {
taskID UniqueID
triggerCondition querypb.TriggerCondition
triggerMu sync.RWMutex
parentTask task
childTasks []task
childTasksMu sync.RWMutex
@ -146,9 +148,19 @@ func (bt *baseTask) traceCtx() context.Context {
}
func (bt *baseTask) getTriggerCondition() querypb.TriggerCondition {
bt.triggerMu.RLock()
defer bt.triggerMu.RUnlock()
return bt.triggerCondition
}
func (bt *baseTask) setTriggerCondition(trigger querypb.TriggerCondition) {
bt.triggerMu.Lock()
defer bt.triggerMu.Unlock()
bt.triggerCondition = trigger
}
func (bt *baseTask) taskPriority() querypb.TriggerCondition {
return bt.triggerCondition
}

3
internal/querycoord/task_scheduler.go
View File

@ -403,6 +403,9 @@ func (scheduler *TaskScheduler) unmarshalTask(taskID UniqueID, t string) (task,
if err != nil {
return nil, err
}
// if triggerCondition == nodeDown, and the queryNode resources are insufficient,
// queryCoord will waits until queryNode can load the data, ensuring that the data is not lost
baseTask.setTriggerCondition(loadReq.BalanceReason)
loadBalanceTask := &loadBalanceTask{
baseTask: baseTask,
LoadBalanceRequest: &loadReq,