Skip to content

M
milvus

milvus / milvus
11 个月 前同步成功

通知 261
Star 22476
Fork 2472
M
milvus

体验新版 GitCode，发现更多精彩内容 >>

未验证 提交 e80fbd1f 编写于 作者: X xige-16 提交者： GitHub
浏览文件
操作

Fix golint error in the task of querycoord (#9866) 

Signed-off-by: Nxige-16 <xi.ge@zilliz.com>
上级 d0f03b2e
展开全部 隐藏空白更改
内联 并排
Showing with 392 addition and 392 deletion
internal/querycoord/condition.go
浏览文件 @ e80fbd1f
......@@ -17,8 +17,8 @@ import (
)
type Condition interface {
WaitToFinish() error
Notify(err error)
waitToFinish() error
notify(err error)
Ctx() context.Context
}
......@@ -27,7 +27,7 @@ type TaskCondition struct {
ctx context.Context
}
func (tc *TaskCondition) WaitToFinish() error {
func (tc *TaskCondition) waitToFinish() error {
for {
select {
case <-tc.ctx.Done():
......@@ -38,7 +38,7 @@ func (tc *TaskCondition) WaitToFinish() error {
}
}
func (tc *TaskCondition) Notify(err error) {
func (tc *TaskCondition) notify(err error) {
tc.done <- err
}
......
internal/querycoord/impl.go
浏览文件 @ e80fbd1f
......@@ -157,7 +157,7 @@ func (qc *QueryCoord) LoadCollection(ctx context.Context, req *querypb.LoadColle
return status, err
}
err = loadCollectionTask.WaitToFinish()
err = loadCollectionTask.waitToFinish()
if err != nil {
status.ErrorCode = commonpb.ErrorCode_UnexpectedError
status.Reason = err.Error()
......@@ -205,7 +205,7 @@ func (qc *QueryCoord) ReleaseCollection(ctx context.Context, req *querypb.Releas
return status, err
}
err = releaseCollectionTask.WaitToFinish()
err = releaseCollectionTask.waitToFinish()
if err != nil {
status.ErrorCode = commonpb.ErrorCode_UnexpectedError
status.Reason = err.Error()
......@@ -348,7 +348,7 @@ func (qc *QueryCoord) LoadPartitions(ctx context.Context, req *querypb.LoadParti
return status, err
}
err = loadPartitionTask.WaitToFinish()
err = loadPartitionTask.waitToFinish()
if err != nil {
status.ErrorCode = commonpb.ErrorCode_UnexpectedError
status.Reason = err.Error()
......@@ -417,7 +417,7 @@ func (qc *QueryCoord) ReleasePartitions(ctx context.Context, req *querypb.Releas
return status, err
}
err = releasePartitionTask.WaitToFinish()
err = releasePartitionTask.waitToFinish()
if err != nil {
status.ErrorCode = commonpb.ErrorCode_UnexpectedError
status.Reason = err.Error()
......
internal/querycoord/task.go
浏览文件 @ e80fbd1f
此差异已折叠。
internal/querycoord/task_scheduler.go
浏览文件 @ e80fbd1f
......@@ -71,7 +71,7 @@ func (queue *TaskQueue) addTask(t task) {
}
for e := queue.tasks.Back(); e != nil; e = e.Prev() {
if t.TaskPriority() > e.Value.(task).TaskPriority() {
if t.taskPriority() > e.Value.(task).taskPriority() {
if e.Prev() == nil {
queue.taskChan <- 1
queue.tasks.InsertBefore(t, e)
......@@ -233,18 +233,18 @@ func (scheduler *TaskScheduler) reloadFromKV() error {
log.Error("reloadFromKV: taskStateInfo and triggerTaskInfo are inconsistent")
continue
}
triggerTasks[taskID].SetState(state)
triggerTasks[taskID].setState(state)
}
var doneTriggerTask task = nil
for _, t := range triggerTasks {
if t.State() == taskDone {
if t.getState() == taskDone {
doneTriggerTask = t
for _, childTask := range activeTasks {
childTask.SetParentTask(t) //replace child task after reScheduler
t.AddChildTask(childTask)
childTask.setParentTask(t) //replace child task after reScheduler
t.addChildTask(childTask)
}
t.SetResultInfo(nil)
t.setResultInfo(nil)
continue
}
scheduler.triggerTaskQueue.addTask(t)
......@@ -399,7 +399,7 @@ func (scheduler *TaskScheduler) unmarshalTask(taskID UniqueID, t string) (task,
return nil, err
}
newTask.SetID(taskID)
newTask.setTaskID(taskID)
return newTask, nil
}
......@@ -410,26 +410,26 @@ func (scheduler *TaskScheduler) Enqueue(t task) error {
log.Error("allocator trigger taskID failed", zap.Error(err))
return err
}
t.SetID(id)
t.setTaskID(id)
kvs := make(map[string]string)
taskKey := fmt.Sprintf("%s/%d", triggerTaskPrefix, t.ID())
blobs, err := t.Marshal()
taskKey := fmt.Sprintf("%s/%d", triggerTaskPrefix, t.getTaskID())
blobs, err := t.marshal()
if err != nil {
log.Error("error when save marshal task", zap.Int64("taskID", t.ID()), zap.Error(err))
log.Error("error when save marshal task", zap.Int64("taskID", t.getTaskID()), zap.Error(err))
return err
}
kvs[taskKey] = string(blobs)
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, t.ID())
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, t.getTaskID())
kvs[stateKey] = strconv.Itoa(int(taskUndo))
err = scheduler.client.MultiSave(kvs)
if err != nil {
//TODO::clean etcd meta
log.Error("error when save trigger task to etcd", zap.Int64("taskID", t.ID()), zap.Error(err))
log.Error("error when save trigger task to etcd", zap.Int64("taskID", t.getTaskID()), zap.Error(err))
return err
}
t.SetState(taskUndo)
t.setState(taskUndo)
scheduler.triggerTaskQueue.addTask(t)
log.Debug("EnQueue a triggerTask and save to etcd", zap.Int64("taskID", t.ID()))
log.Debug("EnQueue a triggerTask and save to etcd", zap.Int64("taskID", t.getTaskID()))
return nil
}
......@@ -440,19 +440,19 @@ func (scheduler *TaskScheduler) processTask(t task) error {
updateKVFn := func(parentTask task) error {
kvs := make(map[string]string)
kvs[taskInfoKey] = strconv.Itoa(int(taskDone))
for _, childTask := range parentTask.GetChildTask() {
for _, childTask := range parentTask.getChildTask() {
id, err := scheduler.taskIDAllocator()
if err != nil {
return err
}
childTask.SetID(id)
childTaskKey := fmt.Sprintf("%s/%d", activeTaskPrefix, childTask.ID())
blobs, err := childTask.Marshal()
childTask.setTaskID(id)
childTaskKey := fmt.Sprintf("%s/%d", activeTaskPrefix, childTask.getTaskID())
blobs, err := childTask.marshal()
if err != nil {
return err
}
kvs[childTaskKey] = string(blobs)
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, childTask.ID())
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, childTask.getTaskID())
kvs[stateKey] = strconv.Itoa(int(taskUndo))
}
err := scheduler.client.MultiSave(kvs)
......@@ -462,35 +462,35 @@ func (scheduler *TaskScheduler) processTask(t task) error {
return nil
}
span, ctx := trace.StartSpanFromContext(t.TraceCtx(),
span, ctx := trace.StartSpanFromContext(t.traceCtx(),
opentracing.Tags{
"Type": t.Type(),
"ID": t.ID(),
"Type": t.msgType(),
"ID": t.getTaskID(),
})
var err error
defer span.Finish()
defer func() {
//task postExecute
span.LogFields(oplog.Int64("processTask: scheduler process PostExecute", t.ID()))
t.PostExecute(ctx)
span.LogFields(oplog.Int64("processTask: scheduler process PostExecute", t.getTaskID()))
t.postExecute(ctx)
}()
// task preExecute
span.LogFields(oplog.Int64("processTask: scheduler process PreExecute", t.ID()))
t.PreExecute(ctx)
taskInfoKey = fmt.Sprintf("%s/%d", taskInfoPrefix, t.ID())
span.LogFields(oplog.Int64("processTask: scheduler process PreExecute", t.getTaskID()))
t.preExecute(ctx)
taskInfoKey = fmt.Sprintf("%s/%d", taskInfoPrefix, t.getTaskID())
err = scheduler.client.Save(taskInfoKey, strconv.Itoa(int(taskDoing)))
if err != nil {
trace.LogError(span, err)
t.SetResultInfo(err)
t.setResultInfo(err)
return err
}
t.SetState(taskDoing)
t.setState(taskDoing)
// task execute
span.LogFields(oplog.Int64("processTask: scheduler process Execute", t.ID()))
err = t.Execute(ctx)
span.LogFields(oplog.Int64("processTask: scheduler process Execute", t.getTaskID()))
err = t.execute(ctx)
if err != nil {
trace.LogError(span, err)
return err
......@@ -498,16 +498,16 @@ func (scheduler *TaskScheduler) processTask(t task) error {
err = updateKVFn(t)
if err != nil {
trace.LogError(span, err)
t.SetResultInfo(err)
t.setResultInfo(err)
return err
}
log.Debug("processTask: update etcd success", zap.Int64("parent taskID", t.ID()))
if t.Type() == commonpb.MsgType_LoadCollection || t.Type() == commonpb.MsgType_LoadPartitions {
t.Notify(nil)
log.Debug("processTask: update etcd success", zap.Int64("parent taskID", t.getTaskID()))
if t.msgType() == commonpb.MsgType_LoadCollection || t.msgType() == commonpb.MsgType_LoadPartitions {
t.notify(nil)
}
t.SetState(taskDone)
t.UpdateTaskProcess()
t.setState(taskDone)
t.updateTaskProcess()
return nil
}
......@@ -521,7 +521,7 @@ func (scheduler *TaskScheduler) scheduleLoop() {
log.Debug("scheduleLoop: num of child task", zap.Int("num child task", len(activateTasks)))
for _, childTask := range activateTasks {
if childTask != nil {
log.Debug("scheduleLoop: add a activate task to activateChan", zap.Int64("taskID", childTask.ID()))
log.Debug("scheduleLoop: add a activate task to activateChan", zap.Int64("taskID", childTask.getTaskID()))
scheduler.activateTaskChan <- childTask
activeTaskWg.Add(1)
go scheduler.waitActivateTaskDone(activeTaskWg, childTask, triggerTask)
......@@ -535,10 +535,10 @@ func (scheduler *TaskScheduler) scheduleLoop() {
removes := make([]string, 0)
childTaskIDs := make([]int64, 0)
for _, t := range originInternalTasks {
childTaskIDs = append(childTaskIDs, t.ID())
taskKey := fmt.Sprintf("%s/%d", activeTaskPrefix, t.ID())
childTaskIDs = append(childTaskIDs, t.getTaskID())
taskKey := fmt.Sprintf("%s/%d", activeTaskPrefix, t.getTaskID())
removes = append(removes, taskKey)
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, t.ID())
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, t.getTaskID())
removes = append(removes, stateKey)
}
......@@ -547,14 +547,14 @@ func (scheduler *TaskScheduler) scheduleLoop() {
if err != nil {
return err
}
t.SetID(id)
taskKey := fmt.Sprintf("%s/%d", activeTaskPrefix, t.ID())
blobs, err := t.Marshal()
t.setTaskID(id)
taskKey := fmt.Sprintf("%s/%d", activeTaskPrefix, t.getTaskID())
blobs, err := t.marshal()
if err != nil {
return err
}
saves[taskKey] = string(blobs)
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, t.ID())
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, t.getTaskID())
saves[stateKey] = strconv.Itoa(int(taskUndo))
}
......@@ -563,10 +563,10 @@ func (scheduler *TaskScheduler) scheduleLoop() {
return err
}
for _, taskID := range childTaskIDs {
triggerTask.RemoveChildTaskByID(taskID)
triggerTask.removeChildTaskByID(taskID)
}
for _, t := range rollBackTasks {
triggerTask.AddChildTask(t)
triggerTask.addChildTask(t)
}
return nil
......@@ -574,14 +574,14 @@ func (scheduler *TaskScheduler) scheduleLoop() {
removeTaskFromKVFn := func(triggerTask task) error {
keys := make([]string, 0)
taskKey := fmt.Sprintf("%s/%d", triggerTaskPrefix, triggerTask.ID())
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, triggerTask.ID())
taskKey := fmt.Sprintf("%s/%d", triggerTaskPrefix, triggerTask.getTaskID())
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, triggerTask.getTaskID())
keys = append(keys, taskKey)
keys = append(keys, stateKey)
childTasks := triggerTask.GetChildTask()
childTasks := triggerTask.getChildTask()
for _, t := range childTasks {
taskKey = fmt.Sprintf("%s/%d", activeTaskPrefix, t.ID())
stateKey = fmt.Sprintf("%s/%d", taskInfoPrefix, t.ID())
taskKey = fmt.Sprintf("%s/%d", activeTaskPrefix, t.getTaskID())
stateKey = fmt.Sprintf("%s/%d", taskInfoPrefix, t.getTaskID())
keys = append(keys, taskKey)
keys = append(keys, stateKey)
}
......@@ -600,36 +600,36 @@ func (scheduler *TaskScheduler) scheduleLoop() {
return
case <-scheduler.triggerTaskQueue.Chan():
triggerTask = scheduler.triggerTaskQueue.PopTask()
log.Debug("scheduleLoop: pop a triggerTask from triggerTaskQueue", zap.Int64("triggerTaskID", triggerTask.ID()))
log.Debug("scheduleLoop: pop a triggerTask from triggerTaskQueue", zap.Int64("triggerTaskID", triggerTask.getTaskID()))
alreadyNotify := true
if triggerTask.State() == taskUndo || triggerTask.State() == taskDoing {
if triggerTask.getState() == taskUndo || triggerTask.getState() == taskDoing {
err = scheduler.processTask(triggerTask)
if err != nil {
log.Debug("scheduleLoop: process triggerTask failed", zap.Int64("triggerTaskID", triggerTask.ID()), zap.Error(err))
log.Debug("scheduleLoop: process triggerTask failed", zap.Int64("triggerTaskID", triggerTask.getTaskID()), zap.Error(err))
alreadyNotify = false
}
}
if triggerTask.Type() != commonpb.MsgType_LoadCollection && triggerTask.Type() != commonpb.MsgType_LoadPartitions {
if triggerTask.msgType() != commonpb.MsgType_LoadCollection && triggerTask.msgType() != commonpb.MsgType_LoadPartitions {
alreadyNotify = false
}
childTasks := triggerTask.GetChildTask()
childTasks := triggerTask.getChildTask()
if len(childTasks) != 0 {
activateTasks := make([]task, len(childTasks))
copy(activateTasks, childTasks)
processInternalTaskFn(activateTasks, triggerTask)
resultStatus := triggerTask.GetResultInfo()
resultStatus := triggerTask.getResultInfo()
if resultStatus.ErrorCode != commonpb.ErrorCode_Success {
rollBackTasks := triggerTask.RollBack(scheduler.ctx)
rollBackTasks := triggerTask.rollBack(scheduler.ctx)
log.Debug("scheduleLoop: start rollBack after triggerTask failed",
zap.Int64("triggerTaskID", triggerTask.ID()),
zap.Int64("triggerTaskID", triggerTask.getTaskID()),
zap.Any("rollBackTasks", rollBackTasks))
err = rollBackInterTaskFn(triggerTask, childTasks, rollBackTasks)
if err != nil {
log.Error("scheduleLoop: rollBackInternalTask error",
zap.Int64("triggerTaskID", triggerTask.ID()),
zap.Int64("triggerTaskID", triggerTask.getTaskID()),
zap.Error(err))
triggerTask.SetResultInfo(err)
triggerTask.setResultInfo(err)
} else {
processInternalTaskFn(rollBackTasks, triggerTask)
}
......@@ -638,23 +638,23 @@ func (scheduler *TaskScheduler) scheduleLoop() {
err = removeTaskFromKVFn(triggerTask)
if err != nil {
log.Error("scheduleLoop: error when remove trigger and internal tasks from etcd", zap.Int64("triggerTaskID", triggerTask.ID()), zap.Error(err))
triggerTask.SetResultInfo(err)
log.Error("scheduleLoop: error when remove trigger and internal tasks from etcd", zap.Int64("triggerTaskID", triggerTask.getTaskID()), zap.Error(err))
triggerTask.setResultInfo(err)
} else {
log.Debug("scheduleLoop: trigger task done and delete from etcd", zap.Int64("triggerTaskID", triggerTask.ID()))
log.Debug("scheduleLoop: trigger task done and delete from etcd", zap.Int64("triggerTaskID", triggerTask.getTaskID()))
}
resultStatus := triggerTask.GetResultInfo()
resultStatus := triggerTask.getResultInfo()
if resultStatus.ErrorCode != commonpb.ErrorCode_Success {
triggerTask.SetState(taskFailed)
triggerTask.setState(taskFailed)
if !alreadyNotify {
triggerTask.Notify(errors.New(resultStatus.Reason))
triggerTask.notify(errors.New(resultStatus.Reason))
}
} else {
triggerTask.UpdateTaskProcess()
triggerTask.SetState(taskExpired)
triggerTask.updateTaskProcess()
triggerTask.setState(taskExpired)
if !alreadyNotify {
triggerTask.Notify(nil)
triggerTask.notify(nil)
}
}
}
......@@ -665,23 +665,23 @@ func (scheduler *TaskScheduler) waitActivateTaskDone(wg *sync.WaitGroup, t task,
defer wg.Done()
var err error
redoFunc1 := func() {
if !t.IsValid() || !t.IsRetryable() {
if !t.isValid() || !t.isRetryable() {
log.Debug("waitActivateTaskDone: reSchedule the activate task",
zap.Int64("taskID", t.ID()),
zap.Int64("triggerTaskID", triggerTask.ID()))
reScheduledTasks, err := t.Reschedule(scheduler.ctx)
zap.Int64("taskID", t.getTaskID()),
zap.Int64("triggerTaskID", triggerTask.getTaskID()))
reScheduledTasks, err := t.reschedule(scheduler.ctx)
if err != nil {
log.Error("waitActivateTaskDone: reschedule task error",
zap.Int64("taskID", t.ID()),
zap.Int64("triggerTaskID", triggerTask.ID()),
zap.Int64("taskID", t.getTaskID()),
zap.Int64("triggerTaskID", triggerTask.getTaskID()),
zap.Error(err))
triggerTask.SetResultInfo(err)
triggerTask.setResultInfo(err)
return
}
removes := make([]string, 0)
taskKey := fmt.Sprintf("%s/%d", activeTaskPrefix, t.ID())
taskKey := fmt.Sprintf("%s/%d", activeTaskPrefix, t.getTaskID())
removes = append(removes, taskKey)
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, t.ID())
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, t.getTaskID())
removes = append(removes, stateKey)
saves := make(map[string]string)
......@@ -690,44 +690,44 @@ func (scheduler *TaskScheduler) waitActivateTaskDone(wg *sync.WaitGroup, t task,
id, err := scheduler.taskIDAllocator()
if err != nil {
log.Error("waitActivateTaskDone: allocate id error",
zap.Int64("triggerTaskID", triggerTask.ID()),
zap.Int64("triggerTaskID", triggerTask.getTaskID()),
zap.Error(err))
triggerTask.SetResultInfo(err)
triggerTask.setResultInfo(err)
return
}
rt.SetID(id)
log.Debug("waitActivateTaskDone: reScheduler set id", zap.Int64("id", rt.ID()))
taskKey := fmt.Sprintf("%s/%d", activeTaskPrefix, rt.ID())
blobs, err := rt.Marshal()
rt.setTaskID(id)
log.Debug("waitActivateTaskDone: reScheduler set id", zap.Int64("id", rt.getTaskID()))
taskKey := fmt.Sprintf("%s/%d", activeTaskPrefix, rt.getTaskID())
blobs, err := rt.marshal()
if err != nil {
log.Error("waitActivateTaskDone: error when marshal active task",
zap.Int64("triggerTaskID", triggerTask.ID()),
zap.Int64("triggerTaskID", triggerTask.getTaskID()),
zap.Error(err))
triggerTask.SetResultInfo(err)
triggerTask.setResultInfo(err)
return
}
saves[taskKey] = string(blobs)
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, rt.ID())
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, rt.getTaskID())
saves[stateKey] = strconv.Itoa(int(taskUndo))
}
}
//TODO::queryNode auto watch queryChannel, then update etcd use same id directly
err = scheduler.client.MultiSaveAndRemove(saves, removes)
if err != nil {
log.Error("waitActivateTaskDone: error when save and remove task from etcd", zap.Int64("triggerTaskID", triggerTask.ID()))
triggerTask.SetResultInfo(err)
log.Error("waitActivateTaskDone: error when save and remove task from etcd", zap.Int64("triggerTaskID", triggerTask.getTaskID()))
triggerTask.setResultInfo(err)
return
}
triggerTask.RemoveChildTaskByID(t.ID())
triggerTask.removeChildTaskByID(t.getTaskID())
log.Debug("waitActivateTaskDone: delete failed active task and save reScheduled task to etcd",
zap.Int64("triggerTaskID", triggerTask.ID()),
zap.Int64("failed taskID", t.ID()),
zap.Int64("triggerTaskID", triggerTask.getTaskID()),
zap.Int64("failed taskID", t.getTaskID()),
zap.Any("reScheduled tasks", reScheduledTasks))
for _, rt := range reScheduledTasks {
if rt != nil {
triggerTask.AddChildTask(rt)
log.Debug("waitActivateTaskDone: add a reScheduled active task to activateChan", zap.Int64("taskID", rt.ID()))
triggerTask.addChildTask(rt)
log.Debug("waitActivateTaskDone: add a reScheduled active task to activateChan", zap.Int64("taskID", rt.getTaskID()))
scheduler.activateTaskChan <- rt
wg.Add(1)
go scheduler.waitActivateTaskDone(wg, rt, triggerTask)
......@@ -736,8 +736,8 @@ func (scheduler *TaskScheduler) waitActivateTaskDone(wg *sync.WaitGroup, t task,
//delete task from etcd
} else {
log.Debug("waitActivateTaskDone: retry the active task",
zap.Int64("taskID", t.ID()),
zap.Int64("triggerTaskID", triggerTask.ID()))
zap.Int64("taskID", t.getTaskID()),
zap.Int64("triggerTaskID", triggerTask.getTaskID()))
scheduler.activateTaskChan <- t
wg.Add(1)
go scheduler.waitActivateTaskDone(wg, t, triggerTask)
......@@ -745,30 +745,30 @@ func (scheduler *TaskScheduler) waitActivateTaskDone(wg *sync.WaitGroup, t task,
}
redoFunc2 := func(err error) {
if t.IsValid() {
if !t.IsRetryable() {
if t.isValid() {
if !t.isRetryable() {
log.Error("waitActivateTaskDone: activate task failed after retry",
zap.Int64("taskID", t.ID()),
zap.Int64("triggerTaskID", triggerTask.ID()))
triggerTask.SetResultInfo(err)
zap.Int64("taskID", t.getTaskID()),
zap.Int64("triggerTaskID", triggerTask.getTaskID()))
triggerTask.setResultInfo(err)
return
}
log.Debug("waitActivateTaskDone: retry the active task",
zap.Int64("taskID", t.ID()),
zap.Int64("triggerTaskID", triggerTask.ID()))
zap.Int64("taskID", t.getTaskID()),
zap.Int64("triggerTaskID", triggerTask.getTaskID()))
scheduler.activateTaskChan <- t
wg.Add(1)
go scheduler.waitActivateTaskDone(wg, t, triggerTask)
}
}
err = t.WaitToFinish()
err = t.waitToFinish()
if err != nil {
log.Debug("waitActivateTaskDone: activate task return err",
zap.Int64("taskID", t.ID()),
zap.Int64("triggerTaskID", triggerTask.ID()),
zap.Int64("taskID", t.getTaskID()),
zap.Int64("triggerTaskID", triggerTask.getTaskID()),
zap.Error(err))
switch t.Type() {
switch t.msgType() {
case commonpb.MsgType_LoadSegments:
redoFunc1()
case commonpb.MsgType_WatchDmChannels:
......@@ -786,8 +786,8 @@ func (scheduler *TaskScheduler) waitActivateTaskDone(wg *sync.WaitGroup, t task,
}
} else {
log.Debug("waitActivateTaskDone: one activate task done",
zap.Int64("taskID", t.ID()),
zap.Int64("triggerTaskID", triggerTask.ID()))
zap.Int64("taskID", t.getTaskID()),
zap.Int64("triggerTaskID", triggerTask.getTaskID()))
}
}
......@@ -801,14 +801,14 @@ func (scheduler *TaskScheduler) processActivateTaskLoop() {
case t := <-scheduler.activateTaskChan:
if t == nil {
log.Error("processActivateTaskLoop: pop a nil active task", zap.Int64("taskID", t.ID()))
log.Error("processActivateTaskLoop: pop a nil active task", zap.Int64("taskID", t.getTaskID()))
continue
}
log.Debug("processActivateTaskLoop: pop a active task from activateChan", zap.Int64("taskID", t.ID()))
log.Debug("processActivateTaskLoop: pop a active task from activateChan", zap.Int64("taskID", t.getTaskID()))
go func() {
err := scheduler.processTask(t)
t.Notify(err)
t.notify(err)
}()
}
}
......
internal/querycoord/task_scheduler_test.go
浏览文件 @ e80fbd1f
......@@ -32,29 +32,29 @@ type testTask struct {
nodeID int64
}
func (tt *testTask) MsgBase() *commonpb.MsgBase {
func (tt *testTask) msgBase() *commonpb.MsgBase {
return tt.baseMsg
}
func (tt *testTask) Marshal() ([]byte, error) {
func (tt *testTask) marshal() ([]byte, error) {
return []byte{}, nil
}
func (tt *testTask) Type() commonpb.MsgType {
func (tt *testTask) msgType() commonpb.MsgType {
return tt.baseMsg.MsgType
}
func (tt *testTask) Timestamp() Timestamp {
func (tt *testTask) timestamp() Timestamp {
return tt.baseMsg.Timestamp
}
func (tt *testTask) PreExecute(ctx context.Context) error {
tt.SetResultInfo(nil)
func (tt *testTask) preExecute(ctx context.Context) error {
tt.setResultInfo(nil)
log.Debug("test task preExecute...")
return nil
}
func (tt *testTask) Execute(ctx context.Context) error {
func (tt *testTask) execute(ctx context.Context) error {
log.Debug("test task execute...")
switch tt.baseMsg.MsgType {
......@@ -75,7 +75,7 @@ func (tt *testTask) Execute(ctx context.Context) error {
cluster: tt.cluster,
excludeNodeIDs: []int64{},
}
tt.AddChildTask(childTask)
tt.addChildTask(childTask)
case commonpb.MsgType_WatchDmChannels:
childTask := &WatchDmChannelTask{
BaseTask: &BaseTask{
......@@ -93,7 +93,7 @@ func (tt *testTask) Execute(ctx context.Context) error {
meta: tt.meta,
excludeNodeIDs: []int64{},
}
tt.AddChildTask(childTask)
tt.addChildTask(childTask)
case commonpb.MsgType_WatchQueryChannels:
childTask := &WatchQueryChannelTask{
BaseTask: &BaseTask{
......@@ -109,13 +109,13 @@ func (tt *testTask) Execute(ctx context.Context) error {
},
cluster: tt.cluster,
}
tt.AddChildTask(childTask)
tt.addChildTask(childTask)
}
return nil
}
func (tt *testTask) PostExecute(ctx context.Context) error {
func (tt *testTask) postExecute(ctx context.Context) error {
log.Debug("test task postExecute...")
return nil
}
......@@ -181,7 +181,7 @@ func TestUnMarshalTask(t *testing.T) {
},
},
}
blobs, err := loadTask.Marshal()
blobs, err := loadTask.marshal()
assert.Nil(t, err)
err = kv.Save("testMarshalLoadCollection", string(blobs))
assert.Nil(t, err)
......@@ -191,7 +191,7 @@ func TestUnMarshalTask(t *testing.T) {
task, err := taskScheduler.unmarshalTask(1000, value)
assert.Nil(t, err)
assert.Equal(t, task.Type(), commonpb.MsgType_LoadCollection)
assert.Equal(t, task.msgType(), commonpb.MsgType_LoadCollection)
})
t.Run("Test LoadPartitionsTask", func(t *testing.T) {
......@@ -202,7 +202,7 @@ func TestUnMarshalTask(t *testing.T) {
},
},
}
blobs, err := loadTask.Marshal()
blobs, err := loadTask.marshal()
assert.Nil(t, err)
err = kv.Save("testMarshalLoadPartition", string(blobs))
assert.Nil(t, err)
......@@ -212,7 +212,7 @@ func TestUnMarshalTask(t *testing.T) {
task, err := taskScheduler.unmarshalTask(1001, value)
assert.Nil(t, err)
assert.Equal(t, task.Type(), commonpb.MsgType_LoadPartitions)
assert.Equal(t, task.msgType(), commonpb.MsgType_LoadPartitions)
})
t.Run("Test ReleaseCollectionTask", func(t *testing.T) {
......@@ -223,7 +223,7 @@ func TestUnMarshalTask(t *testing.T) {
},
},
}
blobs, err := releaseTask.Marshal()
blobs, err := releaseTask.marshal()
assert.Nil(t, err)
err = kv.Save("testMarshalReleaseCollection", string(blobs))
assert.Nil(t, err)
......@@ -233,7 +233,7 @@ func TestUnMarshalTask(t *testing.T) {
task, err := taskScheduler.unmarshalTask(1002, value)
assert.Nil(t, err)
assert.Equal(t, task.Type(), commonpb.MsgType_ReleaseCollection)
assert.Equal(t, task.msgType(), commonpb.MsgType_ReleaseCollection)
})
t.Run("Test ReleasePartitionTask", func(t *testing.T) {
......@@ -244,7 +244,7 @@ func TestUnMarshalTask(t *testing.T) {
},
},
}
blobs, err := releaseTask.Marshal()
blobs, err := releaseTask.marshal()
assert.Nil(t, err)
err = kv.Save("testMarshalReleasePartition", string(blobs))
assert.Nil(t, err)
......@@ -254,7 +254,7 @@ func TestUnMarshalTask(t *testing.T) {
task, err := taskScheduler.unmarshalTask(1003, value)
assert.Nil(t, err)
assert.Equal(t, task.Type(), commonpb.MsgType_ReleasePartitions)
assert.Equal(t, task.msgType(), commonpb.MsgType_ReleasePartitions)
})
t.Run("Test LoadSegmentTask", func(t *testing.T) {
......@@ -265,7 +265,7 @@ func TestUnMarshalTask(t *testing.T) {
},
},
}
blobs, err := loadTask.Marshal()
blobs, err := loadTask.marshal()
assert.Nil(t, err)
err = kv.Save("testMarshalLoadSegment", string(blobs))
assert.Nil(t, err)
......@@ -275,7 +275,7 @@ func TestUnMarshalTask(t *testing.T) {
task, err := taskScheduler.unmarshalTask(1004, value)
assert.Nil(t, err)
assert.Equal(t, task.Type(), commonpb.MsgType_LoadSegments)
assert.Equal(t, task.msgType(), commonpb.MsgType_LoadSegments)
})
t.Run("Test ReleaseSegmentTask", func(t *testing.T) {
......@@ -286,7 +286,7 @@ func TestUnMarshalTask(t *testing.T) {
},
},
}
blobs, err := releaseTask.Marshal()
blobs, err := releaseTask.marshal()
assert.Nil(t, err)
err = kv.Save("testMarshalReleaseSegment", string(blobs))
assert.Nil(t, err)
......@@ -296,7 +296,7 @@ func TestUnMarshalTask(t *testing.T) {
task, err := taskScheduler.unmarshalTask(1005, value)
assert.Nil(t, err)
assert.Equal(t, task.Type(), commonpb.MsgType_ReleaseSegments)
assert.Equal(t, task.msgType(), commonpb.MsgType_ReleaseSegments)
})
t.Run("Test WatchDmChannelTask", func(t *testing.T) {
......@@ -307,7 +307,7 @@ func TestUnMarshalTask(t *testing.T) {
},
},
}
blobs, err := watchTask.Marshal()
blobs, err := watchTask.marshal()
assert.Nil(t, err)
err = kv.Save("testMarshalWatchDmChannel", string(blobs))
assert.Nil(t, err)
......@@ -317,7 +317,7 @@ func TestUnMarshalTask(t *testing.T) {
task, err := taskScheduler.unmarshalTask(1006, value)
assert.Nil(t, err)
assert.Equal(t, task.Type(), commonpb.MsgType_WatchDmChannels)
assert.Equal(t, task.msgType(), commonpb.MsgType_WatchDmChannels)
})
t.Run("Test WatchQueryChannelTask", func(t *testing.T) {
......@@ -328,7 +328,7 @@ func TestUnMarshalTask(t *testing.T) {
},
},
}
blobs, err := watchTask.Marshal()
blobs, err := watchTask.marshal()
assert.Nil(t, err)
err = kv.Save("testMarshalWatchQueryChannel", string(blobs))
assert.Nil(t, err)
......@@ -338,7 +338,7 @@ func TestUnMarshalTask(t *testing.T) {
task, err := taskScheduler.unmarshalTask(1007, value)
assert.Nil(t, err)
assert.Equal(t, task.Type(), commonpb.MsgType_WatchQueryChannels)
assert.Equal(t, task.msgType(), commonpb.MsgType_WatchQueryChannels)
})
t.Run("Test LoadBalanceTask", func(t *testing.T) {
......@@ -350,7 +350,7 @@ func TestUnMarshalTask(t *testing.T) {
},
}
blobs, err := loadBalanceTask.Marshal()
blobs, err := loadBalanceTask.marshal()
assert.Nil(t, err)
err = kv.Save("testMarshalLoadBalanceTask", string(blobs))
assert.Nil(t, err)
......@@ -360,7 +360,7 @@ func TestUnMarshalTask(t *testing.T) {
task, err := taskScheduler.unmarshalTask(1008, value)
assert.Nil(t, err)
assert.Equal(t, task.Type(), commonpb.MsgType_LoadBalanceSegments)
assert.Equal(t, task.msgType(), commonpb.MsgType_LoadBalanceSegments)
})
taskScheduler.Close()
......@@ -387,7 +387,7 @@ func TestReloadTaskFromKV(t *testing.T) {
},
},
}
triggerBlobs, err := triggerTask.Marshal()
triggerBlobs, err := triggerTask.marshal()
assert.Nil(t, err)
triggerTaskKey := fmt.Sprintf("%s/%d", triggerTaskPrefix, 100)
kvs[triggerTaskKey] = string(triggerBlobs)
......@@ -400,7 +400,7 @@ func TestReloadTaskFromKV(t *testing.T) {
},
},
}
activeBlobs, err := activeTask.Marshal()
activeBlobs, err := activeTask.marshal()
assert.Nil(t, err)
activeTaskKey := fmt.Sprintf("%s/%d", activeTaskPrefix, 101)
kvs[activeTaskKey] = string(activeBlobs)
......@@ -413,6 +413,6 @@ func TestReloadTaskFromKV(t *testing.T) {
taskScheduler.reloadFromKV()
task := taskScheduler.triggerTaskQueue.PopTask()
assert.Equal(t, taskDone, task.State())
assert.Equal(t, 1, len(task.GetChildTask()))
assert.Equal(t, taskDone, task.getState())
assert.Equal(t, 1, len(task.getChildTask()))
}
internal/querycoord/task_test.go
浏览文件 @ e80fbd1f
......@@ -159,10 +159,10 @@ func genWatchDmChannelTask(ctx context.Context, queryCoord *QueryCoord, nodeID i
meta: queryCoord.meta,
cluster: queryCoord.cluster,
}
parentTask.SetState(taskDone)
parentTask.SetResultInfo(nil)
parentTask.AddChildTask(watchDmChannelTask)
watchDmChannelTask.SetParentTask(parentTask)
parentTask.setState(taskDone)
parentTask.setResultInfo(nil)
parentTask.addChildTask(watchDmChannelTask)
watchDmChannelTask.setParentTask(parentTask)
queryCoord.meta.addCollection(defaultCollectionID, schema)
return watchDmChannelTask
......@@ -209,10 +209,10 @@ func genLoadSegmentTask(ctx context.Context, queryCoord *QueryCoord, nodeID int6
meta: queryCoord.meta,
cluster: queryCoord.cluster,
}
parentTask.SetState(taskDone)
parentTask.SetResultInfo(nil)
parentTask.AddChildTask(loadSegmentTask)
loadSegmentTask.SetParentTask(parentTask)
parentTask.setState(taskDone)
parentTask.setResultInfo(nil)
parentTask.addChildTask(loadSegmentTask)
loadSegmentTask.setParentTask(parentTask)
queryCoord.meta.addCollection(defaultCollectionID, schema)
return loadSegmentTask
......@@ -220,7 +220,7 @@ func genLoadSegmentTask(ctx context.Context, queryCoord *QueryCoord, nodeID int6
func waitTaskFinalState(t task, state taskState) {
for {
if t.State() == state {
if t.getState() == state {
break
}
}
......@@ -291,7 +291,7 @@ func Test_LoadCollectionAfterLoadPartition(t *testing.T) {
err = queryCoord.scheduler.Enqueue(releaseCollectionTask)
assert.Nil(t, err)
err = releaseCollectionTask.WaitToFinish()
err = releaseCollectionTask.waitToFinish()
assert.Nil(t, err)
node.stop()
......@@ -323,7 +323,7 @@ func Test_RepeatLoadCollection(t *testing.T) {
err = queryCoord.scheduler.Enqueue(releaseCollectionTask)
assert.Nil(t, err)
err = releaseCollectionTask.WaitToFinish()
err = releaseCollectionTask.waitToFinish()
assert.Nil(t, err)
node.stop()
......@@ -342,7 +342,7 @@ func Test_LoadCollectionAssignTaskFail(t *testing.T) {
err = queryCoord.scheduler.Enqueue(loadCollectionTask)
assert.Nil(t, err)
err = loadCollectionTask.WaitToFinish()
err = loadCollectionTask.waitToFinish()
assert.NotNil(t, err)
queryCoord.Stop()
......@@ -384,7 +384,7 @@ func Test_LoadPartitionAssignTaskFail(t *testing.T) {
err = queryCoord.scheduler.Enqueue(loadPartitionTask)
assert.Nil(t, err)
err = loadPartitionTask.WaitToFinish()
err = loadPartitionTask.waitToFinish()
assert.NotNil(t, err)
queryCoord.Stop()
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或