Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
milvus
milvus
提交
b6e51b26
M
milvus
项目概览
milvus
/
milvus
12 个月 前同步成功
通知
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,发现更多精彩内容 >>
提交
b6e51b26
编写于
6月 13, 2019
作者:
J
jinhai
浏览文件
操作
浏览文件
下载
差异文件
Merge remote-tracking branch 'upstream/branch-0.3.0' into branch-0.3.0
Former-commit-id: f431335445d2c54f939dfd84a8dfdf0e4d2df4a2
上级
94612238
49c52830
变更
7
显示空白变更内容
内联
并排
Showing
7 changed file
with
36 addition
and
55 deletion
+36
-55
cpp/CHANGELOG.md
cpp/CHANGELOG.md
+1
-0
cpp/src/db/scheduler/IndexLoaderQueue.cpp
cpp/src/db/scheduler/IndexLoaderQueue.cpp
+1
-6
cpp/src/db/scheduler/IndexLoaderQueue.h
cpp/src/db/scheduler/IndexLoaderQueue.h
+1
-4
cpp/src/db/scheduler/SearchScheduler.cpp
cpp/src/db/scheduler/SearchScheduler.cpp
+24
-16
cpp/src/db/scheduler/SearchScheduler.h
cpp/src/db/scheduler/SearchScheduler.h
+8
-2
cpp/src/db/scheduler/SearchTaskQueue.cpp
cpp/src/db/scheduler/SearchTaskQueue.cpp
+0
-12
cpp/src/db/scheduler/SearchTaskQueue.h
cpp/src/db/scheduler/SearchTaskQueue.h
+1
-15
未找到文件。
cpp/CHANGELOG.md
浏览文件 @
b6e51b26
...
...
@@ -5,6 +5,7 @@ Please mark all change in change log and use the ticket from JIRA.
# MegaSearch 0.3.0 (TBD)
## Bug
-
MS-80 - Fix server hang issue
## Improvement
-
MS-82 - Update server startup welcome message
...
...
cpp/src/db/scheduler/IndexLoaderQueue.cpp
浏览文件 @
b6e51b26
...
...
@@ -13,12 +13,6 @@ namespace zilliz {
namespace
milvus
{
namespace
engine
{
IndexLoaderQueue
&
IndexLoaderQueue
::
GetInstance
()
{
static
IndexLoaderQueue
instance
;
return
instance
;
}
void
IndexLoaderQueue
::
Put
(
const
SearchContextPtr
&
search_context
)
{
std
::
unique_lock
<
std
::
mutex
>
lock
(
mtx
);
...
...
@@ -26,6 +20,7 @@ IndexLoaderQueue::Put(const SearchContextPtr &search_context) {
if
(
search_context
==
nullptr
)
{
queue_
.
push_back
(
nullptr
);
empty_
.
notify_all
();
return
;
}
...
...
cpp/src/db/scheduler/IndexLoaderQueue.h
浏览文件 @
b6e51b26
...
...
@@ -26,18 +26,15 @@ public:
using
IndexLoaderContextPtr
=
std
::
shared_ptr
<
IndexLoaderContext
>
;
class
IndexLoaderQueue
{
p
rivate
:
p
ublic
:
IndexLoaderQueue
()
:
mtx
(),
full_
(),
empty_
()
{}
IndexLoaderQueue
(
const
IndexLoaderQueue
&
rhs
)
=
delete
;
IndexLoaderQueue
&
operator
=
(
const
IndexLoaderQueue
&
rhs
)
=
delete
;
public:
using
LoaderQueue
=
std
::
list
<
IndexLoaderContextPtr
>
;
static
IndexLoaderQueue
&
GetInstance
();
void
Put
(
const
SearchContextPtr
&
search_context
);
IndexLoaderContextPtr
Take
();
...
...
cpp/src/db/scheduler/SearchScheduler.cpp
浏览文件 @
b6e51b26
...
...
@@ -55,8 +55,7 @@ void CollectDurationMetrics(int index_type, double total_time) {
}
SearchScheduler
::
SearchScheduler
()
:
thread_pool_
(
2
),
stopped_
(
true
)
{
:
stopped_
(
true
)
{
Start
();
}
...
...
@@ -75,8 +74,13 @@ SearchScheduler::Start() {
return
true
;
}
thread_pool_
.
enqueue
(
&
SearchScheduler
::
IndexLoadWorker
,
this
);
thread_pool_
.
enqueue
(
&
SearchScheduler
::
SearchWorker
,
this
);
stopped_
=
false
;
search_queue_
.
SetCapacity
(
2
);
index_load_thread_
=
std
::
make_shared
<
std
::
thread
>
(
&
SearchScheduler
::
IndexLoadWorker
,
this
);
search_thread_
=
std
::
make_shared
<
std
::
thread
>
(
&
SearchScheduler
::
SearchWorker
,
this
);
return
true
;
}
...
...
@@ -86,29 +90,34 @@ SearchScheduler::Stop() {
return
true
;
}
IndexLoaderQueue
&
index_queue
=
IndexLoaderQueue
::
GetInstance
();
index_queue
.
Put
(
nullptr
);
if
(
index_load_thread_
)
{
index_load_queue_
.
Put
(
nullptr
);
index_load_thread_
->
join
();
index_load_thread_
=
nullptr
;
}
if
(
search_thread_
)
{
search_queue_
.
Put
(
nullptr
);
search_thread_
->
join
();
search_thread_
=
nullptr
;
}
SearchTaskQueue
&
search_queue
=
SearchTaskQueue
::
GetInstance
();
search_queue
.
Put
(
nullptr
);
stopped_
=
true
;
return
true
;
}
bool
SearchScheduler
::
ScheduleSearchTask
(
SearchContextPtr
&
search_context
)
{
IndexLoaderQueue
&
index_queue
=
IndexLoaderQueue
::
GetInstance
();
index_queue
.
Put
(
search_context
);
index_load_queue_
.
Put
(
search_context
);
return
true
;
}
bool
SearchScheduler
::
IndexLoadWorker
()
{
IndexLoaderQueue
&
index_queue
=
IndexLoaderQueue
::
GetInstance
();
SearchTaskQueue
&
search_queue
=
SearchTaskQueue
::
GetInstance
();
while
(
true
)
{
IndexLoaderContextPtr
context
=
index_
queue
.
Take
();
IndexLoaderContextPtr
context
=
index_
load_queue_
.
Take
();
if
(
context
==
nullptr
)
{
SERVER_LOG_INFO
<<
"Stop thread for index loading"
;
break
;
//exit
...
...
@@ -137,7 +146,7 @@ SearchScheduler::IndexLoadWorker() {
task_ptr
->
index_type_
=
context
->
file_
->
file_type_
;
task_ptr
->
index_engine_
=
index_ptr
;
task_ptr
->
search_contexts_
.
swap
(
context
->
search_contexts_
);
search_queue
.
Put
(
task_ptr
);
search_queue
_
.
Put
(
task_ptr
);
}
return
true
;
...
...
@@ -145,9 +154,8 @@ SearchScheduler::IndexLoadWorker() {
bool
SearchScheduler
::
SearchWorker
()
{
SearchTaskQueue
&
search_queue
=
SearchTaskQueue
::
GetInstance
();
while
(
true
)
{
SearchTaskPtr
task_ptr
=
search_queue
.
Take
();
SearchTaskPtr
task_ptr
=
search_queue
_
.
Take
();
if
(
task_ptr
==
nullptr
)
{
SERVER_LOG_INFO
<<
"Stop thread for searching"
;
break
;
//exit
...
...
cpp/src/db/scheduler/SearchScheduler.h
浏览文件 @
b6e51b26
...
...
@@ -6,7 +6,8 @@
#pragma once
#include "SearchContext.h"
#include "utils/ThreadPool.h"
#include "IndexLoaderQueue.h"
#include "SearchTaskQueue.h"
namespace
zilliz
{
namespace
milvus
{
...
...
@@ -30,7 +31,12 @@ private:
bool
SearchWorker
();
private:
server
::
ThreadPool
thread_pool_
;
std
::
shared_ptr
<
std
::
thread
>
index_load_thread_
;
std
::
shared_ptr
<
std
::
thread
>
search_thread_
;
IndexLoaderQueue
index_load_queue_
;
SearchTaskQueue
search_queue_
;
bool
stopped_
=
true
;
};
...
...
cpp/src/db/scheduler/SearchTaskQueue.cpp
浏览文件 @
b6e51b26
...
...
@@ -94,18 +94,6 @@ void CalcScore(uint64_t vector_count,
}
SearchTaskQueue
::
SearchTaskQueue
()
{
SetCapacity
(
4
);
}
SearchTaskQueue
&
SearchTaskQueue
::
GetInstance
()
{
static
SearchTaskQueue
instance
;
return
instance
;
}
bool
SearchTask
::
DoSearch
()
{
if
(
index_engine_
==
nullptr
)
{
return
false
;
...
...
cpp/src/db/scheduler/SearchTaskQueue.h
浏览文件 @
b6e51b26
...
...
@@ -27,21 +27,7 @@ public:
};
using
SearchTaskPtr
=
std
::
shared_ptr
<
SearchTask
>
;
class
SearchTaskQueue
:
public
server
::
BlockingQueue
<
SearchTaskPtr
>
{
private:
SearchTaskQueue
();
SearchTaskQueue
(
const
SearchTaskQueue
&
rhs
)
=
delete
;
SearchTaskQueue
&
operator
=
(
const
SearchTaskQueue
&
rhs
)
=
delete
;
public:
static
SearchTaskQueue
&
GetInstance
();
private:
};
using
SearchTaskQueue
=
server
::
BlockingQueue
<
SearchTaskPtr
>
;
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录