Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
apache
Incubator Pegasus
提交
34455e35
Incubator Pegasus
项目概览
apache
/
Incubator Pegasus
通知
9
Star
5
Fork
0
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
DevOps
流水线
流水线任务
计划
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
Incubator Pegasus
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
DevOps
DevOps
流水线
流水线任务
计划
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
流水线任务
提交
Issue看板
前往新版Gitcode,体验更适合开发者的 AI 搜索 >>
未验证
提交
34455e35
编写于
10月 19, 2020
作者:
S
Smilencer
提交者:
GitHub
10月 19, 2020
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
feat(hotkey): implement the RPC handler in hotkey_collector (#621)
上级
1461dc55
变更
4
隐藏空白更改
内联
并排
Showing
4 changed file
with
82 addition
and
10 deletion
+82
-10
src/server/hotkey_collector.cpp
src/server/hotkey_collector.cpp
+67
-3
src/server/hotkey_collector.h
src/server/hotkey_collector.h
+8
-4
src/server/pegasus_server_impl_init.cpp
src/server/pegasus_server_impl_init.cpp
+4
-2
src/server/test/capacity_unit_calculator_test.cpp
src/server/test/capacity_unit_calculator_test.cpp
+3
-1
未找到文件。
src/server/hotkey_collector.cpp
浏览文件 @
34455e35
...
...
@@ -17,21 +17,39 @@
#include "hotkey_collector.h"
#include <dsn/dist/replication/replication_enums.h>
#include <dsn/utility/smart_pointers.h>
#include "base/pegasus_key_schema.h"
#include <dsn/dist/fmt_logging.h>
namespace
pegasus
{
namespace
server
{
hotkey_collector
::
hotkey_collector
()
:
_internal_collector
(
dsn
::
make_unique
<
hotkey_empty_data_collector
>
())
hotkey_collector
::
hotkey_collector
(
dsn
::
replication
::
hotkey_type
::
type
hotkey_type
,
dsn
::
replication
::
replica_base
*
r_base
)
:
replica_base
(
r_base
),
_state
(
hotkey_collector_state
::
STOPPED
),
_hotkey_type
(
hotkey_type
),
_internal_collector
(
std
::
make_shared
<
hotkey_empty_data_collector
>
())
{
}
// TODO: (Tangyanzhao) implement these functions
void
hotkey_collector
::
handle_rpc
(
const
dsn
::
replication
::
detect_hotkey_request
&
req
,
dsn
::
replication
::
detect_hotkey_response
&
resp
)
{
switch
(
req
.
action
)
{
case
dsn
::
replication
::
detect_action
::
START
:
on_start_detect
(
resp
);
return
;
case
dsn
::
replication
::
detect_action
::
STOP
:
on_stop_detect
(
resp
);
return
;
default:
std
::
string
hint
=
fmt
::
format
(
"{}: can't find this detect action"
,
req
.
action
);
resp
.
err
=
dsn
::
ERR_INVALID_STATE
;
resp
.
__set_err_hint
(
hint
);
derror_replica
(
hint
);
}
}
void
hotkey_collector
::
capture_raw_key
(
const
dsn
::
blob
&
raw_key
,
int64_t
weight
)
...
...
@@ -49,5 +67,51 @@ void hotkey_collector::capture_hash_key(const dsn::blob &hash_key, int64_t weigh
void
hotkey_collector
::
analyse_data
()
{
_internal_collector
->
analyse_data
();
}
void
hotkey_collector
::
on_start_detect
(
dsn
::
replication
::
detect_hotkey_response
&
resp
)
{
auto
now_state
=
_state
.
load
();
std
::
string
hint
;
switch
(
now_state
)
{
case
hotkey_collector_state
::
COARSE_DETECTING
:
case
hotkey_collector_state
::
FINE_DETECTING
:
resp
.
err
=
dsn
::
ERR_INVALID_STATE
;
hint
=
fmt
::
format
(
"still detecting {} hotkey, state is {}"
,
dsn
::
enum_to_string
(
_hotkey_type
),
enum_to_string
(
now_state
));
dwarn_replica
(
hint
);
return
;
case
hotkey_collector_state
::
FINISHED
:
resp
.
err
=
dsn
::
ERR_INVALID_STATE
;
hint
=
fmt
::
format
(
"{} hotkey result has been found, you can send a stop rpc to restart hotkey detection"
,
dsn
::
enum_to_string
(
_hotkey_type
));
dwarn_replica
(
hint
);
return
;
case
hotkey_collector_state
::
STOPPED
:
// TODO: (Tangyanzhao) start coarse detecting
_state
.
store
(
hotkey_collector_state
::
COARSE_DETECTING
);
resp
.
err
=
dsn
::
ERR_OK
;
hint
=
fmt
::
format
(
"starting to detect {} hotkey"
,
dsn
::
enum_to_string
(
_hotkey_type
));
ddebug_replica
(
hint
);
return
;
default:
hint
=
"invalid collector state"
;
resp
.
err
=
dsn
::
ERR_INVALID_STATE
;
resp
.
__set_err_hint
(
hint
);
derror_replica
(
hint
);
dassert
(
false
,
"invalid collector state"
);
}
}
void
hotkey_collector
::
on_stop_detect
(
dsn
::
replication
::
detect_hotkey_response
&
resp
)
{
_state
.
store
(
hotkey_collector_state
::
STOPPED
);
_internal_collector
.
reset
();
resp
.
err
=
dsn
::
ERR_OK
;
std
::
string
hint
=
fmt
::
format
(
"{} hotkey stopped, cache cleared"
,
dsn
::
enum_to_string
(
_hotkey_type
));
ddebug_replica
(
hint
);
}
}
// namespace server
}
// namespace pegasus
src/server/hotkey_collector.h
浏览文件 @
34455e35
...
...
@@ -17,9 +17,9 @@
#pragma once
#include <dsn/utility/string_view.h>
#include <dsn/dist/replication/replication_types.h>
#include "hotkey_collector_state.h"
#include <dsn/dist/replication/replica_base.h>
namespace
pegasus
{
namespace
server
{
...
...
@@ -63,10 +63,11 @@ class internal_collector_base;
// | | | Hotkey |
// +--------------------+ +----------------------------------------------------+
class
hotkey_collector
class
hotkey_collector
:
public
dsn
::
replication
::
replica_base
{
public:
hotkey_collector
();
hotkey_collector
(
dsn
::
replication
::
hotkey_type
::
type
hotkey_type
,
dsn
::
replication
::
replica_base
*
r_base
);
// TODO: (Tangyanzhao) capture_*_key should be consistent with hotspot detection
// weight: calculate the weight according to the specific situation
void
capture_raw_key
(
const
dsn
::
blob
&
raw_key
,
int64_t
weight
);
...
...
@@ -76,8 +77,11 @@ public:
/*out*/
dsn
::
replication
::
detect_hotkey_response
&
resp
);
private:
std
::
unique_ptr
<
internal_collector_base
>
_internal_collector
;
void
on_start_detect
(
dsn
::
replication
::
detect_hotkey_response
&
resp
);
void
on_stop_detect
(
dsn
::
replication
::
detect_hotkey_response
&
resp
);
std
::
atomic
<
hotkey_collector_state
>
_state
;
const
dsn
::
replication
::
hotkey_type
::
type
_hotkey_type
;
std
::
shared_ptr
<
internal_collector_base
>
_internal_collector
;
};
class
internal_collector_base
...
...
src/server/pegasus_server_impl_init.cpp
浏览文件 @
34455e35
...
...
@@ -43,8 +43,10 @@ pegasus_server_impl::pegasus_server_impl(dsn::replication::replica *r)
_primary_address
=
dsn
::
rpc_address
(
dsn_primary_address
()).
to_string
();
_gpid
=
get_gpid
();
_read_hotkey_collector
=
std
::
make_shared
<
hotkey_collector
>
();
_write_hotkey_collector
=
std
::
make_shared
<
hotkey_collector
>
();
_read_hotkey_collector
=
std
::
make_shared
<
hotkey_collector
>
(
dsn
::
replication
::
hotkey_type
::
READ
,
this
);
_write_hotkey_collector
=
std
::
make_shared
<
hotkey_collector
>
(
dsn
::
replication
::
hotkey_type
::
WRITE
,
this
);
_verbose_log
=
dsn_config_get_value_bool
(
"pegasus.server"
,
"rocksdb_verbose_log"
,
...
...
src/server/test/capacity_unit_calculator_test.cpp
浏览文件 @
34455e35
...
...
@@ -29,7 +29,9 @@ public:
explicit
mock_capacity_unit_calculator
(
dsn
::
replication
::
replica_base
*
r
)
:
capacity_unit_calculator
(
r
,
std
::
make_shared
<
hotkey_collector
>
(),
std
::
make_shared
<
hotkey_collector
>
())
r
,
std
::
make_shared
<
hotkey_collector
>
(
dsn
::
replication
::
hotkey_type
::
READ
,
this
),
std
::
make_shared
<
hotkey_collector
>
(
dsn
::
replication
::
hotkey_type
::
WRITE
,
this
))
{
}
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录