Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
BaiXuePrincess
milvus
提交
a724141e
milvus
项目概览
BaiXuePrincess
/
milvus
与 Fork 源项目一致
从无法访问的项目Fork
通知
7
Star
4
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
milvus
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
a724141e
编写于
4月 30, 2019
作者:
G
groot
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
use multithreads to do id mapping
Former-commit-id: 10b201d3801b146fa6335e4dfd20c877cfb2e948
上级
b67c731b
变更
3
显示空白变更内容
内联
并排
Showing
3 changed file
with
48 addition
and
12 deletion
+48
-12
cpp/src/server/VecServiceTask.cpp
cpp/src/server/VecServiceTask.cpp
+44
-10
cpp/src/server/VecServiceTask.h
cpp/src/server/VecServiceTask.h
+2
-0
cpp/src/utils/ThreadPool.h
cpp/src/utils/ThreadPool.h
+2
-2
未找到文件。
cpp/src/server/VecServiceTask.cpp
浏览文件 @
a724141e
...
...
@@ -9,10 +9,12 @@
#include "utils/CommonUtil.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ThreadPool.h"
#include "db/DB.h"
#include "db/Env.h"
#include "db/Meta.h"
namespace
zilliz
{
namespace
vecwise
{
namespace
server
{
...
...
@@ -21,6 +23,7 @@ static const std::string DQL_TASK_GROUP = "dql";
static
const
std
::
string
DDL_DML_TASK_GROUP
=
"ddl_dml"
;
static
const
std
::
string
VECTOR_UID
=
"uid"
;
static
const
uint64_t
USE_MT
=
5000
;
using
DB_META
=
zilliz
::
vecwise
::
engine
::
meta
::
Meta
;
using
DB_DATE
=
zilliz
::
vecwise
::
engine
::
meta
::
DateT
;
...
...
@@ -61,6 +64,11 @@ namespace {
CommonUtil
::
ConvertTime
(
dt
.
year
,
dt
.
month
,
dt
.
day
,
dt
.
hour
,
dt
.
minute
,
dt
.
second
,
t_t
);
return
DB_META
::
GetDate
(
t_t
);
}
ThreadPool
&
GetThreadPool
()
{
static
ThreadPool
pool
(
6
);
return
pool
;
}
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
...
...
@@ -371,6 +379,19 @@ const AttribMap& AddBatchVectorTask::GetVecAttrib(uint64_t index) const {
}
}
void
AddBatchVectorTask
::
ProcessIdMapping
(
engine
::
IDNumbers
&
vector_ids
,
uint64_t
from
,
uint64_t
to
)
{
std
::
string
nid_prefix
=
group_id_
+
"_"
;
for
(
size_t
i
=
from
;
i
<
to
;
i
++
)
{
std
::
string
uid
=
GetVecID
(
i
);
std
::
string
nid
=
nid_prefix
+
std
::
to_string
(
vector_ids
[
i
]);
AttribMap
attrib
=
GetVecAttrib
(
i
);
attrib
[
VECTOR_UID
]
=
uid
;
std
::
string
attrib_str
;
AttributeSerializer
::
Encode
(
attrib
,
attrib_str
);
IVecIdMapper
::
GetInstance
()
->
Put
(
nid
,
attrib_str
);
}
}
ServerError
AddBatchVectorTask
::
OnExecute
()
{
try
{
TimeRecorder
rc
(
"AddBatchVectorTask"
);
...
...
@@ -415,17 +436,30 @@ ServerError AddBatchVectorTask::OnExecute() {
SERVER_LOG_ERROR
<<
"Vector ID not returned"
;
return
SERVER_UNEXPECTED_ERROR
;
}
else
{
std
::
string
nid_prefix
=
group_id_
+
"_"
;
for
(
size_t
i
=
0
;
i
<
vec_count
;
i
++
)
{
std
::
string
uid
=
GetVecID
(
i
);
std
::
string
nid
=
nid_prefix
+
std
::
to_string
(
vector_ids
[
i
]);
AttribMap
attrib
=
GetVecAttrib
(
i
);
attrib
[
VECTOR_UID
]
=
uid
;
std
::
string
attrib_str
;
AttributeSerializer
::
Encode
(
attrib
,
attrib_str
);
IVecIdMapper
::
GetInstance
()
->
Put
(
nid
,
attrib_str
);
if
(
vec_count
<
USE_MT
)
{
ProcessIdMapping
(
vector_ids
,
0
,
vec_count
);
rc
.
Record
(
"built id mapping"
);
}
else
{
std
::
list
<
std
::
future
<
void
>>
threads_list
;
uint64_t
begin_index
=
0
,
end_index
=
USE_MT
;
while
(
end_index
<
vec_count
)
{
threads_list
.
push_back
(
GetThreadPool
().
enqueue
(
&
AddBatchVectorTask
::
ProcessIdMapping
,
this
,
vector_ids
,
begin_index
,
end_index
));
begin_index
=
end_index
;
end_index
+=
USE_MT
;
if
(
end_index
>
vec_count
)
{
end_index
=
vec_count
;
}
}
for
(
std
::
list
<
std
::
future
<
void
>>::
iterator
it
=
threads_list
.
begin
();
it
!=
threads_list
.
end
();
it
++
)
{
it
->
wait
();
}
rc
.
Record
(
"built id mapping by multi-threads:"
+
std
::
to_string
(
threads_list
.
size
()));
}
rc
.
Record
(
"build id mapping"
);
}
}
...
...
cpp/src/server/VecServiceTask.h
浏览文件 @
a724141e
...
...
@@ -119,6 +119,8 @@ protected:
std
::
string
GetVecID
(
uint64_t
index
)
const
;
const
AttribMap
&
GetVecAttrib
(
uint64_t
index
)
const
;
void
ProcessIdMapping
(
engine
::
IDNumbers
&
vector_ids
,
uint64_t
from
,
uint64_t
to
);
ServerError
OnExecute
()
override
;
private:
...
...
cpp/src/utils/ThreadPool.h
浏览文件 @
a724141e
...
...
@@ -19,8 +19,8 @@
#define MAX_THREADS_NUM 32
namespace
zilliz
{
namespace
sql
{
namespace
s
torage
{
namespace
vecwise
{
namespace
s
erver
{
class
ThreadPool
{
public:
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录