mirror of
https://gitee.com/milvus-io/milvus.git
synced 2024-12-04 21:09:06 +08:00
f469a315d6
* Refactor the logic of assign tasks in IndexCoord Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Fix bugs Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Fix bug for unittest Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Add lock for map Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Improve code Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Fix unittest bug Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Reduce duriation for timetick Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Update orm version Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Reset sdk version Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Fix bug Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Reset orm version Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Reset test ip Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Fix bug Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Fix bug for unissued Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Rename some variables Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Fix bug Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com> * Use break instead of continue in select::case Signed-off-by: xiaocai2333 <cai.zhang@zilliz.com>
121 lines
2.7 KiB
Go
121 lines
2.7 KiB
Go
// Copyright (C) 2019-2020 Zilliz. All rights reserved.
|
|
//
|
|
// 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 indexcoord
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
|
|
"github.com/milvus-io/milvus/internal/log"
|
|
"go.uber.org/zap"
|
|
|
|
"github.com/milvus-io/milvus/internal/allocator"
|
|
"github.com/milvus-io/milvus/internal/proto/indexpb"
|
|
)
|
|
|
|
const (
|
|
IndexAddTaskName = "IndexAddTask"
|
|
)
|
|
|
|
type task interface {
|
|
Ctx() context.Context
|
|
ID() UniqueID // return ReqID
|
|
SetID(uid UniqueID) // set ReqID
|
|
Name() string
|
|
PreExecute(ctx context.Context) error
|
|
Execute(ctx context.Context) error
|
|
PostExecute(ctx context.Context) error
|
|
WaitToFinish() error
|
|
Notify(err error)
|
|
OnEnqueue() error
|
|
}
|
|
|
|
type BaseTask struct {
|
|
done chan error
|
|
ctx context.Context
|
|
id UniqueID
|
|
table *metaTable
|
|
}
|
|
|
|
func (bt *BaseTask) ID() UniqueID {
|
|
return bt.id
|
|
}
|
|
|
|
func (bt *BaseTask) setID(id UniqueID) {
|
|
bt.id = id
|
|
}
|
|
|
|
func (bt *BaseTask) WaitToFinish() error {
|
|
select {
|
|
case <-bt.ctx.Done():
|
|
return errors.New("Task wait to finished timeout")
|
|
case err := <-bt.done:
|
|
return err
|
|
}
|
|
}
|
|
|
|
func (bt *BaseTask) Notify(err error) {
|
|
bt.done <- err
|
|
}
|
|
|
|
type IndexAddTask struct {
|
|
BaseTask
|
|
req *indexpb.BuildIndexRequest
|
|
indexBuildID UniqueID
|
|
idAllocator *allocator.GlobalIDAllocator
|
|
}
|
|
|
|
func (it *IndexAddTask) Ctx() context.Context {
|
|
return it.ctx
|
|
}
|
|
|
|
func (it *IndexAddTask) ID() UniqueID {
|
|
return it.id
|
|
}
|
|
|
|
func (it *IndexAddTask) SetID(ID UniqueID) {
|
|
it.BaseTask.setID(ID)
|
|
}
|
|
|
|
func (it *IndexAddTask) Name() string {
|
|
return IndexAddTaskName
|
|
}
|
|
|
|
func (it *IndexAddTask) OnEnqueue() error {
|
|
var err error
|
|
it.indexBuildID, err = it.idAllocator.AllocOne()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (it *IndexAddTask) PreExecute(ctx context.Context) error {
|
|
log.Debug("IndexCoord IndexAddTask PreExecute", zap.Any("IndexBuildID", it.indexBuildID))
|
|
it.req.IndexBuildID = it.indexBuildID
|
|
return nil
|
|
}
|
|
|
|
func (it *IndexAddTask) Execute(ctx context.Context) error {
|
|
log.Debug("IndexCoord IndexAddTask Execute", zap.Any("IndexBuildID", it.indexBuildID))
|
|
err := it.table.AddIndex(it.indexBuildID, it.req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (it *IndexAddTask) PostExecute(ctx context.Context) error {
|
|
log.Debug("IndexCoord IndexAddTask PostExecute", zap.Any("IndexBuildID", it.indexBuildID))
|
|
return nil
|
|
}
|