Skip to content

milvus
milvus

BaiXuePrincess / milvus
与 Fork 源项目一致

Fork自 milvus / milvus

通知 7
Star 4
Fork 0
milvus
milvus

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

提交 573c9cf8 编写于 作者: W wxyu
浏览文件
操作

MS-555 Remove old scheduler 


Former-commit-id: 4dc119775c6ea4605bcb9366d6933de1633916d8
上级 4d484de8
隐藏空白更改
内联 并排
Showing with 184 addition and 1524 deletion
cpp/CHANGELOG.md
浏览文件 @ 573c9cf8
......@@ -13,6 +13,7 @@ Please mark all change in change log and use the ticket from JIRA.
- MS-556 - Add Job Definition in Scheduler
- MS-558 - Refine status code
- MS-562 - Add JobMgr and TaskCreator in Scheduler
- MS-555 - Remove old scheduler
## New Feature
......
cpp/src/db/DBImpl.cpp
浏览文件 @ 573c9cf8
......@@ -24,8 +24,8 @@
#include "meta/MetaFactory.h"
#include "meta/MetaConsts.h"
#include "metrics/Metrics.h"
#include "scheduler/TaskScheduler.h"
#include "scheduler/context/DeleteContext.h"
#include "scheduler/job/SearchJob.h"
#include "scheduler/job/DeleteJob.h"
#include "scheduler/SchedInst.h"
#include "utils/TimeRecorder.h"
#include "utils/Log.h"
......@@ -133,12 +133,10 @@ Status DBImpl::DeleteTable(const std::string& table_id, const meta::DatesT& date
meta_ptr_->DeleteTable(table_id); //soft delete table
//scheduler will determine when to delete table files
TaskScheduler& scheduler = TaskScheduler::GetInstance();
DeleteContextPtr context = std::make_shared<DeleteContext>(table_id,
meta_ptr_,
ResMgrInst::GetInstance()->GetNumOfComputeResource());
scheduler.Schedule(context);
context->WaitAndDelete();
auto nres = ResMgrInst::GetInstance()->GetNumOfComputeResource();
scheduler::DeleteJobPtr job = std::make_shared<scheduler::DeleteJob>(0, table_id, meta_ptr_, nres);
JobMgrInst::GetInstance()->Put(job);
job->WaitAndDelete();
} else {
meta_ptr_->DropPartitionsByDates(table_id, dates);
}
......@@ -418,51 +416,50 @@ Status DBImpl::Size(uint64_t& result) {
Status DBImpl::QueryAsync(const std::string& table_id, const meta::TableFilesSchema& files,
uint64_t k, uint64_t nq, uint64_t nprobe, const float* vectors,
const meta::DatesT& dates, QueryResults& results) {
using namespace scheduler;
server::CollectQueryMetrics metrics(nq);
TimeRecorder rc("");
//step 1: get files to search
ENGINE_LOG_DEBUG << "Engine query begin, index file count: " << files.size() << " date range count: " << dates.size();
SearchContextPtr context = std::make_shared<SearchContext>(k, nq, nprobe, vectors);
for (auto &file : files) {
SearchJobPtr job = std::make_shared<SearchJob>(0, k, nq, nprobe, vectors);
for (auto &file : files) {
TableFileSchemaPtr file_ptr = std::make_shared<meta::TableFileSchema>(file);
context->AddIndexFile(file_ptr);
job->AddIndexFile(file_ptr);
}
//step 2: put search task to scheduler
TaskScheduler& scheduler = TaskScheduler::GetInstance();
scheduler.Schedule(context);
context->WaitResult();
if (!context->GetStatus().ok()) {
return context->GetStatus();
JobMgrInst::GetInstance()->Put(job);
job->WaitResult();
if (!job->GetStatus().ok()) {
return job->GetStatus();
}
//step 3: print time cost information
double load_cost = context->LoadCost();
double search_cost = context->SearchCost();
double reduce_cost = context->ReduceCost();
std::string load_info = TimeRecorder::GetTimeSpanStr(load_cost);
std::string search_info = TimeRecorder::GetTimeSpanStr(search_cost);
std::string reduce_info = TimeRecorder::GetTimeSpanStr(reduce_cost);
if(search_cost > 0.0 || reduce_cost > 0.0) {
double total_cost = load_cost + search_cost + reduce_cost;
double load_percent = load_cost/total_cost;
double search_percent = search_cost/total_cost;
double reduce_percent = reduce_cost/total_cost;
ENGINE_LOG_DEBUG << "Engine load index totally cost: " << load_info << " percent: " << load_percent*100 << "%";
ENGINE_LOG_DEBUG << "Engine search index totally cost: " << search_info << " percent: " << search_percent*100 << "%";
ENGINE_LOG_DEBUG << "Engine reduce topk totally cost: " << reduce_info << " percent: " << reduce_percent*100 << "%";
} else {
ENGINE_LOG_DEBUG << "Engine load cost: " << load_info
<< " search cost: " << search_info
<< " reduce cost: " << reduce_info;
}
// double load_cost = context->LoadCost();
// double search_cost = context->SearchCost();
// double reduce_cost = context->ReduceCost();
// std::string load_info = TimeRecorder::GetTimeSpanStr(load_cost);
// std::string search_info = TimeRecorder::GetTimeSpanStr(search_cost);
// std::string reduce_info = TimeRecorder::GetTimeSpanStr(reduce_cost);
// if(search_cost > 0.0 || reduce_cost > 0.0) {
// double total_cost = load_cost + search_cost + reduce_cost;
// double load_percent = load_cost/total_cost;
// double search_percent = search_cost/total_cost;
// double reduce_percent = reduce_cost/total_cost;
//
// ENGINE_LOG_DEBUG << "Engine load index totally cost: " << load_info << " percent: " << load_percent*100 << "%";
// ENGINE_LOG_DEBUG << "Engine search index totally cost: " << search_info << " percent: " << search_percent*100 << "%";
// ENGINE_LOG_DEBUG << "Engine reduce topk totally cost: " << reduce_info << " percent: " << reduce_percent*100 << "%";
// } else {
// ENGINE_LOG_DEBUG << "Engine load cost: " << load_info
// << " search cost: " << search_info
// << " reduce cost: " << reduce_info;
// }
//step 4: construct results
results = context->GetResult();
results = job->GetResult();
rc.ElapseFromBegin("Engine query totally cost");
return Status::OK();
......
cpp/src/db/DBImpl.h
浏览文件 @ 573c9cf8
......@@ -29,7 +29,6 @@
#include <thread>
#include <list>
#include <set>
#include "scheduler/context/SearchContext.h"
namespace zilliz {
......
cpp/src/db/scheduler/TaskDispatchQueue.cpp 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#include "TaskDispatchQueue.h"
#include "TaskDispatchStrategy.h"
#include "utils/Error.h"
#include "utils/Log.h"
namespace zilliz {
namespace milvus {
namespace engine {
void
TaskDispatchQueue::Put(const ScheduleContextPtr &context) {
std::unique_lock <std::mutex> lock(mtx);
full_.wait(lock, [this] { return (queue_.size() < capacity_); });
if(context == nullptr) {
queue_.push_front(nullptr);
empty_.notify_all();
return;
}
TaskDispatchStrategy::Schedule(context, queue_);
empty_.notify_all();
}
ScheduleTaskPtr
TaskDispatchQueue::Take() {
std::unique_lock <std::mutex> lock(mtx);
empty_.wait(lock, [this] { return !queue_.empty(); });
ScheduleTaskPtr front(queue_.front());
queue_.pop_front();
full_.notify_all();
return front;
}
size_t
TaskDispatchQueue::Size() {
std::lock_guard <std::mutex> lock(mtx);
return queue_.size();
}
ScheduleTaskPtr
TaskDispatchQueue::Front() {
std::unique_lock <std::mutex> lock(mtx);
empty_.wait(lock, [this] { return !queue_.empty(); });
if (queue_.empty()) {
std::string error_msg = "blocking queue empty";
SERVER_LOG_ERROR << error_msg;
throw server::ServerException(SERVER_BLOCKING_QUEUE_EMPTY, error_msg);
}
ScheduleTaskPtr front(queue_.front());
return front;
}
ScheduleTaskPtr
TaskDispatchQueue::Back() {
std::unique_lock <std::mutex> lock(mtx);
empty_.wait(lock, [this] { return !queue_.empty(); });
if (queue_.empty()) {
std::string error_msg = "blocking queue empty";
SERVER_LOG_ERROR << error_msg;
throw server::ServerException(SERVER_BLOCKING_QUEUE_EMPTY, error_msg);
}
ScheduleTaskPtr back(queue_.back());
return back;
}
bool
TaskDispatchQueue::Empty() {
std::unique_lock <std::mutex> lock(mtx);
return queue_.empty();
}
void
TaskDispatchQueue::SetCapacity(const size_t capacity) {
capacity_ = (capacity > 0 ? capacity : capacity_);
}
}
}
}
\ No newline at end of file
cpp/src/db/scheduler/TaskDispatchQueue.h 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#pragma once
#include "context/IScheduleContext.h"
#include "task/IScheduleTask.h"
#include <condition_variable>
#include <iostream>
#include <queue>
#include <list>
namespace zilliz {
namespace milvus {
namespace engine {
class TaskDispatchQueue {
public:
TaskDispatchQueue() : mtx(), full_(), empty_() {}
TaskDispatchQueue(const TaskDispatchQueue &rhs) = delete;
TaskDispatchQueue &operator=(const TaskDispatchQueue &rhs) = delete;
using TaskList = std::list<ScheduleTaskPtr>;
void Put(const ScheduleContextPtr &context);
ScheduleTaskPtr Take();
ScheduleTaskPtr Front();
ScheduleTaskPtr Back();
size_t Size();
bool Empty();
void SetCapacity(const size_t capacity);
private:
mutable std::mutex mtx;
std::condition_variable full_;
std::condition_variable empty_;
TaskList queue_;
size_t capacity_ = 1000000;
};
}
}
}
cpp/src/db/scheduler/TaskDispatchStrategy.cpp 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#include "TaskDispatchStrategy.h"
#include "context/SearchContext.h"
#include "context/DeleteContext.h"
#include "task/IndexLoadTask.h"
#include "task/DeleteTask.h"
#include "cache/CpuCacheMgr.h"
#include "utils/Error.h"
#include "utils/Log.h"
namespace zilliz {
namespace milvus {
namespace engine {
class ReuseCacheIndexStrategy {
public:
bool Schedule(const SearchContextPtr &context, std::list<ScheduleTaskPtr>& task_list) {
if(context == nullptr) {
ENGINE_LOG_ERROR << "Task Dispatch context doesn't exist";
return false;
}
SearchContext::Id2IndexMap index_files = context->GetIndexMap();
//some index loader alread exists
for(auto& task : task_list) {
if(task->type() != ScheduleTaskType::kIndexLoad) {
continue;
}
IndexLoadTaskPtr loader = std::static_pointer_cast<IndexLoadTask>(task);
if(index_files.find(loader->file_->id_) != index_files.end()){
ENGINE_LOG_DEBUG << "Append SearchContext to exist IndexLoaderContext";
index_files.erase(loader->file_->id_);
loader->search_contexts_.push_back(context);
}
}
//index_files still contains some index files, create new loader
for(auto& pair : index_files) {
ENGINE_LOG_DEBUG << "Create new IndexLoaderContext for: " << pair.second->location_;
IndexLoadTaskPtr new_loader = std::make_shared<IndexLoadTask>();
new_loader->search_contexts_.push_back(context);
new_loader->file_ = pair.second;
auto index = zilliz::milvus::cache::CpuCacheMgr::GetInstance()->GetIndex(pair.second->location_);
if(index != nullptr) {
//if the index file has been in memory, increase its priority
task_list.push_front(new_loader);
} else {
//index file not in memory, put it to tail
task_list.push_back(new_loader);
}
}
return true;
}
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class DeleteTableStrategy {
public:
bool Schedule(const DeleteContextPtr &context, std::list<ScheduleTaskPtr> &task_list) {
if (context == nullptr) {
ENGINE_LOG_ERROR << "Task Dispatch context doesn't exist";
return false;
}
DeleteTaskPtr delete_task = std::make_shared<DeleteTask>(context);
if(task_list.empty()) {
task_list.push_back(delete_task);
return true;
}
std::string table_id = context->table_id();
//put delete task to proper position
//for example: task_list has 10 IndexLoadTask, only the No.5 IndexLoadTask is for table1
//if user want to delete table1, the DeleteTask will be insert into No.6 position
for(std::list<ScheduleTaskPtr>::reverse_iterator iter = task_list.rbegin(); iter != task_list.rend(); ++iter) {
if((*iter)->type() != ScheduleTaskType::kIndexLoad) {
continue;
}
IndexLoadTaskPtr loader = std::static_pointer_cast<IndexLoadTask>(*iter);
if(loader->file_->table_id_ != table_id) {
continue;
}
task_list.insert(iter.base(), delete_task);
return true;
}
//no task is searching this table, put DeleteTask to front of list so that the table will be delete asap
task_list.push_front(delete_task);
return true;
}
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
bool TaskDispatchStrategy::Schedule(const ScheduleContextPtr &context_ptr,
std::list<zilliz::milvus::engine::ScheduleTaskPtr> &task_list) {
if(context_ptr == nullptr) {
ENGINE_LOG_ERROR << "Task Dispatch context doesn't exist";
return false;
}
switch(context_ptr->type()) {
case ScheduleContextType::kSearch: {
SearchContextPtr search_context = std::static_pointer_cast<SearchContext>(context_ptr);
ReuseCacheIndexStrategy strategy;
return strategy.Schedule(search_context, task_list);
}
case ScheduleContextType::kDelete: {
DeleteContextPtr delete_context = std::static_pointer_cast<DeleteContext>(context_ptr);
DeleteTableStrategy strategy;
return strategy.Schedule(delete_context, task_list);
}
default:
ENGINE_LOG_ERROR << "Invalid schedule task type";
return false;
}
}
}
}
}
\ No newline at end of file
cpp/src/db/scheduler/TaskDispatchStrategy.h 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#pragma once
#include "context/IScheduleContext.h"
#include "task/IScheduleTask.h"
#include <list>
namespace zilliz {
namespace milvus {
namespace engine {
class TaskDispatchStrategy {
public:
static bool Schedule(const ScheduleContextPtr &context_ptr, std::list<ScheduleTaskPtr>& task_list);
};
}
}
}
cpp/src/db/scheduler/TaskScheduler.cpp 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#include "server/ServerConfig.h"
#include "TaskScheduler.h"
#include "TaskDispatchQueue.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "db/engine/EngineFactory.h"
#include "scheduler/task/TaskConvert.h"
#include "scheduler/SchedInst.h"
#include "scheduler/ResourceFactory.h"
namespace zilliz {
namespace milvus {
namespace engine {
TaskScheduler::TaskScheduler()
: stopped_(true) {
Start();
}
TaskScheduler::~TaskScheduler() {
Stop();
}
TaskScheduler& TaskScheduler::GetInstance() {
static TaskScheduler s_instance;
return s_instance;
}
bool
TaskScheduler::Start() {
if(!stopped_) {
SERVER_LOG_INFO << "Task Scheduler isn't started";
return true;
}
stopped_ = false;
task_queue_.SetCapacity(2);
task_dispatch_thread_ = std::make_shared<std::thread>(&TaskScheduler::TaskDispatchWorker, this);
task_thread_ = std::make_shared<std::thread>(&TaskScheduler::TaskWorker, this);
return true;
}
bool
TaskScheduler::Stop() {
if(stopped_) {
SERVER_LOG_INFO << "Task Scheduler already stopped";
return true;
}
if(task_dispatch_thread_) {
task_dispatch_queue_.Put(nullptr);
task_dispatch_thread_->join();
task_dispatch_thread_ = nullptr;
}
if(task_thread_) {
task_queue_.Put(nullptr);
task_thread_->join();
task_thread_ = nullptr;
}
stopped_ = true;
return true;
}
bool
TaskScheduler::Schedule(ScheduleContextPtr context) {
task_dispatch_queue_.Put(context);
return true;
}
bool
TaskScheduler::TaskDispatchWorker() {
while(true) {
ScheduleTaskPtr task_ptr = task_dispatch_queue_.Take();
if(task_ptr == nullptr) {
SERVER_LOG_INFO << "Stop db task dispatch thread";
return true;
}
// TODO: Put task into Disk-TaskTable
auto task = TaskConvert(task_ptr);
auto disk_list = ResMgrInst::GetInstance()->GetDiskResources();
if (!disk_list.empty()) {
if (auto disk = disk_list[0].lock()) {
disk->task_table().Put(task);
}
}
}
}
bool
TaskScheduler::TaskWorker() {
while(true) {
// TODO: expected blocking forever
ScheduleTaskPtr task_ptr = task_queue_.Take();
if(task_ptr == nullptr) {
SERVER_LOG_INFO << "Stop db task worker thread";
return true;
}
//execute task
ScheduleTaskPtr next_task = task_ptr->Execute();
if(next_task != nullptr) {
task_queue_.Put(next_task);
}
}
}
}
}
}
cpp/src/db/scheduler/TaskScheduler.h 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#pragma once
#include "context/IScheduleContext.h"
#include "task/IScheduleTask.h"
#include "TaskDispatchQueue.h"
#include "utils/BlockingQueue.h"
#include <thread>
namespace zilliz {
namespace milvus {
namespace engine {
class TaskScheduler {
private:
TaskScheduler();
virtual ~TaskScheduler();
public:
static TaskScheduler& GetInstance();
bool Schedule(ScheduleContextPtr context);
private:
bool Start();
bool Stop();
bool TaskDispatchWorker();
bool TaskWorker();
private:
std::shared_ptr<std::thread> task_dispatch_thread_;
std::shared_ptr<std::thread> task_thread_;
TaskDispatchQueue task_dispatch_queue_;
using TaskQueue = server::BlockingQueue<ScheduleTaskPtr>;
TaskQueue task_queue_;
bool stopped_ = true;
};
}
}
}
cpp/src/db/scheduler/context/DeleteContext.cpp 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#include "DeleteContext.h"
namespace zilliz {
namespace milvus {
namespace engine {
DeleteContext::DeleteContext(const std::string &table_id, meta::MetaPtr &meta_ptr, uint64_t num_resource)
: IScheduleContext(ScheduleContextType::kDelete),
table_id_(table_id),
meta_ptr_(meta_ptr),
num_resource_(num_resource) {
}
void DeleteContext::WaitAndDelete() {
#ifdef NEW_SCHEDULER
std::unique_lock<std::mutex> lock(mutex_);
cv_.wait(lock, [&] { return done_resource == num_resource_; });
meta_ptr_->DeleteTableFiles(table_id_);
#endif
}
void DeleteContext::ResourceDone() {
{
std::lock_guard<std::mutex> lock(mutex_);
++done_resource;
}
cv_.notify_one();
}
}
}
}
\ No newline at end of file
cpp/src/db/scheduler/context/DeleteContext.h 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#pragma once
#include "IScheduleContext.h"
#include "db/meta/Meta.h"
#include <mutex>
#include <condition_variable>
namespace zilliz {
namespace milvus {
namespace engine {
class DeleteContext : public IScheduleContext {
public:
DeleteContext(const std::string& table_id, meta::MetaPtr& meta_ptr, uint64_t num_resource);
std::string table_id() const { return table_id_; }
meta::MetaPtr meta() const { return meta_ptr_; }
void WaitAndDelete();
void ResourceDone();
private:
std::string table_id_;
meta::MetaPtr meta_ptr_;
uint64_t num_resource_;
uint64_t done_resource = 0;
std::mutex mutex_;
std::condition_variable cv_;
};
using DeleteContextPtr = std::shared_ptr<DeleteContext>;
}
}
}
cpp/src/db/scheduler/context/IScheduleContext.h 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#pragma once
#include <memory>
namespace zilliz {
namespace milvus {
namespace engine {
enum class ScheduleContextType {
kUnknown = 0,
kSearch,
kDelete,
};
class IScheduleContext {
public:
IScheduleContext(ScheduleContextType type)
: type_(type) {
}
virtual ~IScheduleContext() = default;
ScheduleContextType type() const { return type_; }
protected:
ScheduleContextType type_;
};
using ScheduleContextPtr = std::shared_ptr<IScheduleContext>;
}
}
}
cpp/src/db/scheduler/context/SearchContext.cpp 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#include "SearchContext.h"
#include "utils/Log.h"
#include <chrono>
namespace zilliz {
namespace milvus {
namespace engine {
SearchContext::SearchContext(uint64_t topk, uint64_t nq, uint64_t nprobe, const float* vectors)
: IScheduleContext(ScheduleContextType::kSearch),
topk_(topk),
nq_(nq),
nprobe_(nprobe),
vectors_(vectors) {
//use current time to identify this context
std::chrono::system_clock::time_point tp = std::chrono::system_clock::now();
long id = tp.time_since_epoch().count();
identity_ = std::to_string(id);
}
bool
SearchContext::AddIndexFile(TableFileSchemaPtr& index_file) {
std::unique_lock <std::mutex> lock(mtx_);
if(index_file == nullptr || map_index_files_.find(index_file->id_) != map_index_files_.end()) {
return false;
}
SERVER_LOG_DEBUG << "SearchContext " << identity_ << " add index file: " << index_file->id_;
map_index_files_[index_file->id_] = index_file;
return true;
}
void
SearchContext::IndexSearchDone(size_t index_id) {
std::unique_lock <std::mutex> lock(mtx_);
map_index_files_.erase(index_id);
done_cond_.notify_all();
SERVER_LOG_DEBUG << "SearchContext " << identity_ << " finish index file: " << index_id;
}
void
SearchContext::WaitResult() {
std::unique_lock <std::mutex> lock(mtx_);
done_cond_.wait(lock, [this] { return map_index_files_.empty(); });
SERVER_LOG_DEBUG << "SearchContext " << identity_ << " all done";
}
}
}
}
\ No newline at end of file
cpp/src/db/scheduler/context/SearchContext.h 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#pragma once
#include "IScheduleContext.h"
#include "db/meta/MetaTypes.h"
#include <unordered_map>
#include <vector>
#include <memory>
#include <condition_variable>
namespace zilliz {
namespace milvus {
namespace engine {
using TableFileSchemaPtr = std::shared_ptr<meta::TableFileSchema>;
class SearchContext : public IScheduleContext {
public:
SearchContext(uint64_t topk, uint64_t nq, uint64_t nprobe, const float* vectors);
bool AddIndexFile(TableFileSchemaPtr& index_file);
uint64_t topk() const { return topk_; }
uint64_t nq() const { return nq_; }
uint64_t nprobe() const { return nprobe_; }
const float* vectors() const { return vectors_; }
using Id2IndexMap = std::unordered_map<size_t, TableFileSchemaPtr>;
const Id2IndexMap& GetIndexMap() const { return map_index_files_; }
using Id2DistanceMap = std::vector<std::pair<int64_t, double>>;
using ResultSet = std::vector<Id2DistanceMap>;
const ResultSet& GetResult() const { return result_; }
ResultSet& GetResult() { return result_; }
const std::string& Identity() const { return identity_; }
const Status& GetStatus() const { return status_; }
Status& GetStatus() { return status_; }
void IndexSearchDone(size_t index_id);
void WaitResult();
void AccumLoadCost(double span) { time_cost_load_ += span; }
void AccumSearchCost(double span) { time_cost_search_ += span; }
void AccumReduceCost(double span) { time_cost_reduce_ += span; }
double LoadCost() const { return time_cost_load_; }
double SearchCost() const { return time_cost_search_; }
double ReduceCost() const { return time_cost_reduce_; }
private:
uint64_t topk_ = 0;
uint64_t nq_ = 0;
uint64_t nprobe_ = 10;
const float* vectors_ = nullptr;
Id2IndexMap map_index_files_;
ResultSet result_;
std::mutex mtx_;
std::condition_variable done_cond_;
std::string identity_; //for debug
Status status_;
double time_cost_load_ = 0.0; //time cost for load all index files, unit: us
double time_cost_search_ = 0.0; //time cost for entire search, unit: us
double time_cost_reduce_ = 0.0; //time cost for entire reduce, unit: us
};
using SearchContextPtr = std::shared_ptr<SearchContext>;
}
}
}
cpp/src/db/scheduler/task/DeleteTask.cpp 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#include "DeleteTask.h"
namespace zilliz {
namespace milvus {
namespace engine {
DeleteTask::DeleteTask(const DeleteContextPtr& context)
: IScheduleTask(ScheduleTaskType::kDelete),
context_(context) {
}
std::shared_ptr<IScheduleTask> DeleteTask::Execute() {
return nullptr;
}
}
}
}
\ No newline at end of file
cpp/src/db/scheduler/task/DeleteTask.h 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#pragma once
#include "IScheduleTask.h"
#include "db/scheduler/context/DeleteContext.h"
namespace zilliz {
namespace milvus {
namespace engine {
class DeleteTask : public IScheduleTask {
public:
DeleteTask(const DeleteContextPtr& context);
virtual std::shared_ptr<IScheduleTask> Execute() override;
public:
DeleteContextPtr context_;
};
using DeleteTaskPtr = std::shared_ptr<DeleteTask>;
}
}
}
cpp/src/db/scheduler/task/IScheduleTask.h 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#pragma once
#include <memory>
namespace zilliz {
namespace milvus {
namespace engine {
enum class ScheduleTaskType {
kUnknown = 0,
kIndexLoad,
kSearch,
kDelete,
};
class IScheduleTask {
public:
IScheduleTask(ScheduleTaskType type)
: type_(type) {
}
virtual ~IScheduleTask() = default;
ScheduleTaskType type() const { return type_; }
virtual std::shared_ptr<IScheduleTask> Execute() = 0;
protected:
ScheduleTaskType type_;
};
using ScheduleTaskPtr = std::shared_ptr<IScheduleTask>;
}
}
}
cpp/src/db/scheduler/task/IndexLoadTask.cpp 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#include "IndexLoadTask.h"
#include "SearchTask.h"
#include "db/engine/EngineFactory.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "metrics/Metrics.h"
namespace zilliz {
namespace milvus {
namespace engine {
IndexLoadTask::IndexLoadTask()
: IScheduleTask(ScheduleTaskType::kIndexLoad) {
}
std::shared_ptr<IScheduleTask> IndexLoadTask::Execute() {
return nullptr;
}
}
}
}
\ No newline at end of file
cpp/src/db/scheduler/task/IndexLoadTask.h 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#pragma once
#include "IScheduleTask.h"
#include "db/scheduler/context/SearchContext.h"
namespace zilliz {
namespace milvus {
namespace engine {
class IndexLoadTask : public IScheduleTask {
public:
IndexLoadTask();
virtual std::shared_ptr<IScheduleTask> Execute() override;
public:
TableFileSchemaPtr file_;
std::vector<SearchContextPtr> search_contexts_;
};
using IndexLoadTaskPtr = std::shared_ptr<IndexLoadTask>;
}
}
}
cpp/src/db/scheduler/task/SearchTask.cpp 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#include "SearchTask.h"
#include "metrics/Metrics.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include <thread>
namespace zilliz {
namespace milvus {
namespace engine {
SearchTask::SearchTask()
: IScheduleTask(ScheduleTaskType::kSearch) {
}
std::shared_ptr<IScheduleTask> SearchTask::Execute() {
return nullptr;
}
}
}
}
cpp/src/db/scheduler/task/SearchTask.h 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#pragma once
#include "IScheduleTask.h"
#include "db/scheduler/context/SearchContext.h"
#include "db/engine/ExecutionEngine.h"
namespace zilliz {
namespace milvus {
namespace engine {
class SearchTask : public IScheduleTask {
public:
SearchTask();
virtual std::shared_ptr<IScheduleTask> Execute() override;
public:
size_t index_id_ = 0;
int file_type_ = 0; //for metrics
ExecutionEnginePtr index_engine_;
std::vector<SearchContextPtr> search_contexts_;
};
using SearchTaskPtr = std::shared_ptr<SearchTask>;
}
}
}
\ No newline at end of file
cpp/src/scheduler/JobMgr.h
浏览文件 @ 573c9cf8
......@@ -73,6 +73,8 @@ private:
ResourceMgrPtr res_mgr_ = nullptr;
};
using JobMgrPtr = std::shared_ptr<JobMgr>;
}
}
}
cpp/src/scheduler/SchedInst.cpp
浏览文件 @ 573c9cf8
......@@ -33,6 +33,9 @@ std::mutex ResMgrInst::mutex_;
SchedulerPtr SchedInst::instance = nullptr;
std::mutex SchedInst::mutex_;
scheduler::JobMgrPtr JobMgrInst::instance = nullptr;
std::mutex JobMgrInst::mutex_;
void
load_simple_config() {
server::ConfigNode &config = server::ServerConfig::GetInstance().GetConfig(server::CONFIG_RESOURCE);
......@@ -151,12 +154,14 @@ StartSchedulerService() {
// load_advance_config();
ResMgrInst::GetInstance()->Start();
SchedInst::GetInstance()->Start();
JobMgrInst::GetInstance()->Start();
}
void
StopSchedulerService() {
ResMgrInst::GetInstance()->Stop();
JobMgrInst::GetInstance()->Stop();
SchedInst::GetInstance()->Stop();
ResMgrInst::GetInstance()->Stop();
}
}
}
......
cpp/src/scheduler/SchedInst.h
浏览文件 @ 573c9cf8
......@@ -19,6 +19,7 @@
#include "ResourceMgr.h"
#include "Scheduler.h"
#include "JobMgr.h"
#include <mutex>
#include <memory>
......@@ -64,6 +65,24 @@ private:
static std::mutex mutex_;
};
class JobMgrInst {
public:
static scheduler::JobMgrPtr
GetInstance() {
if (instance == nullptr) {
std::lock_guard<std::mutex> lock(mutex_);
if (instance == nullptr) {
instance = std::make_shared<scheduler::JobMgr>(ResMgrInst::GetInstance());
}
}
return instance;
}
private:
static scheduler::JobMgrPtr instance;
static std::mutex mutex_;
};
void
StartSchedulerService();
......
cpp/src/scheduler/TaskCreator.cpp
浏览文件 @ 573c9cf8
......@@ -15,7 +15,9 @@
// specific language governing permissions and limitations
// under the License.
#include <src/scheduler/tasklabel/BroadcastLabel.h>
#include "TaskCreator.h"
#include "tasklabel/DefaultLabel.h"
namespace zilliz {
......@@ -43,6 +45,8 @@ TaskCreator::Create(const SearchJobPtr &job) {
std::vector<TaskPtr> tasks;
for (auto &index_file : job->index_files()) {
auto task = std::make_shared<XSearchTask>(index_file.second);
task->label() = std::make_shared<engine::DefaultLabel>();
task->job_ = job;
tasks.emplace_back(task);
}
......@@ -52,8 +56,10 @@ TaskCreator::Create(const SearchJobPtr &job) {
std::vector<TaskPtr>
TaskCreator::Create(const DeleteJobPtr &job) {
std::vector<TaskPtr> tasks;
// auto task = std::make_shared<XDeleteTask>(job);
// tasks.emplace_back(task);
auto task = std::make_shared<XDeleteTask>(job);
task->label() = std::make_shared<engine::BroadcastLabel>();
task->job_ = job;
tasks.emplace_back(task);
return tasks;
}
......
cpp/src/scheduler/job/Job.h
浏览文件 @ 573c9cf8
......@@ -62,6 +62,7 @@ private:
};
using JobPtr = std::shared_ptr<Job>;
using JobWPtr = std::weak_ptr<Job>;
}
}
......
cpp/src/scheduler/job/SearchJob.cpp
浏览文件 @ 573c9cf8
......@@ -28,7 +28,12 @@ SearchJob::SearchJob(zilliz::milvus::scheduler::JobId id,
uint64_t topk,
uint64_t nq,
uint64_t nprobe,
const float *vectors) : Job(id, JobType::SEARCH) {}
const float *vectors)
: Job(id, JobType::SEARCH),
topk_(topk),
nq_(nq),
nprobe_(nprobe),
vectors_(vectors) {}
bool
SearchJob::AddIndexFile(const TableFileSchemaPtr &index_file) {
......@@ -64,6 +69,11 @@ SearchJob::GetResult() {
return result_;
}
Status&
SearchJob::GetStatus() {
return status_;
}
}
}
......
cpp/src/scheduler/job/SearchJob.h
浏览文件 @ 573c9cf8
......@@ -59,6 +59,9 @@ public:
ResultSet &
GetResult();
Status &
GetStatus();
public:
uint64_t
topk() const {
......@@ -94,6 +97,7 @@ private:
Id2IndexMap index_files_;
// TODO: column-base better ?
ResultSet result_;
Status status_;
std::mutex mutex_;
std::condition_variable cv_;
......
cpp/src/scheduler/task/DeleteTask.cpp
浏览文件 @ 573c9cf8
......@@ -23,8 +23,8 @@ namespace zilliz {
namespace milvus {
namespace engine {
XDeleteTask::XDeleteTask(DeleteContextPtr &delete_context)
: Task(TaskType::DeleteTask), delete_context_ptr_(delete_context) {}
XDeleteTask::XDeleteTask(const scheduler::DeleteJobPtr &delete_job)
: Task(TaskType::DeleteTask), delete_job_(delete_job) {}
void
XDeleteTask::Load(LoadType type, uint8_t device_id) {
......@@ -33,7 +33,7 @@ XDeleteTask::Load(LoadType type, uint8_t device_id) {
void
XDeleteTask::Execute() {
delete_context_ptr_->ResourceDone();
delete_job_->ResourceDone();
}
}
......
cpp/src/scheduler/task/DeleteTask.h
浏览文件 @ 573c9cf8
......@@ -17,7 +17,7 @@
#pragma once
#include <src/db/scheduler/context/DeleteContext.h>
#include "scheduler/job/DeleteJob.h"
#include "Task.h"
......@@ -28,7 +28,7 @@ namespace engine {
class XDeleteTask : public Task {
public:
explicit
XDeleteTask(DeleteContextPtr &delete_context);
XDeleteTask(const scheduler::DeleteJobPtr &job);
void
Load(LoadType type, uint8_t device_id) override;
......@@ -37,7 +37,7 @@ public:
Execute() override;
public:
DeleteContextPtr delete_context_ptr_;
scheduler::DeleteJobPtr delete_job_;
};
}
......
cpp/src/scheduler/task/SearchTask.cpp
浏览文件 @ 573c9cf8
......@@ -22,6 +22,7 @@
#include "utils/Log.h"
#include <thread>
#include "scheduler/job/SearchJob.h"
namespace zilliz {
......@@ -94,7 +95,7 @@ CollectFileMetrics(int file_type, size_t file_size) {
}
}
XSearchTask::XSearchTask(TableFileSchemaPtr file)
XSearchTask::XSearchTask(meta::TableFileSchemaPtr file)
: Task(TaskType::SearchTask), file_(file) {
if (file_) {
index_engine_ = EngineFactory::Build(file_->dimension_,
......@@ -143,9 +144,10 @@ XSearchTask::Load(LoadType type, uint8_t device_id) {
s = Status(SERVER_UNEXPECTED_ERROR, error_msg);
}
for (auto &context : search_contexts_) {
context->IndexSearchDone(file_->id_);//mark as done avoid dead lock, even failed
context->GetStatus() = s;
if (auto job = job_.lock()){
auto search_job = std::static_pointer_cast<scheduler::SearchJob>(job);
search_job->SearchDone(file_->id_);
search_job->GetStatus() = s;
}
return;
......@@ -156,16 +158,16 @@ XSearchTask::Load(LoadType type, uint8_t device_id) {
std::string info = "Load file id:" + std::to_string(file_->id_) + " file type:" + std::to_string(file_->file_type_)
+ " size:" + std::to_string(file_size) + " bytes from location: " + file_->location_ + " totally cost";
double span = rc.ElapseFromBegin(info);
for (auto &context : search_contexts_) {
context->AccumLoadCost(span);
}
// for (auto &context : search_contexts_) {
// context->AccumLoadCost(span);
// }
CollectFileMetrics(file_->file_type_, file_size);
//step 2: return search task for later execution
index_id_ = file_->id_;
index_type_ = file_->file_type_;
search_contexts_.swap(search_contexts_);
// search_contexts_.swap(search_contexts_);
}
void
......@@ -174,8 +176,8 @@ XSearchTask::Execute() {
return;
}
ENGINE_LOG_DEBUG << "Searching in file id:" << index_id_ << " with "
<< search_contexts_.size() << " tasks";
// ENGINE_LOG_DEBUG << "Searching in file id:" << index_id_ << " with "
// << search_contexts_.size() << " tasks";
TimeRecorder rc("DoSearch file id:" + std::to_string(index_id_));
......@@ -183,16 +185,18 @@ XSearchTask::Execute() {
std::vector<long> output_ids;
std::vector<float> output_distance;
for (auto &context : search_contexts_) {
if (auto job = job_.lock()) {
auto search_job = std::static_pointer_cast<scheduler::SearchJob>(job);
//step 1: allocate memory
uint64_t nq = context->nq();
uint64_t topk = context->topk();
uint64_t nprobe = context->nprobe();
const float* vectors = context->vectors();
uint64_t nq = search_job->nq();
uint64_t topk = search_job->topk();
uint64_t nprobe = search_job->nprobe();
const float* vectors = search_job->vectors();
output_ids.resize(topk * nq);
output_distance.resize(topk * nq);
std::string hdr = "context " + context->Identity() +
std::string hdr = "job " + std::to_string(search_job->id()) +
" nq " + std::to_string(nq) +
" topk " + std::to_string(topk);
......@@ -201,30 +205,29 @@ XSearchTask::Execute() {
index_engine_->Search(nq, vectors, topk, nprobe, output_distance.data(), output_ids.data());
double span = rc.RecordSection(hdr + ", do search");
context->AccumSearchCost(span);
// search_job->AccumSearchCost(span);
//step 3: cluster result
SearchContext::ResultSet result_set;
scheduler::ResultSet result_set;
auto spec_k = index_engine_->Count() < topk ? index_engine_->Count() : topk;
XSearchTask::ClusterResult(output_ids, output_distance, nq, spec_k, result_set);
span = rc.RecordSection(hdr + ", cluster result");
context->AccumReduceCost(span);
// search_job->AccumReduceCost(span);
// step 4: pick up topk result
XSearchTask::TopkResult(result_set, topk, metric_l2, context->GetResult());
XSearchTask::TopkResult(result_set, topk, metric_l2, search_job->GetResult());
span = rc.RecordSection(hdr + ", reduce topk");
context->AccumReduceCost(span);
// search_job->AccumReduceCost(span);
} catch (std::exception &ex) {
ENGINE_LOG_ERROR << "SearchTask encounter exception: " << ex.what();
context->IndexSearchDone(index_id_);//mark as done avoid dead lock, even search failed
continue;
// search_job->IndexSearchDone(index_id_);//mark as done avoid dead lock, even search failed
}
//step 5: notify to send result to client
context->IndexSearchDone(index_id_);
search_job->SearchDone(index_id_);
}
rc.ElapseFromBegin("totally cost");
......@@ -237,7 +240,7 @@ Status XSearchTask::ClusterResult(const std::vector<long> &output_ids,
const std::vector<float> &output_distance,
uint64_t nq,
uint64_t topk,
SearchContext::ResultSet &result_set) {
scheduler::ResultSet &result_set) {
if (output_ids.size() < nq * topk || output_distance.size() < nq * topk) {
std::string msg = "Invalid id array size: " + std::to_string(output_ids.size()) +
" distance array size: " + std::to_string(output_distance.size());
......@@ -250,7 +253,7 @@ Status XSearchTask::ClusterResult(const std::vector<long> &output_ids,
std::function<void(size_t, size_t)> reduce_worker = [&](size_t from_index, size_t to_index) {
for (auto i = from_index; i < to_index; i++) {
SearchContext::Id2DistanceMap id_distance;
scheduler::Id2DistanceMap id_distance;
id_distance.reserve(topk);
for (auto k = 0; k < topk; k++) {
uint64_t index = i * topk + k;
......@@ -272,8 +275,8 @@ Status XSearchTask::ClusterResult(const std::vector<long> &output_ids,
return Status::OK();
}
Status XSearchTask::MergeResult(SearchContext::Id2DistanceMap &distance_src,
SearchContext::Id2DistanceMap &distance_target,
Status XSearchTask::MergeResult(scheduler::Id2DistanceMap &distance_src,
scheduler::Id2DistanceMap &distance_target,
uint64_t topk,
bool ascending) {
//Note: the score_src and score_target are already arranged by score in ascending order
......@@ -290,7 +293,7 @@ Status XSearchTask::MergeResult(SearchContext::Id2DistanceMap &distance_src,
size_t src_count = distance_src.size();
size_t target_count = distance_target.size();
SearchContext::Id2DistanceMap distance_merged;
scheduler::Id2DistanceMap distance_merged;
distance_merged.reserve(topk);
size_t src_index = 0, target_index = 0;
while (true) {
......@@ -346,10 +349,10 @@ Status XSearchTask::MergeResult(SearchContext::Id2DistanceMap &distance_src,
return Status::OK();
}
Status XSearchTask::TopkResult(SearchContext::ResultSet &result_src,
Status XSearchTask::TopkResult(scheduler::ResultSet &result_src,
uint64_t topk,
bool ascending,
SearchContext::ResultSet &result_target) {
scheduler::ResultSet &result_target) {
if (result_target.empty()) {
result_target.swap(result_src);
return Status::OK();
......@@ -363,8 +366,8 @@ Status XSearchTask::TopkResult(SearchContext::ResultSet &result_src,
std::function<void(size_t, size_t)> ReduceWorker = [&](size_t from_index, size_t to_index) {
for (size_t i = from_index; i < to_index; i++) {
SearchContext::Id2DistanceMap &score_src = result_src[i];
SearchContext::Id2DistanceMap &score_target = result_target[i];
scheduler::Id2DistanceMap &score_src = result_src[i];
scheduler::Id2DistanceMap &score_target = result_target[i];
XSearchTask::MergeResult(score_src, score_target, topk, ascending);
}
};
......
cpp/src/scheduler/task/SearchTask.h
浏览文件 @ 573c9cf8
......@@ -18,6 +18,8 @@
#pragma once
#include "Task.h"
#include "db/meta/MetaTypes.h"
#include "scheduler/job/SearchJob.h"
namespace zilliz {
......@@ -28,7 +30,7 @@ namespace engine {
class XSearchTask : public Task {
public:
explicit
XSearchTask(TableFileSchemaPtr file);
XSearchTask(meta::TableFileSchemaPtr file);
void
Load(LoadType type, uint8_t device_id) override;
......@@ -41,20 +43,20 @@ public:
const std::vector<float> &output_distence,
uint64_t nq,
uint64_t topk,
SearchContext::ResultSet &result_set);
scheduler::ResultSet &result_set);
static Status MergeResult(SearchContext::Id2DistanceMap &distance_src,
SearchContext::Id2DistanceMap &distance_target,
static Status MergeResult(scheduler::Id2DistanceMap &distance_src,
scheduler::Id2DistanceMap &distance_target,
uint64_t topk,
bool ascending);
static Status TopkResult(SearchContext::ResultSet &result_src,
static Status TopkResult(scheduler::ResultSet &result_src,
uint64_t topk,
bool ascending,
SearchContext::ResultSet &result_target);
scheduler::ResultSet &result_target);
public:
TableFileSchemaPtr file_;
meta::TableFileSchemaPtr file_;
size_t index_id_ = 0;
int index_type_ = 0;
......
cpp/src/scheduler/task/Task.h
浏览文件 @ 573c9cf8
......@@ -17,9 +17,9 @@
#pragma once
#include "db/scheduler/context/SearchContext.h"
#include "db/scheduler/task/IScheduleTask.h"
#include "scheduler/tasklabel/TaskLabel.h"
#include "scheduler/job/Job.h"
#include "utils/Status.h"
#include "Path.h"
#include <string>
......@@ -84,7 +84,8 @@ public:
public:
Path task_path_;
std::vector<SearchContextPtr> search_contexts_;
// std::vector<SearchContextPtr> search_contexts_;
scheduler::JobWPtr job_;
TaskType type_;
TaskLabelPtr label_ = nullptr;
};
......
cpp/src/scheduler/task/TaskConvert.cpp 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#include "TaskConvert.h"
#include "scheduler/tasklabel/DefaultLabel.h"
#include "scheduler/tasklabel/BroadcastLabel.h"
namespace zilliz {
namespace milvus {
namespace engine {
TaskPtr
TaskConvert(const ScheduleTaskPtr &schedule_task) {
switch (schedule_task->type()) {
case ScheduleTaskType::kIndexLoad: {
auto load_task = std::static_pointer_cast<IndexLoadTask>(schedule_task);
auto task = std::make_shared<XSearchTask>(load_task->file_);
task->label() = std::make_shared<DefaultLabel>();
task->search_contexts_ = load_task->search_contexts_;
return task;
}
case ScheduleTaskType::kDelete: {
auto delete_task = std::static_pointer_cast<DeleteTask>(schedule_task);
auto task = std::make_shared<XDeleteTask>(delete_task->context_);
task->label() = std::make_shared<BroadcastLabel>();
return task;
}
default: {
// TODO: unexpected !!!
return nullptr;
}
}
}
}
}
}
cpp/src/scheduler/task/TaskConvert.h 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#include "db/scheduler/task/DeleteTask.h"
#include "db/scheduler/task/IndexLoadTask.h"
#include "Task.h"
#include "SearchTask.h"
#include "DeleteTask.h"
namespace zilliz {
namespace milvus {
namespace engine {
TaskPtr
TaskConvert(const ScheduleTaskPtr &schedule_task);
}
}
}
cpp/src/scheduler/task/TestTask.cpp
浏览文件 @ 573c9cf8
......@@ -25,7 +25,7 @@ namespace milvus {
namespace engine {
TestTask::TestTask(TableFileSchemaPtr &file) : XSearchTask(file) {}
TestTask::TestTask(meta::TableFileSchemaPtr &file) : XSearchTask(file) {}
void
TestTask::Load(LoadType type, uint8_t device_id) {
......
cpp/src/scheduler/task/TestTask.h
浏览文件 @ 573c9cf8
......@@ -26,7 +26,8 @@ namespace engine {
class TestTask : public XSearchTask {
public:
TestTask(TableFileSchemaPtr& file);
explicit
TestTask(meta::TableFileSchemaPtr& file);
public:
void
......
cpp/unittest/db/CMakeLists.txt
浏览文件 @ 573c9cf8
......@@ -26,6 +26,7 @@ aux_source_directory(${MILVUS_ENGINE_SRC}/cache cache_srcs)
aux_source_directory(${MILVUS_ENGINE_SRC}/wrapper/knowhere knowhere_src)
aux_source_directory(${MILVUS_ENGINE_SRC}/scheduler/action scheduler_action_srcs)
aux_source_directory(${MILVUS_ENGINE_SRC}/scheduler/event scheduler_event_srcs)
aux_source_directory(${MILVUS_ENGINE_SRC}/scheduler/job scheduler_job_srcs)
aux_source_directory(${MILVUS_ENGINE_SRC}/scheduler/resource scheduler_resource_srcs)
aux_source_directory(${MILVUS_ENGINE_SRC}/scheduler/task scheduler_task_srcs)
aux_source_directory(${MILVUS_ENGINE_SRC}/scheduler scheduler_srcs)
......@@ -37,14 +38,6 @@ set(util_files
${MILVUS_ENGINE_SRC}/utils/ValidationUtil.cpp
${MILVUS_ENGINE_SRC}/utils/easylogging++.cc)
aux_source_directory(${MILVUS_ENGINE_SRC}/db/scheduler scheduler_files)
aux_source_directory(${MILVUS_ENGINE_SRC}/db/scheduler/context scheduler_context_files)
aux_source_directory(${MILVUS_ENGINE_SRC}/db/scheduler/task scheduler_task_files)
set(db_scheduler_srcs
${scheduler_files}
${scheduler_context_files}
${scheduler_task_files}
)
include_directories(/usr/local/cuda/include)
link_directories("/usr/local/cuda/lib64")
......@@ -62,6 +55,7 @@ set(db_test_src
${wrapper_src}
${scheduler_action_srcs}
${scheduler_event_srcs}
${scheduler_job_srcs}
${scheduler_resource_srcs}
${scheduler_task_srcs}
${scheduler_srcs}
......
cpp/unittest/db/scheduler_test.cpp 已删除 100644 → 0
浏览文件 @ 4d484de8
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#include <gtest/gtest.h>
#include <thread>
#include "utils/easylogging++.h"
#include <boost/filesystem.hpp>
#include "db/scheduler/TaskScheduler.h"
#include "db/scheduler/TaskDispatchStrategy.h"
#include "db/scheduler/TaskDispatchQueue.h"
#include "db/scheduler/task/SearchTask.h"
#include "db/scheduler/task/DeleteTask.h"
#include "db/scheduler/task/IndexLoadTask.h"
using namespace zilliz::milvus;
namespace {
engine::TableFileSchemaPtr CreateTabileFileStruct(size_t id, const std::string& table_id) {
auto file = std::make_shared<engine::meta::TableFileSchema>();
file->id_ = id;
file->table_id_ = table_id;
return file;
}
}
TEST(DBSchedulerTest, TASK_QUEUE_TEST) {
engine::TaskDispatchQueue queue;
queue.SetCapacity(1000);
queue.Put(nullptr);
ASSERT_EQ(queue.Size(), 1UL);
auto ptr = queue.Take();
ASSERT_EQ(ptr, nullptr);
ASSERT_TRUE(queue.Empty());
engine::SearchContextPtr context_ptr = std::make_shared<engine::SearchContext>(1, 1, 10, nullptr);
for(size_t i = 0; i < 10; i++) {
auto file = CreateTabileFileStruct(i, "tbl");
context_ptr->AddIndexFile(file);
}
queue.Put(context_ptr);
ASSERT_EQ(queue.Size(), 10);
auto index_files = context_ptr->GetIndexMap();
ptr = queue.Front();
ASSERT_EQ(ptr->type(), engine::ScheduleTaskType::kIndexLoad);
engine::IndexLoadTaskPtr load_task = std::static_pointer_cast<engine::IndexLoadTask>(ptr);
ASSERT_EQ(load_task->file_->id_, index_files.begin()->first);
ptr = queue.Back();
ASSERT_EQ(ptr->type(), engine::ScheduleTaskType::kIndexLoad);
load_task->Execute();
}
TEST(DBSchedulerTest, SEARCH_SCHEDULER_TEST) {
std::list<engine::ScheduleTaskPtr> task_list;
bool ret = engine::TaskDispatchStrategy::Schedule(nullptr, task_list);
ASSERT_FALSE(ret);
for(size_t i = 10; i < 30; i++) {
engine::IndexLoadTaskPtr task_ptr = std::make_shared<engine::IndexLoadTask>();
task_ptr->file_ = CreateTabileFileStruct(i, "tbl");
task_list.push_back(task_ptr);
}
engine::SearchContextPtr context_ptr = std::make_shared<engine::SearchContext>(1, 1, 10, nullptr);
for(size_t i = 0; i < 20; i++) {
auto file = CreateTabileFileStruct(i, "tbl");
context_ptr->AddIndexFile(file);
}
ret = engine::TaskDispatchStrategy::Schedule(context_ptr, task_list);
ASSERT_TRUE(ret);
ASSERT_EQ(task_list.size(), 30);
}
TEST(DBSchedulerTest, DELETE_SCHEDULER_TEST) {
std::list<engine::ScheduleTaskPtr> task_list;
bool ret = engine::TaskDispatchStrategy::Schedule(nullptr, task_list);
ASSERT_FALSE(ret);
const std::string table_id = "to_delete_table";
for(size_t i = 0; i < 10; i++) {
engine::IndexLoadTaskPtr task_ptr = std::make_shared<engine::IndexLoadTask>();
task_ptr->file_ = CreateTabileFileStruct(i, table_id);
task_list.push_back(task_ptr);
}
for(size_t i = 0; i < 10; i++) {
engine::IndexLoadTaskPtr task_ptr = std::make_shared<engine::IndexLoadTask>();
task_ptr->file_ = CreateTabileFileStruct(i, "other_table");
task_list.push_back(task_ptr);
}
engine::meta::MetaPtr meta_ptr;
engine::DeleteContextPtr context_ptr = std::make_shared<engine::DeleteContext>(table_id, meta_ptr, 0);
ret = engine::TaskDispatchStrategy::Schedule(context_ptr, task_list);
ASSERT_TRUE(ret);
ASSERT_EQ(task_list.size(), 21);
auto temp_list = task_list;
for(size_t i = 0; ; i++) {
engine::ScheduleTaskPtr task_ptr = temp_list.front();
temp_list.pop_front();
if(task_ptr->type() == engine::ScheduleTaskType::kDelete) {
ASSERT_EQ(i, 10);
break;
}
}
context_ptr = std::make_shared<engine::DeleteContext>("no_task_table", meta_ptr, 0);
ret = engine::TaskDispatchStrategy::Schedule(context_ptr, task_list);
ASSERT_TRUE(ret);
ASSERT_EQ(task_list.size(), 22);
engine::ScheduleTaskPtr task_ptr = task_list.front();
ASSERT_EQ(task_ptr->type(), engine::ScheduleTaskType::kDelete);
}
cpp/unittest/db/search_test.cpp
浏览文件 @ 573c9cf8
......@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
#include "db/scheduler/task/SearchTask.h"
#include "server/ServerConfig.h"
#include "utils/TimeRecorder.h"
......@@ -50,9 +49,9 @@ void BuildResult(uint64_t nq,
}
}
void CheckResult(const engine::SearchContext::Id2DistanceMap& src_1,
const engine::SearchContext::Id2DistanceMap& src_2,
const engine::SearchContext::Id2DistanceMap& target,
void CheckResult(const scheduler::Id2DistanceMap& src_1,
const scheduler::Id2DistanceMap& src_2,
const scheduler::Id2DistanceMap& target,
bool ascending) {
for(uint64_t i = 0; i < target.size() - 1; i++) {
if(ascending) {
......@@ -81,7 +80,7 @@ void CheckResult(const engine::SearchContext::Id2DistanceMap& src_1,
void CheckCluster(const std::vector<long>& target_ids,
const std::vector<float>& target_distence,
const engine::SearchContext::ResultSet& src_result,
const scheduler::ResultSet& src_result,
int64_t nq,
int64_t topk) {
ASSERT_EQ(src_result.size(), nq);
......@@ -98,7 +97,7 @@ void CheckCluster(const std::vector<long>& target_ids,
}
}
void CheckTopkResult(const engine::SearchContext::ResultSet& src_result,
void CheckTopkResult(const scheduler::ResultSet& src_result,
bool ascending,
int64_t nq,
int64_t topk) {
......@@ -127,7 +126,7 @@ TEST(DBSearchTest, TOPK_TEST) {
bool ascending = true;
std::vector<long> target_ids;
std::vector<float> target_distence;
engine::SearchContext::ResultSet src_result;
scheduler::ResultSet src_result;
auto status = engine::XSearchTask::ClusterResult(target_ids, target_distence, NQ, TOP_K, src_result);
ASSERT_FALSE(status.ok());
ASSERT_TRUE(src_result.empty());
......@@ -137,7 +136,7 @@ TEST(DBSearchTest, TOPK_TEST) {
ASSERT_TRUE(status.ok());
ASSERT_EQ(src_result.size(), NQ);
engine::SearchContext::ResultSet target_result;
scheduler::ResultSet target_result;
status = engine::XSearchTask::TopkResult(target_result, TOP_K, ascending, target_result);
ASSERT_TRUE(status.ok());
......@@ -179,7 +178,7 @@ TEST(DBSearchTest, MERGE_TEST) {
std::vector<float> target_distence;
std::vector<long> src_ids;
std::vector<float> src_distence;
engine::SearchContext::ResultSet src_result, target_result;
scheduler::ResultSet src_result, target_result;
uint64_t src_count = 5, target_count = 8;
BuildResult(1, src_count, ascending, src_ids, src_distence);
......@@ -190,8 +189,8 @@ TEST(DBSearchTest, MERGE_TEST) {
ASSERT_TRUE(status.ok());
{
engine::SearchContext::Id2DistanceMap src = src_result[0];
engine::SearchContext::Id2DistanceMap target = target_result[0];
scheduler::Id2DistanceMap src = src_result[0];
scheduler::Id2DistanceMap target = target_result[0];
status = engine::XSearchTask::MergeResult(src, target, 10, ascending);
ASSERT_TRUE(status.ok());
ASSERT_EQ(target.size(), 10);
......@@ -199,8 +198,8 @@ TEST(DBSearchTest, MERGE_TEST) {
}
{
engine::SearchContext::Id2DistanceMap src = src_result[0];
engine::SearchContext::Id2DistanceMap target;
scheduler::Id2DistanceMap src = src_result[0];
scheduler::Id2DistanceMap target;
status = engine::XSearchTask::MergeResult(src, target, 10, ascending);
ASSERT_TRUE(status.ok());
ASSERT_EQ(target.size(), src_count);
......@@ -209,8 +208,8 @@ TEST(DBSearchTest, MERGE_TEST) {
}
{
engine::SearchContext::Id2DistanceMap src = src_result[0];
engine::SearchContext::Id2DistanceMap target = target_result[0];
scheduler::Id2DistanceMap src = src_result[0];
scheduler::Id2DistanceMap target = target_result[0];
status = engine::XSearchTask::MergeResult(src, target, 30, ascending);
ASSERT_TRUE(status.ok());
ASSERT_EQ(target.size(), src_count + target_count);
......@@ -218,8 +217,8 @@ TEST(DBSearchTest, MERGE_TEST) {
}
{
engine::SearchContext::Id2DistanceMap target = src_result[0];
engine::SearchContext::Id2DistanceMap src = target_result[0];
scheduler::Id2DistanceMap target = src_result[0];
scheduler::Id2DistanceMap src = target_result[0];
status = engine::XSearchTask::MergeResult(src, target, 30, ascending);
ASSERT_TRUE(status.ok());
ASSERT_EQ(target.size(), src_count + target_count);
......@@ -235,7 +234,7 @@ TEST(DBSearchTest, PARALLEL_CLUSTER_TEST) {
bool ascending = true;
std::vector<long> target_ids;
std::vector<float> target_distence;
engine::SearchContext::ResultSet src_result;
scheduler::ResultSet src_result;
auto DoCluster = [&](int64_t nq, int64_t topk) {
TimeRecorder rc("DoCluster");
......@@ -270,11 +269,11 @@ TEST(DBSearchTest, PARALLEL_TOPK_TEST) {
std::vector<long> target_ids;
std::vector<float> target_distence;
engine::SearchContext::ResultSet src_result;
scheduler::ResultSet src_result;
std::vector<long> insufficient_ids;
std::vector<float> insufficient_distence;
engine::SearchContext::ResultSet insufficient_result;
scheduler::ResultSet insufficient_result;
auto DoTopk = [&](int64_t nq, int64_t topk,int64_t insufficient_topk, bool ascending) {
src_result.clear();
......
cpp/unittest/db/utils.cpp
浏览文件 @ 573c9cf8
......@@ -80,7 +80,7 @@ void DBTest::SetUp() {
auto res_mgr = engine::ResMgrInst::GetInstance();
res_mgr->Clear();
res_mgr->Add(engine::ResourceFactory::Create("disk", "DISK", 0, true, false));
res_mgr->Add(engine::ResourceFactory::Create("cpu", "CPU", 0, true, true));
res_mgr->Add(engine::ResourceFactory::Create("cpu", "CPU", 0, true, false));
res_mgr->Add(engine::ResourceFactory::Create("gtx1660", "GPU", 0, true, true));
auto default_conn = engine::Connection("IO", 500.0);
......@@ -90,6 +90,8 @@ void DBTest::SetUp() {
res_mgr->Start();
engine::SchedInst::GetInstance()->Start();
engine::JobMgrInst::GetInstance()->Start();
auto options = GetOptions();
db_ = engine::DBFactory::Build(options);
}
......@@ -100,8 +102,9 @@ void DBTest::TearDown() {
BaseTest::TearDown();
engine::ResMgrInst::GetInstance()->Stop();
engine::JobMgrInst::GetInstance()->Stop();
engine::SchedInst::GetInstance()->Stop();
engine::ResMgrInst::GetInstance()->Stop();
auto options = GetOptions();
boost::filesystem::remove_all(options.meta.path);
......
cpp/unittest/scheduler/CMakeLists.txt
浏览文件 @ 573c9cf8
......@@ -30,6 +30,7 @@ aux_source_directory(${MILVUS_ENGINE_SRC}/db/scheduler/task scheduler_task_files
aux_source_directory(${MILVUS_ENGINE_SRC}/scheduler/action scheduler_action_srcs)
aux_source_directory(${MILVUS_ENGINE_SRC}/scheduler/event scheduler_event_srcs)
aux_source_directory(${MILVUS_ENGINE_SRC}/scheduler/job scheduler_job_srcs)
aux_source_directory(${MILVUS_ENGINE_SRC}/scheduler/resource scheduler_resource_srcs)
aux_source_directory(${MILVUS_ENGINE_SRC}/scheduler/task scheduler_task_srcs)
aux_source_directory(${MILVUS_ENGINE_SRC}/scheduler scheduler_srcs)
......@@ -55,6 +56,7 @@ set(scheduler_test_src
${test_srcs}
${scheduler_action_srcs}
${scheduler_event_srcs}
${scheduler_job_srcs}
${scheduler_resource_srcs}
${scheduler_task_srcs}
${scheduler_srcs}
......
cpp/unittest/scheduler/normal_test.cpp
浏览文件 @ 573c9cf8
......@@ -46,7 +46,7 @@ TEST(NormalTest, INST_TEST) {
const uint64_t NUM_TASK = 1000;
std::vector<std::shared_ptr<TestTask>> tasks;
TableFileSchemaPtr dummy = nullptr;
meta::TableFileSchemaPtr dummy = nullptr;
auto disks = res_mgr->GetDiskResources();
ASSERT_FALSE(disks.empty());
......
cpp/unittest/scheduler/resource_mgr_test.cpp
浏览文件 @ 573c9cf8
......@@ -187,7 +187,7 @@ TEST_F(ResourceMgrAdvanceTest, REGISTER_SUBSCRIBER) {
flag = true;
};
mgr1_->RegisterSubscriber(callback);
TableFileSchemaPtr dummy = nullptr;
meta::TableFileSchemaPtr dummy = nullptr;
disk_res->task_table().Put(std::make_shared<TestTask>(dummy));
sleep(1);
ASSERT_TRUE(flag);
......
cpp/unittest/scheduler/resource_test.cpp
浏览文件 @ 573c9cf8
......@@ -180,7 +180,7 @@ protected:
TEST_F(ResourceAdvanceTest, DISK_RESOURCE_TEST) {
const uint64_t NUM = 100;
std::vector<std::shared_ptr<TestTask>> tasks;
TableFileSchemaPtr dummy = nullptr;
meta::TableFileSchemaPtr dummy = nullptr;
for (uint64_t i = 0; i < NUM; ++i) {
auto task = std::make_shared<TestTask>(dummy);
tasks.push_back(task);
......@@ -205,7 +205,7 @@ TEST_F(ResourceAdvanceTest, DISK_RESOURCE_TEST) {
TEST_F(ResourceAdvanceTest, CPU_RESOURCE_TEST) {
const uint64_t NUM = 100;
std::vector<std::shared_ptr<TestTask>> tasks;
TableFileSchemaPtr dummy = nullptr;
meta::TableFileSchemaPtr dummy = nullptr;
for (uint64_t i = 0; i < NUM; ++i) {
auto task = std::make_shared<TestTask>(dummy);
tasks.push_back(task);
......@@ -230,7 +230,7 @@ TEST_F(ResourceAdvanceTest, CPU_RESOURCE_TEST) {
TEST_F(ResourceAdvanceTest, GPU_RESOURCE_TEST) {
const uint64_t NUM = 100;
std::vector<std::shared_ptr<TestTask>> tasks;
TableFileSchemaPtr dummy = nullptr;
meta::TableFileSchemaPtr dummy = nullptr;
for (uint64_t i = 0; i < NUM; ++i) {
auto task = std::make_shared<TestTask>(dummy);
tasks.push_back(task);
......@@ -255,7 +255,7 @@ TEST_F(ResourceAdvanceTest, GPU_RESOURCE_TEST) {
TEST_F(ResourceAdvanceTest, TEST_RESOURCE_TEST) {
const uint64_t NUM = 100;
std::vector<std::shared_ptr<TestTask>> tasks;
TableFileSchemaPtr dummy = nullptr;
meta::TableFileSchemaPtr dummy = nullptr;
for (uint64_t i = 0; i < NUM; ++i) {
auto task = std::make_shared<TestTask>(dummy);
tasks.push_back(task);
......
cpp/unittest/scheduler/scheduler_test.cpp
浏览文件 @ 573c9cf8
......@@ -157,7 +157,7 @@ insert_dummy_index_into_gpu_cache(uint64_t device_id) {
TEST_F(SchedulerTest, ON_LOAD_COMPLETED) {
const uint64_t NUM = 10;
std::vector<std::shared_ptr<TestTask>> tasks;
TableFileSchemaPtr dummy = std::make_shared<meta::TableFileSchema>();
meta::TableFileSchemaPtr dummy = std::make_shared<meta::TableFileSchema>();
dummy->location_ = "location";
insert_dummy_index_into_gpu_cache(1);
......@@ -177,7 +177,7 @@ TEST_F(SchedulerTest, ON_LOAD_COMPLETED) {
TEST_F(SchedulerTest, PUSH_TASK_TO_NEIGHBOUR_RANDOMLY_TEST) {
const uint64_t NUM = 10;
std::vector<std::shared_ptr<TestTask>> tasks;
TableFileSchemaPtr dummy1 = std::make_shared<meta::TableFileSchema>();
meta::TableFileSchemaPtr dummy1 = std::make_shared<meta::TableFileSchema>();
dummy1->location_ = "location";
tasks.clear();
......@@ -248,7 +248,7 @@ protected:
TEST_F(SchedulerTest2, SPECIFIED_RESOURCE_TEST) {
const uint64_t NUM = 10;
std::vector<std::shared_ptr<TestTask>> tasks;
TableFileSchemaPtr dummy = std::make_shared<meta::TableFileSchema>();
meta::TableFileSchemaPtr dummy = std::make_shared<meta::TableFileSchema>();
dummy->location_ = "location";
for (uint64_t i = 0; i < NUM; ++i) {
......
cpp/unittest/scheduler/tasktable_test.cpp
浏览文件 @ 573c9cf8
......@@ -169,7 +169,7 @@ class TaskTableBaseTest : public ::testing::Test {
protected:
void
SetUp() override {
TableFileSchemaPtr dummy = nullptr;
meta::TableFileSchemaPtr dummy = nullptr;
invalid_task_ = nullptr;
task1_ = std::make_shared<TestTask>(dummy);
task2_ = std::make_shared<TestTask>(dummy);
......@@ -339,7 +339,7 @@ class TaskTableAdvanceTest : public ::testing::Test {
protected:
void
SetUp() override {
TableFileSchemaPtr dummy = nullptr;
meta::TableFileSchemaPtr dummy = nullptr;
for (uint64_t i = 0; i < 8; ++i) {
auto task = std::make_shared<TestTask>(dummy);
table1_.Put(task);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或