Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
milvus
milvus
提交
46ed8749
M
milvus
项目概览
milvus
/
milvus
大约 1 年 前同步成功
通知
261
Star
22476
Fork
2472
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
M
milvus
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
46ed8749
编写于
6月 30, 2019
作者:
S
starlord
提交者:
jinhai
7月 13, 2019
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
add more unitest
Former-commit-id: 7c7d2337f0d0c7abd82533fc2ff2ded0ee07344b
上级
269f727f
变更
4
隐藏空白更改
内联
并排
Showing
4 changed file
with
138 addition
and
24 deletion
+138
-24
cpp/src/db/scheduler/TaskDispatchQueue.cpp
cpp/src/db/scheduler/TaskDispatchQueue.cpp
+0
-14
cpp/src/db/scheduler/TaskDispatchStrategy.cpp
cpp/src/db/scheduler/TaskDispatchStrategy.cpp
+12
-6
cpp/unittest/db/CMakeLists.txt
cpp/unittest/db/CMakeLists.txt
+2
-4
cpp/unittest/db/scheduler_test.cpp
cpp/unittest/db/scheduler_test.cpp
+124
-0
未找到文件。
cpp/src/db/scheduler/TaskDispatchQueue.cpp
浏览文件 @
46ed8749
...
...
@@ -24,14 +24,6 @@ TaskDispatchQueue::Put(const ScheduleContextPtr &context) {
return
;
}
if
(
queue_
.
size
()
>=
capacity_
)
{
std
::
string
error_msg
=
"blocking queue is full, capacity: "
+
std
::
to_string
(
capacity_
)
+
" queue_size: "
+
std
::
to_string
(
queue_
.
size
());
SERVER_LOG_ERROR
<<
error_msg
;
throw
server
::
ServerException
(
server
::
SERVER_BLOCKING_QUEUE_EMPTY
,
error_msg
);
}
TaskDispatchStrategy
::
Schedule
(
context
,
queue_
);
empty_
.
notify_all
();
...
...
@@ -42,12 +34,6 @@ TaskDispatchQueue::Take() {
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
::
SERVER_BLOCKING_QUEUE_EMPTY
,
error_msg
);
}
ScheduleTaskPtr
front
(
queue_
.
front
());
queue_
.
pop_front
();
full_
.
notify_all
();
...
...
cpp/src/db/scheduler/TaskDispatchStrategy.cpp
浏览文件 @
46ed8749
...
...
@@ -74,20 +74,26 @@ public:
}
std
::
string
table_id
=
context
->
table_id
();
for
(
auto
iter
=
task_list
.
begin
();
iter
!=
task_list
.
end
();
++
iter
)
{
//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
;
}
//put delete task to proper position
IndexLoadTaskPtr
loader
=
std
::
static_pointer_cast
<
IndexLoadTask
>
(
*
iter
);
if
(
loader
->
file_
->
table_id_
==
table_id
)
{
task_list
.
insert
(
++
iter
,
delete_task
);
break
;
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
;
}
};
...
...
cpp/unittest/db/CMakeLists.txt
浏览文件 @
46ed8749
...
...
@@ -7,6 +7,7 @@ aux_source_directory(${MILVUS_ENGINE_SRC}/db db_srcs)
aux_source_directory
(
${
MILVUS_ENGINE_SRC
}
/config config_files
)
aux_source_directory
(
${
MILVUS_ENGINE_SRC
}
/cache cache_srcs
)
aux_source_directory
(
${
MILVUS_ENGINE_SRC
}
/wrapper wrapper_src
)
aux_source_directory
(
./ test_srcs
)
aux_source_directory
(
${
MILVUS_ENGINE_SRC
}
/db/scheduler scheduler_files
)
aux_source_directory
(
${
MILVUS_ENGINE_SRC
}
/db/scheduler/context scheduler_context_files
)
...
...
@@ -30,10 +31,7 @@ set(db_test_src
${
db_scheduler_srcs
}
${
wrapper_src
}
${
require_files
}
MySQLMetaImpl_test.cpp
utils.cpp
db_tests.cpp
meta_tests.cpp
)
${
test_srcs
}
)
cuda_add_executable
(
db_test
${
db_test_src
}
)
...
...
cpp/unittest/db/scheduler_test.cpp
0 → 100644
浏览文件 @
46ed8749
////////////////////////////////////////////////////////////////////////////////
// Copyright 上海赜睿信息科技有限公司(Zilliz) - All Rights Reserved
// Unauthorized copying of this file, via any medium is strictly prohibited.
// Proprietary and confidential.
////////////////////////////////////////////////////////////////////////////////
#include <gtest/gtest.h>
#include <thread>
#include <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
,
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
);
}
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
,
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
::
Meta
::
Ptr
meta_ptr
;
engine
::
DeleteContextPtr
context_ptr
=
std
::
make_shared
<
engine
::
DeleteContext
>
(
table_id
,
meta_ptr
);
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
);
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
);
}
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录