You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "chenlinzhong (via GitHub)" <gi...@apache.org> on 2023/02/11 15:26:59 UTC
[GitHub] [doris] chenlinzhong opened a new pull request, #16639: [Improvement](brpc) replace brpc bthread with pthread
chenlinzhong opened a new pull request, #16639:
URL: https://github.com/apache/doris/pull/16639
# Proposed changes
Issue Number: close #16634
## Problem summary
mainly include:
1. brpc service adds two types of thread pools. The number of "light" and "heavy" thread pools is different
2. Classify the interfaces of be. Those related to data transmission are classified as heavy interfaces and others as light interfaces
3. Add some monitoring to the thread pool, including the queue size and the number of active threads. Use these indicators to guide the configuration of the number of threads
主要包括
- 1.brpc服务增加2类线程池，"轻","重" 不同线程池的数量配置不一样
- 2.对be的接口进行分类，跟数据传输有关的归类为重接口，其他为轻接口
- 3.增加一些监控，包括be每个接口qps和线程池队列大小、活跃线程个数，根据这些指标来指导线程数的配置
<img width="1359" alt="image" src="https://user-images.githubusercontent.com/11487604/218252833-b0648a66-c5e3-4cbf-9930-3de5a86cfd08.png">
## Checklist(Required)
1. Does it affect the original behavior:
- [x] Yes
- [ ] No
- [ ] I don't know
2. Has unit tests been added:
- [ ] Yes
- [ ] No
- [x] No Need
3. Has document been added or modified:
- [ ] Yes
- [x] No
- [ ] No Need
4. Does it need to update dependencies:
- [ ] Yes
- [x] No
5. Are there any changes that cannot be rolled back:
- [ ] Yes (If Yes, please explain WHY)
- [x] No
## Further comments
If this is a relatively large or complex change, kick off the discussion at [dev@doris.apache.org](mailto:dev@doris.apache.org) by explaining why you chose the solution you did and what alternatives you considered, etc...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1437744241
run feut
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on a diff in pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1104572834
##########
be/src/util/doris_metrics.cpp:
##########
@@ -197,6 +230,37 @@ DorisMetrics::DorisMetrics() : _metric_registry(_s_registry_name) {
_server_metric_entity = _metric_registry.register_entity("server");
INT_COUNTER_METRIC_REGISTER(_server_metric_entity, fragment_requests_total);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, transmit_data);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, transmit_data_by_http);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, exec_plan_fragment);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, exec_plan_fragment_prepare);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, exec_plan_fragment_start);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, cancel_plan_fragment);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, fetch_data);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, fetch_table_schema);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, tablet_writer_open);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, tablet_writer_add_block);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, tablet_writer_add_block_by_http);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, tablet_writer_cancel);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, get_info);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, update_cache);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, fetch_cache);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, clear_cache);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, merge_filter);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, apply_filter);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, transmit_block);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, transmit_block_by_http);
Review Comment:
Provide two metrics interfaces for the same node
- http://ip:http_port/metrics
- http://ip:brpc_port/brpc_metrics
It is not very reasonable from the perspective of use. I think it is reasonable and acceptable to have redundant data here, and it does not affect performance
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] eldenmoon commented on a diff in pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "eldenmoon (via GitHub)" <gi...@apache.org>.
eldenmoon commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103940279
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
please Do not put this RPC interface into thread pool, it's very light weight and can't be mixed with other rpc interface, since it's latency sensitive
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on a diff in pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103954749
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
- _exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
- }
- Status::OK().to_protobuf(response->mutable_status());
+ Status::OK().to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController* controller,
const PUpdateCacheRequest* request,
PCacheResponse* response, google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ DorisMetrics::instance()->update_cache->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->update(request, response);
+ });
}
void PInternalServiceImpl::fetch_cache(google::protobuf::RpcController* controller,
const PFetchCacheRequest* request, PFetchCacheResult* result,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->fetch(request, result);
+ DorisMetrics::instance()->fetch_cache->increment(1);
+ _heavy_work_pool.offer([this, request, result, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->fetch(request, result);
+ });
}
Review Comment:
if fails, just means rpc service is shoutdown otherwise will block until the request handled
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1437927675
run p0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1430880388
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1437936478
run p0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on a diff in pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103646780
##########
be/src/service/internal_service.cpp:
##########
@@ -247,15 +292,18 @@ void PInternalServiceImpl::tablet_writer_cancel(google::protobuf::RpcController*
const PTabletWriterCancelRequest* request,
PTabletWriterCancelResult* response,
google::protobuf::Closure* done) {
- VLOG_RPC << "tablet writer cancel, id=" << request->id() << ", index_id=" << request->index_id()
- << ", sender_id=" << request->sender_id();
- brpc::ClosureGuard closure_guard(done);
- auto st = _exec_env->load_channel_mgr()->cancel(*request);
- if (!st.ok()) {
- LOG(WARNING) << "tablet writer cancel failed, id=" << request->id()
- << ", index_id=" << request->index_id()
- << ", sender_id=" << request->sender_id();
- }
+ DorisMetrics::instance()->tablet_writer_cancel->increment(1);
+ _light_work_pool.offer([this, controller, request, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, done]() {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] eldenmoon commented on a diff in pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "eldenmoon (via GitHub)" <gi...@apache.org>.
eldenmoon commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103964516
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
Other heavy tasks could block all pthread and no worker thread for this light weight rpc and infect latency
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] dataroaring commented on a diff in pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "dataroaring (via GitHub)" <gi...@apache.org>.
dataroaring commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1104568288
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
- _exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
- }
- Status::OK().to_protobuf(response->mutable_status());
+ Status::OK().to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController* controller,
const PUpdateCacheRequest* request,
PCacheResponse* response, google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ DorisMetrics::instance()->update_cache->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->update(request, response);
+ });
}
void PInternalServiceImpl::fetch_cache(google::protobuf::RpcController* controller,
const PFetchCacheRequest* request, PFetchCacheResult* result,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->fetch(request, result);
+ DorisMetrics::instance()->fetch_cache->increment(1);
+ _heavy_work_pool.offer([this, request, result, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->fetch(request, result);
+ });
}
Review Comment:
Do you test this case like mock a offer failure? In the case, we do not set status, so i do not know how brpc handle it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1428063041
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1429021856
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1439447562
run p0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1437744320
run beut
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on a diff in pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103977135
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
Even if I don't add this pr, will have the same problem
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1439294961
run p0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1439316628
run p0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1439335238
run p0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong merged pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong merged PR #16639:
URL: https://github.com/apache/doris/pull/16639
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1434604807
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1436540634
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on a diff in pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103646679
##########
be/src/service/internal_service.cpp:
##########
@@ -247,15 +292,18 @@
const PTabletWriterCancelRequest* request,
PTabletWriterCancelResult* response,
google::protobuf::Closure* done) {
- VLOG_RPC << "tablet writer cancel, id=" << request->id() << ", index_id=" << request->index_id()
- << ", sender_id=" << request->sender_id();
- brpc::ClosureGuard closure_guard(done);
- auto st = _exec_env->load_channel_mgr()->cancel(*request);
- if (!st.ok()) {
- LOG(WARNING) << "tablet writer cancel failed, id=" << request->id()
- << ", index_id=" << request->index_id()
- << ", sender_id=" << request->sender_id();
- }
+ DorisMetrics::instance()->tablet_writer_cancel->increment(1);
+ _light_work_pool.offer([this, controller, request, done]() {
Review Comment:
warning: lambda capture 'response' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, controller, request, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -247,15 +292,18 @@ void PInternalServiceImpl::tablet_writer_cancel(google::protobuf::RpcController*
const PTabletWriterCancelRequest* request,
PTabletWriterCancelResult* response,
google::protobuf::Closure* done) {
- VLOG_RPC << "tablet writer cancel, id=" << request->id() << ", index_id=" << request->index_id()
- << ", sender_id=" << request->sender_id();
- brpc::ClosureGuard closure_guard(done);
- auto st = _exec_env->load_channel_mgr()->cancel(*request);
- if (!st.ok()) {
- LOG(WARNING) << "tablet writer cancel failed, id=" << request->id()
- << ", index_id=" << request->index_id()
- << ", sender_id=" << request->sender_id();
- }
+ DorisMetrics::instance()->tablet_writer_cancel->increment(1);
+ _light_work_pool.offer([this, controller, request, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -286,125 +334,134 @@
}
}
-void PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController* controller,
const PCancelPlanFragmentRequest* request,
PCancelPlanFragmentResult* result,
google::protobuf::Closure* done) {
- auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start", cntl_base);
- auto scope = OpentelemetryScope {span};
- brpc::ClosureGuard closure_guard(done);
- TUniqueId tid;
- tid.__set_hi(request->finst_id().hi());
- tid.__set_lo(request->finst_id().lo());
-
- Status st = Status::OK();
- if (request->has_cancel_reason()) {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
- << ", reason: " << request->cancel_reason();
- _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
- } else {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
- _exec_env->fragment_mgr()->cancel(tid);
- }
-
- // TODO: the logic seems useless, cancel only return Status::OK. remove it
- st.to_protobuf(result->mutable_status());
+ DorisMetrics::instance()->cancel_plan_fragment->increment(1);
+ _light_work_pool.offer([this, controller, request, result, done]() {
+ auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start", controller);
+ auto scope = OpentelemetryScope {span};
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId tid;
+ tid.__set_hi(request->finst_id().hi());
+ tid.__set_lo(request->finst_id().lo());
+
+ Status st = Status::OK();
+ if (request->has_cancel_reason()) {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
+ << ", reason: " << request->cancel_reason();
+ _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
+ } else {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
+ _exec_env->fragment_mgr()->cancel(tid);
+ }
+ // TODO: the logic seems useless, cancel only return Status::OK. remove it
+ st.to_protobuf(result->mutable_status());
+ });
}
-void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* controller,
const PFetchDataRequest* request, PFetchDataResult* result,
google::protobuf::Closure* done) {
- brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
- GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
- _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ DorisMetrics::instance()->fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, result, done]() {
+ brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
+ _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ });
}
void PInternalServiceImpl::fetch_table_schema(google::protobuf::RpcController* controller,
const PFetchTableSchemaRequest* request,
PFetchTableSchemaResult* result,
google::protobuf::Closure* done) {
- VLOG_RPC << "fetch table schema";
- brpc::ClosureGuard closure_guard(done);
- TFileScanRange file_scan_range;
- Status st = Status::OK();
- {
- const uint8_t* buf = (const uint8_t*)(request->file_scan_range().data());
- uint32_t len = request->file_scan_range().size();
- st = deserialize_thrift_msg(buf, &len, false, &file_scan_range);
+ DorisMetrics::instance()->fetch_table_schema->increment(1);
+ _light_work_pool.offer([this, request, result, done]() {
Review Comment:
warning: lambda capture 'this' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([ controller, request, result, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -286,125 +334,134 @@
}
}
-void PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController* controller,
const PCancelPlanFragmentRequest* request,
PCancelPlanFragmentResult* result,
google::protobuf::Closure* done) {
- auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start", cntl_base);
- auto scope = OpentelemetryScope {span};
- brpc::ClosureGuard closure_guard(done);
- TUniqueId tid;
- tid.__set_hi(request->finst_id().hi());
- tid.__set_lo(request->finst_id().lo());
-
- Status st = Status::OK();
- if (request->has_cancel_reason()) {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
- << ", reason: " << request->cancel_reason();
- _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
- } else {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
- _exec_env->fragment_mgr()->cancel(tid);
- }
-
- // TODO: the logic seems useless, cancel only return Status::OK. remove it
- st.to_protobuf(result->mutable_status());
+ DorisMetrics::instance()->cancel_plan_fragment->increment(1);
+ _light_work_pool.offer([this, controller, request, result, done]() {
+ auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start", controller);
+ auto scope = OpentelemetryScope {span};
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId tid;
+ tid.__set_hi(request->finst_id().hi());
+ tid.__set_lo(request->finst_id().lo());
+
+ Status st = Status::OK();
+ if (request->has_cancel_reason()) {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
+ << ", reason: " << request->cancel_reason();
+ _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
+ } else {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
+ _exec_env->fragment_mgr()->cancel(tid);
+ }
+ // TODO: the logic seems useless, cancel only return Status::OK. remove it
+ st.to_protobuf(result->mutable_status());
+ });
}
-void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* controller,
const PFetchDataRequest* request, PFetchDataResult* result,
google::protobuf::Closure* done) {
- brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
- GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
- _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ DorisMetrics::instance()->fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, result, done]() {
+ brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
+ _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ });
}
void PInternalServiceImpl::fetch_table_schema(google::protobuf::RpcController* controller,
const PFetchTableSchemaRequest* request,
PFetchTableSchemaResult* result,
google::protobuf::Closure* done) {
- VLOG_RPC << "fetch table schema";
- brpc::ClosureGuard closure_guard(done);
- TFileScanRange file_scan_range;
- Status st = Status::OK();
- {
- const uint8_t* buf = (const uint8_t*)(request->file_scan_range().data());
- uint32_t len = request->file_scan_range().size();
- st = deserialize_thrift_msg(buf, &len, false, &file_scan_range);
+ DorisMetrics::instance()->fetch_table_schema->increment(1);
+ _light_work_pool.offer([this, request, result, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, result, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -693,75 +790,86 @@
const PCheckRPCChannelRequest* request,
PCheckRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->data().size() != request->size()) {
- std::stringstream ss;
- ss << "data size not same, expected: " << request->size()
- << ", actual: " << request->data().size();
- response->mutable_status()->add_error_msgs(ss.str());
- response->mutable_status()->set_status_code(1);
-
- } else {
- Md5Digest digest;
- digest.update(static_cast<const void*>(request->data().c_str()), request->data().size());
- digest.digest();
- if (!iequal(digest.hex(), request->md5())) {
+ DorisMetrics::instance()->check_rpc_channel->increment(1);
+ _light_work_pool.offer([request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(0);
+ if (request->data().size() != request->size()) {
std::stringstream ss;
- ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
+ ss << "data size not same, expected: " << request->size()
+ << ", actual: " << request->data().size();
response->mutable_status()->add_error_msgs(ss.str());
response->mutable_status()->set_status_code(1);
+
+ } else {
+ Md5Digest digest;
+ digest.update(static_cast<const void*>(request->data().c_str()),
+ request->data().size());
+ digest.digest();
+ if (!iequal(digest.hex(), request->md5())) {
+ std::stringstream ss;
+ ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
+ response->mutable_status()->add_error_msgs(ss.str());
+ response->mutable_status()->set_status_code(1);
+ }
}
- }
+ });
}
void PInternalServiceImpl::reset_rpc_channel(google::protobuf::RpcController* controller,
const PResetRPCChannelRequest* request,
PResetRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->all()) {
- int size = ExecEnv::GetInstance()->brpc_internal_client_cache()->size();
- if (size > 0) {
- std::vector<std::string> endpoints;
- ExecEnv::GetInstance()->brpc_internal_client_cache()->get_all(&endpoints);
- ExecEnv::GetInstance()->brpc_internal_client_cache()->clear();
- *response->mutable_channels() = {endpoints.begin(), endpoints.end()};
- }
- } else {
- for (const std::string& endpoint : request->endpoints()) {
- if (!ExecEnv::GetInstance()->brpc_internal_client_cache()->exist(endpoint)) {
- response->mutable_status()->add_error_msgs(endpoint + ": not found.");
- continue;
+ DorisMetrics::instance()->reset_rpc_channel->increment(1);
+ _light_work_pool.offer([controller, request, response, done]() {
Review Comment:
warning: lambda capture 'this' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([ controller, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -693,75 +790,86 @@
const PCheckRPCChannelRequest* request,
PCheckRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->data().size() != request->size()) {
- std::stringstream ss;
- ss << "data size not same, expected: " << request->size()
- << ", actual: " << request->data().size();
- response->mutable_status()->add_error_msgs(ss.str());
- response->mutable_status()->set_status_code(1);
-
- } else {
- Md5Digest digest;
- digest.update(static_cast<const void*>(request->data().c_str()), request->data().size());
- digest.digest();
- if (!iequal(digest.hex(), request->md5())) {
+ DorisMetrics::instance()->check_rpc_channel->increment(1);
+ _light_work_pool.offer([request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(0);
+ if (request->data().size() != request->size()) {
std::stringstream ss;
- ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
+ ss << "data size not same, expected: " << request->size()
+ << ", actual: " << request->data().size();
response->mutable_status()->add_error_msgs(ss.str());
response->mutable_status()->set_status_code(1);
+
+ } else {
+ Md5Digest digest;
+ digest.update(static_cast<const void*>(request->data().c_str()),
+ request->data().size());
+ digest.digest();
+ if (!iequal(digest.hex(), request->md5())) {
+ std::stringstream ss;
+ ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
+ response->mutable_status()->add_error_msgs(ss.str());
+ response->mutable_status()->set_status_code(1);
+ }
}
- }
+ });
}
void PInternalServiceImpl::reset_rpc_channel(google::protobuf::RpcController* controller,
const PResetRPCChannelRequest* request,
PResetRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->all()) {
- int size = ExecEnv::GetInstance()->brpc_internal_client_cache()->size();
- if (size > 0) {
- std::vector<std::string> endpoints;
- ExecEnv::GetInstance()->brpc_internal_client_cache()->get_all(&endpoints);
- ExecEnv::GetInstance()->brpc_internal_client_cache()->clear();
- *response->mutable_channels() = {endpoints.begin(), endpoints.end()};
- }
- } else {
- for (const std::string& endpoint : request->endpoints()) {
- if (!ExecEnv::GetInstance()->brpc_internal_client_cache()->exist(endpoint)) {
- response->mutable_status()->add_error_msgs(endpoint + ": not found.");
- continue;
+ DorisMetrics::instance()->reset_rpc_channel->increment(1);
+ _light_work_pool.offer([controller, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -693,75 +790,86 @@
const PCheckRPCChannelRequest* request,
PCheckRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->data().size() != request->size()) {
- std::stringstream ss;
- ss << "data size not same, expected: " << request->size()
- << ", actual: " << request->data().size();
- response->mutable_status()->add_error_msgs(ss.str());
- response->mutable_status()->set_status_code(1);
-
- } else {
- Md5Digest digest;
- digest.update(static_cast<const void*>(request->data().c_str()), request->data().size());
- digest.digest();
- if (!iequal(digest.hex(), request->md5())) {
+ DorisMetrics::instance()->check_rpc_channel->increment(1);
+ _light_work_pool.offer([request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(0);
+ if (request->data().size() != request->size()) {
std::stringstream ss;
- ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
+ ss << "data size not same, expected: " << request->size()
+ << ", actual: " << request->data().size();
response->mutable_status()->add_error_msgs(ss.str());
response->mutable_status()->set_status_code(1);
+
+ } else {
+ Md5Digest digest;
+ digest.update(static_cast<const void*>(request->data().c_str()),
+ request->data().size());
+ digest.digest();
+ if (!iequal(digest.hex(), request->md5())) {
+ std::stringstream ss;
+ ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
+ response->mutable_status()->add_error_msgs(ss.str());
+ response->mutable_status()->set_status_code(1);
+ }
}
- }
+ });
}
void PInternalServiceImpl::reset_rpc_channel(google::protobuf::RpcController* controller,
const PResetRPCChannelRequest* request,
PResetRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->all()) {
- int size = ExecEnv::GetInstance()->brpc_internal_client_cache()->size();
- if (size > 0) {
- std::vector<std::string> endpoints;
- ExecEnv::GetInstance()->brpc_internal_client_cache()->get_all(&endpoints);
- ExecEnv::GetInstance()->brpc_internal_client_cache()->clear();
- *response->mutable_channels() = {endpoints.begin(), endpoints.end()};
- }
- } else {
- for (const std::string& endpoint : request->endpoints()) {
- if (!ExecEnv::GetInstance()->brpc_internal_client_cache()->exist(endpoint)) {
- response->mutable_status()->add_error_msgs(endpoint + ": not found.");
- continue;
+ DorisMetrics::instance()->reset_rpc_channel->increment(1);
+ _light_work_pool.offer([controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(0);
+ if (request->all()) {
+ int size = ExecEnv::GetInstance()->brpc_internal_client_cache()->size();
+ if (size > 0) {
+ std::vector<std::string> endpoints;
+ ExecEnv::GetInstance()->brpc_internal_client_cache()->get_all(&endpoints);
+ ExecEnv::GetInstance()->brpc_internal_client_cache()->clear();
+ *response->mutable_channels() = {endpoints.begin(), endpoints.end()};
}
+ } else {
+ for (const std::string& endpoint : request->endpoints()) {
+ if (!ExecEnv::GetInstance()->brpc_internal_client_cache()->exist(endpoint)) {
+ response->mutable_status()->add_error_msgs(endpoint + ": not found.");
+ continue;
+ }
- if (ExecEnv::GetInstance()->brpc_internal_client_cache()->erase(endpoint)) {
- response->add_channels(endpoint);
- } else {
- response->mutable_status()->add_error_msgs(endpoint + ": reset failed.");
+ if (ExecEnv::GetInstance()->brpc_internal_client_cache()->erase(endpoint)) {
+ response->add_channels(endpoint);
+ } else {
+ response->mutable_status()->add_error_msgs(endpoint + ": reset failed.");
+ }
+ }
+ if (request->endpoints_size() != response->channels_size()) {
+ response->mutable_status()->set_status_code(1);
}
}
- if (request->endpoints_size() != response->channels_size()) {
- response->mutable_status()->set_status_code(1);
- }
- }
+ });
}
-void PInternalServiceImpl::hand_shake(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::hand_shake(google::protobuf::RpcController* controller,
const PHandShakeRequest* request,
PHandShakeResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- if (request->has_hello()) {
- response->set_hello(request->hello());
- }
- response->mutable_status()->set_status_code(0);
+ DorisMetrics::instance()->hand_shake->increment(1);
+ _light_work_pool.offer([request, response, done]() {
Review Comment:
warning: lambda capture 'this' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([ request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -693,75 +790,86 @@
const PCheckRPCChannelRequest* request,
PCheckRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->data().size() != request->size()) {
- std::stringstream ss;
- ss << "data size not same, expected: " << request->size()
- << ", actual: " << request->data().size();
- response->mutable_status()->add_error_msgs(ss.str());
- response->mutable_status()->set_status_code(1);
-
- } else {
- Md5Digest digest;
- digest.update(static_cast<const void*>(request->data().c_str()), request->data().size());
- digest.digest();
- if (!iequal(digest.hex(), request->md5())) {
+ DorisMetrics::instance()->check_rpc_channel->increment(1);
+ _light_work_pool.offer([request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([ request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -693,75 +790,86 @@ void PInternalServiceImpl::check_rpc_channel(google::protobuf::RpcController* co
const PCheckRPCChannelRequest* request,
PCheckRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->data().size() != request->size()) {
- std::stringstream ss;
- ss << "data size not same, expected: " << request->size()
- << ", actual: " << request->data().size();
- response->mutable_status()->add_error_msgs(ss.str());
- response->mutable_status()->set_status_code(1);
-
- } else {
- Md5Digest digest;
- digest.update(static_cast<const void*>(request->data().c_str()), request->data().size());
- digest.digest();
- if (!iequal(digest.hex(), request->md5())) {
+ DorisMetrics::instance()->check_rpc_channel->increment(1);
+ _light_work_pool.offer([request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(0);
+ if (request->data().size() != request->size()) {
std::stringstream ss;
- ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
+ ss << "data size not same, expected: " << request->size()
+ << ", actual: " << request->data().size();
response->mutable_status()->add_error_msgs(ss.str());
response->mutable_status()->set_status_code(1);
+
+ } else {
+ Md5Digest digest;
+ digest.update(static_cast<const void*>(request->data().c_str()),
+ request->data().size());
+ digest.digest();
+ if (!iequal(digest.hex(), request->md5())) {
+ std::stringstream ss;
+ ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
+ response->mutable_status()->add_error_msgs(ss.str());
+ response->mutable_status()->set_status_code(1);
+ }
}
- }
+ });
}
void PInternalServiceImpl::reset_rpc_channel(google::protobuf::RpcController* controller,
const PResetRPCChannelRequest* request,
PResetRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->all()) {
- int size = ExecEnv::GetInstance()->brpc_internal_client_cache()->size();
- if (size > 0) {
- std::vector<std::string> endpoints;
- ExecEnv::GetInstance()->brpc_internal_client_cache()->get_all(&endpoints);
- ExecEnv::GetInstance()->brpc_internal_client_cache()->clear();
- *response->mutable_channels() = {endpoints.begin(), endpoints.end()};
- }
- } else {
- for (const std::string& endpoint : request->endpoints()) {
- if (!ExecEnv::GetInstance()->brpc_internal_client_cache()->exist(endpoint)) {
- response->mutable_status()->add_error_msgs(endpoint + ": not found.");
- continue;
+ DorisMetrics::instance()->reset_rpc_channel->increment(1);
+ _light_work_pool.offer([controller, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([ request, response, done]() {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] morningman commented on pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1427216658
Please also update this document:
https://doris.apache.org/zh-CN/docs/dev/admin-manual/maint-monitor/monitor-metrics/metrics
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on a diff in pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1104596492
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
- _exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
- }
- Status::OK().to_protobuf(response->mutable_status());
+ Status::OK().to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController* controller,
const PUpdateCacheRequest* request,
PCacheResponse* response, google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ DorisMetrics::instance()->update_cache->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->update(request, response);
+ });
}
void PInternalServiceImpl::fetch_cache(google::protobuf::RpcController* controller,
const PFetchCacheRequest* request, PFetchCacheResult* result,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->fetch(request, result);
+ DorisMetrics::instance()->fetch_cache->increment(1);
+ _heavy_work_pool.offer([this, request, result, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->fetch(request, result);
+ });
}
Review Comment:
I didn't do such test
only rpc shutdown will offer failure, other situations will always return to success
##########
be/src/common/config.h:
##########
@@ -35,8 +35,10 @@ CONF_Int32(be_port, "9060");
// port for brpc
CONF_Int32(brpc_port, "8060");
-// the number of bthreads for brpc, the default value is set to -1, which means the number of bthreads is #cpu-cores
-CONF_Int32(brpc_num_threads, "-1");
+// the number of bthreads for brpc, the default value is set to 32
+// brpc only for network service send or accept request
+// no more process any logic
+CONF_Int32(brpc_num_threads, "32");
Review Comment:
for exmaple if the host have 128 cores, the brpc default behavior is create 128 pthread to handle the bthread work , since we move the logic to thread pool, brpc only for network i/o , i don't think need 128 pthread for brpc, so i think we need to decrease the brpc thread and increase thread pool thread
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1428126937
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong closed pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong closed pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
URL: https://github.com/apache/doris/pull/16639
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] morningman commented on a diff in pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "morningman (via GitHub)" <gi...@apache.org>.
morningman commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103927268
##########
be/src/service/internal_service.cpp:
##########
@@ -286,125 +334,134 @@ Status PInternalServiceImpl::_exec_plan_fragment(const std::string& ser_request,
}
}
-void PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController* controller,
const PCancelPlanFragmentRequest* request,
PCancelPlanFragmentResult* result,
google::protobuf::Closure* done) {
- auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start", cntl_base);
- auto scope = OpentelemetryScope {span};
- brpc::ClosureGuard closure_guard(done);
- TUniqueId tid;
- tid.__set_hi(request->finst_id().hi());
- tid.__set_lo(request->finst_id().lo());
-
- Status st = Status::OK();
- if (request->has_cancel_reason()) {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
- << ", reason: " << request->cancel_reason();
- _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
- } else {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
- _exec_env->fragment_mgr()->cancel(tid);
- }
-
- // TODO: the logic seems useless, cancel only return Status::OK. remove it
- st.to_protobuf(result->mutable_status());
+ DorisMetrics::instance()->cancel_plan_fragment->increment(1);
+ _light_work_pool.offer([this, controller, request, result, done]() {
+ auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start", controller);
+ auto scope = OpentelemetryScope {span};
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId tid;
+ tid.__set_hi(request->finst_id().hi());
+ tid.__set_lo(request->finst_id().lo());
+
+ Status st = Status::OK();
+ if (request->has_cancel_reason()) {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
+ << ", reason: " << request->cancel_reason();
+ _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
+ } else {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
+ _exec_env->fragment_mgr()->cancel(tid);
+ }
+ // TODO: the logic seems useless, cancel only return Status::OK. remove it
+ st.to_protobuf(result->mutable_status());
+ });
}
-void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* controller,
const PFetchDataRequest* request, PFetchDataResult* result,
google::protobuf::Closure* done) {
- brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
- GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
- _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ DorisMetrics::instance()->fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, result, done]() {
+ brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
+ _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ });
}
void PInternalServiceImpl::fetch_table_schema(google::protobuf::RpcController* controller,
const PFetchTableSchemaRequest* request,
PFetchTableSchemaResult* result,
google::protobuf::Closure* done) {
- VLOG_RPC << "fetch table schema";
- brpc::ClosureGuard closure_guard(done);
- TFileScanRange file_scan_range;
- Status st = Status::OK();
- {
- const uint8_t* buf = (const uint8_t*)(request->file_scan_range().data());
- uint32_t len = request->file_scan_range().size();
- st = deserialize_thrift_msg(buf, &len, false, &file_scan_range);
+ DorisMetrics::instance()->fetch_table_schema->increment(1);
+ _light_work_pool.offer([request, result, done]() {
Review Comment:
This is a heavy work, it may read the remote file
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
Review Comment:
This is a heavy work, it will visit kafka broker, which is a remote service.
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
- _exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
- }
- Status::OK().to_protobuf(response->mutable_status());
+ Status::OK().to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController* controller,
const PUpdateCacheRequest* request,
PCacheResponse* response, google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ DorisMetrics::instance()->update_cache->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->update(request, response);
+ });
}
void PInternalServiceImpl::fetch_cache(google::protobuf::RpcController* controller,
const PFetchCacheRequest* request, PFetchCacheResult* result,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->fetch(request, result);
+ DorisMetrics::instance()->fetch_cache->increment(1);
+ _heavy_work_pool.offer([this, request, result, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->fetch(request, result);
+ });
}
void PInternalServiceImpl::clear_cache(google::protobuf::RpcController* controller,
const PClearCacheRequest* request, PCacheResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->clear(request, response);
+ DorisMetrics::instance()->clear_cache->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->clear(request, response);
+ });
}
void PInternalServiceImpl::merge_filter(::google::protobuf::RpcController* controller,
const ::doris::PMergeFilterRequest* request,
::doris::PMergeFilterResponse* response,
::google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- auto attachment = static_cast<brpc::Controller*>(controller)->request_attachment();
- butil::IOBufAsZeroCopyInputStream zero_copy_input_stream(attachment);
- Status st = _exec_env->fragment_mgr()->merge_filter(request, &zero_copy_input_stream);
- if (!st.ok()) {
- LOG(WARNING) << "merge meet error" << st.to_string();
- }
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->merge_filter->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ auto attachment = static_cast<brpc::Controller*>(controller)->request_attachment();
+ butil::IOBufAsZeroCopyInputStream zero_copy_input_stream(attachment);
+ Status st = _exec_env->fragment_mgr()->merge_filter(request, &zero_copy_input_stream);
+ if (!st.ok()) {
+ LOG(WARNING) << "merge meet error" << st.to_string();
+ }
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::apply_filter(::google::protobuf::RpcController* controller,
const ::doris::PPublishFilterRequest* request,
::doris::PPublishFilterResponse* response,
::google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- auto attachment = static_cast<brpc::Controller*>(controller)->request_attachment();
- butil::IOBufAsZeroCopyInputStream zero_copy_input_stream(attachment);
- UniqueId unique_id(request->query_id());
- VLOG_NOTICE << "rpc apply_filter recv";
- Status st = _exec_env->fragment_mgr()->apply_filter(request, &zero_copy_input_stream);
- if (!st.ok()) {
- LOG(WARNING) << "apply filter meet error: " << st.to_string();
- }
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->apply_filter->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ auto attachment = static_cast<brpc::Controller*>(controller)->request_attachment();
+ butil::IOBufAsZeroCopyInputStream zero_copy_input_stream(attachment);
+ UniqueId unique_id(request->query_id());
+ VLOG_NOTICE << "rpc apply_filter recv";
+ Status st = _exec_env->fragment_mgr()->apply_filter(request, &zero_copy_input_stream);
+ if (!st.ok()) {
+ LOG(WARNING) << "apply filter meet error: " << st.to_string();
+ }
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::send_data(google::protobuf::RpcController* controller,
const PSendDataRequest* request, PSendDataResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- TUniqueId fragment_instance_id;
- fragment_instance_id.hi = request->fragment_instance_id().hi();
- fragment_instance_id.lo = request->fragment_instance_id().lo();
-
- auto pipe = _exec_env->fragment_mgr()->get_pipe(fragment_instance_id);
- if (pipe == nullptr) {
- response->mutable_status()->set_status_code(1);
- response->mutable_status()->add_error_msgs("pipe is null");
- } else {
- for (int i = 0; i < request->data_size(); ++i) {
- PDataRow* row = new PDataRow();
- row->CopyFrom(request->data(i));
- pipe->append_and_flush(reinterpret_cast<char*>(&row), sizeof(row),
- sizeof(row) + row->ByteSizeLong());
+ DorisMetrics::instance()->send_data->increment(1);
+ _heavy_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId fragment_instance_id;
+ fragment_instance_id.hi = request->fragment_instance_id().hi();
+ fragment_instance_id.lo = request->fragment_instance_id().lo();
+
+ auto pipe = _exec_env->fragment_mgr()->get_pipe(fragment_instance_id);
+ if (pipe == nullptr) {
+ response->mutable_status()->set_status_code(1);
+ response->mutable_status()->add_error_msgs("pipe is null");
+ } else {
+ for (int i = 0; i < request->data_size(); ++i) {
+ PDataRow* row = new PDataRow();
+ row->CopyFrom(request->data(i));
+ pipe->append_and_flush(reinterpret_cast<char*>(&row), sizeof(row),
+ sizeof(row) + row->ByteSizeLong());
+ }
+ response->mutable_status()->set_status_code(0);
}
- response->mutable_status()->set_status_code(0);
- }
+ });
}
void PInternalServiceImpl::commit(google::protobuf::RpcController* controller,
const PCommitRequest* request, PCommitResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- TUniqueId fragment_instance_id;
- fragment_instance_id.hi = request->fragment_instance_id().hi();
- fragment_instance_id.lo = request->fragment_instance_id().lo();
-
- auto pipe = _exec_env->fragment_mgr()->get_pipe(fragment_instance_id);
- if (pipe == nullptr) {
- response->mutable_status()->set_status_code(1);
- response->mutable_status()->add_error_msgs("pipe is null");
- } else {
- pipe->finish();
- response->mutable_status()->set_status_code(0);
- }
+ DorisMetrics::instance()->commit->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId fragment_instance_id;
+ fragment_instance_id.hi = request->fragment_instance_id().hi();
+ fragment_instance_id.lo = request->fragment_instance_id().lo();
+
+ auto pipe = _exec_env->fragment_mgr()->get_pipe(fragment_instance_id);
+ if (pipe == nullptr) {
+ response->mutable_status()->set_status_code(1);
+ response->mutable_status()->add_error_msgs("pipe is null");
+ } else {
+ pipe->finish();
+ response->mutable_status()->set_status_code(0);
+ }
+ });
}
void PInternalServiceImpl::rollback(google::protobuf::RpcController* controller,
const PRollbackRequest* request, PRollbackResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- TUniqueId fragment_instance_id;
- fragment_instance_id.hi = request->fragment_instance_id().hi();
- fragment_instance_id.lo = request->fragment_instance_id().lo();
-
- auto pipe = _exec_env->fragment_mgr()->get_pipe(fragment_instance_id);
- if (pipe == nullptr) {
- response->mutable_status()->set_status_code(1);
- response->mutable_status()->add_error_msgs("pipe is null");
- } else {
- pipe->cancel("rollback");
- response->mutable_status()->set_status_code(0);
- }
+ DorisMetrics::instance()->rollback->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId fragment_instance_id;
+ fragment_instance_id.hi = request->fragment_instance_id().hi();
+ fragment_instance_id.lo = request->fragment_instance_id().lo();
+
+ auto pipe = _exec_env->fragment_mgr()->get_pipe(fragment_instance_id);
+ if (pipe == nullptr) {
+ response->mutable_status()->set_status_code(1);
+ response->mutable_status()->add_error_msgs("pipe is null");
+ } else {
+ pipe->cancel("rollback");
+ response->mutable_status()->set_status_code(0);
+ }
+ });
}
-void PInternalServiceImpl::fold_constant_expr(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::fold_constant_expr(google::protobuf::RpcController* controller,
const PConstantExprRequest* request,
PConstantExprResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
+ DorisMetrics::instance()->fold_constant_expr->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- Status st = Status::OK();
- if (request->has_request()) {
- st = _fold_constant_expr(request->request(), response);
- } else {
- // TODO(yangzhengguo) this is just for compatible with old version, this should be removed in the release 0.15
- st = _fold_constant_expr(cntl->request_attachment().to_string(), response);
- }
- if (!st.ok()) {
- LOG(WARNING) << "exec fold constant expr failed, errmsg=" << st;
- }
- st.to_protobuf(response->mutable_status());
+ Status st = Status::OK();
+ if (request->has_request()) {
+ st = _fold_constant_expr(request->request(), response);
+ } else {
+ // TODO(yangzhengguo) this is just for compatible with old version, this should be removed in the release 0.15
Review Comment:
Remove this TODO?
##########
be/src/service/internal_service.cpp:
##########
@@ -1097,9 +1208,8 @@ void PInternalServiceImpl::multiget_data(google::protobuf::RpcController* contro
const PMultiGetRequest* request,
PMultiGetResponse* response,
google::protobuf::Closure* done) {
- // Submit task to seperate ThreadPool for avoiding block bthread working pthread
- ThreadPool* task_pool = StorageEngine::instance()->get_bg_multiget_threadpool();
- Status submit_st = task_pool->submit_func([request, response, done, this]() {
+ DorisMetrics::instance()->multiget_data->increment(1);
+ _light_work_pool.offer([request, response, done, this]() {
Review Comment:
This is a heavy work, because it will open and read segment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] dataroaring commented on a diff in pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "dataroaring (via GitHub)" <gi...@apache.org>.
dataroaring commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103931956
##########
be/src/util/doris_metrics.cpp:
##########
@@ -197,6 +230,37 @@ DorisMetrics::DorisMetrics() : _metric_registry(_s_registry_name) {
_server_metric_entity = _metric_registry.register_entity("server");
INT_COUNTER_METRIC_REGISTER(_server_metric_entity, fragment_requests_total);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, transmit_data);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, transmit_data_by_http);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, exec_plan_fragment);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, exec_plan_fragment_prepare);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, exec_plan_fragment_start);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, cancel_plan_fragment);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, fetch_data);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, fetch_table_schema);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, tablet_writer_open);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, tablet_writer_add_block);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, tablet_writer_add_block_by_http);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, tablet_writer_cancel);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, get_info);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, update_cache);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, fetch_cache);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, clear_cache);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, merge_filter);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, apply_filter);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, transmit_block);
+ INT_COUNTER_METRIC_REGISTER(_server_metric_entity, transmit_block_by_http);
Review Comment:
actually, there are these metrics in bvar in brpc already.
##########
be/src/common/config.h:
##########
@@ -35,8 +35,10 @@ CONF_Int32(be_port, "9060");
// port for brpc
CONF_Int32(brpc_port, "8060");
-// the number of bthreads for brpc, the default value is set to -1, which means the number of bthreads is #cpu-cores
-CONF_Int32(brpc_num_threads, "-1");
+// the number of bthreads for brpc, the default value is set to 32
+// brpc only for network service send or accept request
+// no more process any logic
+CONF_Int32(brpc_num_threads, "32");
Review Comment:
We do not need to change default value here, e.g. if a user runs server on a machine with 128 cores, it would break his/her config used.
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
- _exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
- }
- Status::OK().to_protobuf(response->mutable_status());
+ Status::OK().to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController* controller,
const PUpdateCacheRequest* request,
PCacheResponse* response, google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ DorisMetrics::instance()->update_cache->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->update(request, response);
+ });
}
void PInternalServiceImpl::fetch_cache(google::protobuf::RpcController* controller,
const PFetchCacheRequest* request, PFetchCacheResult* result,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->fetch(request, result);
+ DorisMetrics::instance()->fetch_cache->increment(1);
+ _heavy_work_pool.offer([this, request, result, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->fetch(request, result);
+ });
}
Review Comment:
If offer fails, what's the result of brpc in result ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on a diff in pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103938347
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
Review Comment:
ok
##########
be/src/service/internal_service.cpp:
##########
@@ -286,125 +334,134 @@ Status PInternalServiceImpl::_exec_plan_fragment(const std::string& ser_request,
}
}
-void PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController* controller,
const PCancelPlanFragmentRequest* request,
PCancelPlanFragmentResult* result,
google::protobuf::Closure* done) {
- auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start", cntl_base);
- auto scope = OpentelemetryScope {span};
- brpc::ClosureGuard closure_guard(done);
- TUniqueId tid;
- tid.__set_hi(request->finst_id().hi());
- tid.__set_lo(request->finst_id().lo());
-
- Status st = Status::OK();
- if (request->has_cancel_reason()) {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
- << ", reason: " << request->cancel_reason();
- _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
- } else {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
- _exec_env->fragment_mgr()->cancel(tid);
- }
-
- // TODO: the logic seems useless, cancel only return Status::OK. remove it
- st.to_protobuf(result->mutable_status());
+ DorisMetrics::instance()->cancel_plan_fragment->increment(1);
+ _light_work_pool.offer([this, controller, request, result, done]() {
+ auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start", controller);
+ auto scope = OpentelemetryScope {span};
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId tid;
+ tid.__set_hi(request->finst_id().hi());
+ tid.__set_lo(request->finst_id().lo());
+
+ Status st = Status::OK();
+ if (request->has_cancel_reason()) {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
+ << ", reason: " << request->cancel_reason();
+ _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
+ } else {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
+ _exec_env->fragment_mgr()->cancel(tid);
+ }
+ // TODO: the logic seems useless, cancel only return Status::OK. remove it
+ st.to_protobuf(result->mutable_status());
+ });
}
-void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* controller,
const PFetchDataRequest* request, PFetchDataResult* result,
google::protobuf::Closure* done) {
- brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
- GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
- _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ DorisMetrics::instance()->fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, result, done]() {
+ brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
+ _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ });
}
void PInternalServiceImpl::fetch_table_schema(google::protobuf::RpcController* controller,
const PFetchTableSchemaRequest* request,
PFetchTableSchemaResult* result,
google::protobuf::Closure* done) {
- VLOG_RPC << "fetch table schema";
- brpc::ClosureGuard closure_guard(done);
- TFileScanRange file_scan_range;
- Status st = Status::OK();
- {
- const uint8_t* buf = (const uint8_t*)(request->file_scan_range().data());
- uint32_t len = request->file_scan_range().size();
- st = deserialize_thrift_msg(buf, &len, false, &file_scan_range);
+ DorisMetrics::instance()->fetch_table_schema->increment(1);
+ _light_work_pool.offer([request, result, done]() {
Review Comment:
ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on a diff in pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103937266
##########
be/src/common/config.h:
##########
@@ -35,8 +35,10 @@ CONF_Int32(be_port, "9060");
// port for brpc
CONF_Int32(brpc_port, "8060");
-// the number of bthreads for brpc, the default value is set to -1, which means the number of bthreads is #cpu-cores
-CONF_Int32(brpc_num_threads, "-1");
+// the number of bthreads for brpc, the default value is set to 32
+// brpc only for network service send or accept request
+// no more process any logic
+CONF_Int32(brpc_num_threads, "32");
Review Comment:
brpc now only for network i/o, i think no need to configure so many threads
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] dataroaring commented on a diff in pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "dataroaring (via GitHub)" <gi...@apache.org>.
dataroaring commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1104568288
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
- _exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
- }
- Status::OK().to_protobuf(response->mutable_status());
+ Status::OK().to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController* controller,
const PUpdateCacheRequest* request,
PCacheResponse* response, google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ DorisMetrics::instance()->update_cache->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->update(request, response);
+ });
}
void PInternalServiceImpl::fetch_cache(google::protobuf::RpcController* controller,
const PFetchCacheRequest* request, PFetchCacheResult* result,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->fetch(request, result);
+ DorisMetrics::instance()->fetch_cache->increment(1);
+ _heavy_work_pool.offer([this, request, result, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->fetch(request, result);
+ });
}
Review Comment:
Do you test this case like mock a offer failure? In the case, we do not set status, so i do not how brpc handle it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1439340656
run p0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] dataroaring commented on a diff in pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "dataroaring (via GitHub)" <gi...@apache.org>.
dataroaring commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1104566072
##########
be/src/common/config.h:
##########
@@ -35,8 +35,10 @@ CONF_Int32(be_port, "9060");
// port for brpc
CONF_Int32(brpc_port, "8060");
-// the number of bthreads for brpc, the default value is set to -1, which means the number of bthreads is #cpu-cores
-CONF_Int32(brpc_num_threads, "-1");
+// the number of bthreads for brpc, the default value is set to 32
+// brpc only for network service send or accept request
+// no more process any logic
+CONF_Int32(brpc_num_threads, "32");
Review Comment:
brpc creates a pthread for each cpu, this is a default behavior, and we do not care it on any machine if we use it in a right way. I can not see any benefits on changing it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on a diff in pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103646730
##########
be/src/service/internal_service.cpp:
##########
@@ -286,125 +334,134 @@ Status PInternalServiceImpl::_exec_plan_fragment(const std::string& ser_request,
}
}
-void PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController* controller,
const PCancelPlanFragmentRequest* request,
PCancelPlanFragmentResult* result,
google::protobuf::Closure* done) {
- auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start", cntl_base);
- auto scope = OpentelemetryScope {span};
- brpc::ClosureGuard closure_guard(done);
- TUniqueId tid;
- tid.__set_hi(request->finst_id().hi());
- tid.__set_lo(request->finst_id().lo());
-
- Status st = Status::OK();
- if (request->has_cancel_reason()) {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
- << ", reason: " << request->cancel_reason();
- _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
- } else {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
- _exec_env->fragment_mgr()->cancel(tid);
- }
-
- // TODO: the logic seems useless, cancel only return Status::OK. remove it
- st.to_protobuf(result->mutable_status());
+ DorisMetrics::instance()->cancel_plan_fragment->increment(1);
+ _light_work_pool.offer([this, controller, request, result, done]() {
+ auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start", controller);
+ auto scope = OpentelemetryScope {span};
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId tid;
+ tid.__set_hi(request->finst_id().hi());
+ tid.__set_lo(request->finst_id().lo());
+
+ Status st = Status::OK();
+ if (request->has_cancel_reason()) {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
+ << ", reason: " << request->cancel_reason();
+ _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
+ } else {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
+ _exec_env->fragment_mgr()->cancel(tid);
+ }
+ // TODO: the logic seems useless, cancel only return Status::OK. remove it
+ st.to_protobuf(result->mutable_status());
+ });
}
-void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* controller,
const PFetchDataRequest* request, PFetchDataResult* result,
google::protobuf::Closure* done) {
- brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
- GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
- _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ DorisMetrics::instance()->fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, result, done]() {
+ brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
+ _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ });
}
void PInternalServiceImpl::fetch_table_schema(google::protobuf::RpcController* controller,
const PFetchTableSchemaRequest* request,
PFetchTableSchemaResult* result,
google::protobuf::Closure* done) {
- VLOG_RPC << "fetch table schema";
- brpc::ClosureGuard closure_guard(done);
- TFileScanRange file_scan_range;
- Status st = Status::OK();
- {
- const uint8_t* buf = (const uint8_t*)(request->file_scan_range().data());
- uint32_t len = request->file_scan_range().size();
- st = deserialize_thrift_msg(buf, &len, false, &file_scan_range);
+ DorisMetrics::instance()->fetch_table_schema->increment(1);
+ _light_work_pool.offer([this, request, result, done]() {
Review Comment:
warning: lambda capture 'this' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([ request, result, done]() {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1426811566
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1437845411
run p0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1437746162
run beut
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on a diff in pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1111618765
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +530,278 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ bool ret = _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
+ if (!ret) {
+ LOG(WARNING) << "fail to offer request to the work pool";
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(TStatusCode::CANCELLED);
+ response->mutable_status()->add_error_msgs("fail to offer request to the work pool");
+ }
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ bool ret = _heavy_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
- _exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
+ Status::OK().to_protobuf(response->mutable_status());
+ });
+ if (!ret) {
+ LOG(WARNING) << "fail to offer request to the work pool";
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(TStatusCode::CANCELLED);
+ response->mutable_status()->add_error_msgs("fail to offer request to the work pool");
}
- Status::OK().to_protobuf(response->mutable_status());
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController* controller,
const PUpdateCacheRequest* request,
PCacheResponse* response, google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ bool ret = _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->update(request, response);
+ });
+ if (!ret) {
+ LOG(WARNING) << "fail to offer request to the work pool";
+ brpc::ClosureGuard closure_guard(done);
+ response->set_status(PCacheStatus::CANCELED);
Review Comment:
warning: no member named 'CANCELED' in 'doris::PCacheStatus' [clang-diagnostic-error]
```cpp
response->set_status(PCacheStatus::CANCELED);
^
```
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +530,278 @@
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ bool ret = _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
+ if (!ret) {
+ LOG(WARNING) << "fail to offer request to the work pool";
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(TStatusCode::CANCELLED);
+ response->mutable_status()->add_error_msgs("fail to offer request to the work pool");
+ }
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ bool ret = _heavy_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
- _exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
+ Status::OK().to_protobuf(response->mutable_status());
+ });
+ if (!ret) {
+ LOG(WARNING) << "fail to offer request to the work pool";
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(TStatusCode::CANCELLED);
+ response->mutable_status()->add_error_msgs("fail to offer request to the work pool");
}
- Status::OK().to_protobuf(response->mutable_status());
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController* controller,
const PUpdateCacheRequest* request,
PCacheResponse* response, google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ bool ret = _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->update(request, response);
+ });
+ if (!ret) {
+ LOG(WARNING) << "fail to offer request to the work pool";
+ brpc::ClosureGuard closure_guard(done);
+ response->set_status(PCacheStatus::CANCELED);
+ }
}
void PInternalServiceImpl::fetch_cache(google::protobuf::RpcController* controller,
const PFetchCacheRequest* request, PFetchCacheResult* result,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->fetch(request, result);
+ bool ret = _heavy_work_pool.offer([this, request, result, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->fetch(request, result);
+ });
+ if (!ret) {
+ LOG(WARNING) << "fail to offer request to the work pool";
+ brpc::ClosureGuard closure_guard(done);
+ result->set_status(PCacheStatus::CANCELED);
Review Comment:
warning: no member named 'CANCELED' in 'doris::PCacheStatus' [clang-diagnostic-error]
```cpp
result->set_status(PCacheStatus::CANCELED);
^
```
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +530,278 @@
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ bool ret = _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
+ if (!ret) {
+ LOG(WARNING) << "fail to offer request to the work pool";
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(TStatusCode::CANCELLED);
+ response->mutable_status()->add_error_msgs("fail to offer request to the work pool");
+ }
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ bool ret = _heavy_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
- _exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
+ Status::OK().to_protobuf(response->mutable_status());
+ });
+ if (!ret) {
+ LOG(WARNING) << "fail to offer request to the work pool";
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(TStatusCode::CANCELLED);
+ response->mutable_status()->add_error_msgs("fail to offer request to the work pool");
}
- Status::OK().to_protobuf(response->mutable_status());
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController* controller,
const PUpdateCacheRequest* request,
PCacheResponse* response, google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ bool ret = _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->update(request, response);
+ });
+ if (!ret) {
+ LOG(WARNING) << "fail to offer request to the work pool";
+ brpc::ClosureGuard closure_guard(done);
+ response->set_status(PCacheStatus::CANCELED);
+ }
}
void PInternalServiceImpl::fetch_cache(google::protobuf::RpcController* controller,
const PFetchCacheRequest* request, PFetchCacheResult* result,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->fetch(request, result);
+ bool ret = _heavy_work_pool.offer([this, request, result, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->fetch(request, result);
+ });
+ if (!ret) {
+ LOG(WARNING) << "fail to offer request to the work pool";
+ brpc::ClosureGuard closure_guard(done);
+ result->set_status(PCacheStatus::CANCELED);
+ }
}
void PInternalServiceImpl::clear_cache(google::protobuf::RpcController* controller,
const PClearCacheRequest* request, PCacheResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->clear(request, response);
+ bool ret = _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->clear(request, response);
+ });
+ if (!ret) {
+ LOG(WARNING) << "fail to offer request to the work pool";
+ brpc::ClosureGuard closure_guard(done);
+ response->set_status(PCacheStatus::CANCELED);
Review Comment:
warning: no member named 'CANCELED' in 'doris::PCacheStatus' [clang-diagnostic-error]
```cpp
response->set_status(PCacheStatus::CANCELED);
^
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on a diff in pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103645984
##########
be/src/service/internal_service.cpp:
##########
@@ -247,15 +292,18 @@
const PTabletWriterCancelRequest* request,
PTabletWriterCancelResult* response,
google::protobuf::Closure* done) {
- VLOG_RPC << "tablet writer cancel, id=" << request->id() << ", index_id=" << request->index_id()
- << ", sender_id=" << request->sender_id();
- brpc::ClosureGuard closure_guard(done);
- auto st = _exec_env->load_channel_mgr()->cancel(*request);
- if (!st.ok()) {
- LOG(WARNING) << "tablet writer cancel failed, id=" << request->id()
- << ", index_id=" << request->index_id()
- << ", sender_id=" << request->sender_id();
- }
+ DorisMetrics::instance()->tablet_writer_cancel->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
warning: lambda capture 'response' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, controller, request, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -286,125 +334,134 @@
}
}
-void PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController* controller,
const PCancelPlanFragmentRequest* request,
PCancelPlanFragmentResult* result,
google::protobuf::Closure* done) {
- auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start", cntl_base);
- auto scope = OpentelemetryScope {span};
- brpc::ClosureGuard closure_guard(done);
- TUniqueId tid;
- tid.__set_hi(request->finst_id().hi());
- tid.__set_lo(request->finst_id().lo());
-
- Status st = Status::OK();
- if (request->has_cancel_reason()) {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
- << ", reason: " << request->cancel_reason();
- _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
- } else {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
- _exec_env->fragment_mgr()->cancel(tid);
- }
-
- // TODO: the logic seems useless, cancel only return Status::OK. remove it
- st.to_protobuf(result->mutable_status());
+ DorisMetrics::instance()->cancel_plan_fragment->increment(1);
+ _light_work_pool.offer([this, controller, request, result, done]() {
+ auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start", controller);
+ auto scope = OpentelemetryScope {span};
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId tid;
+ tid.__set_hi(request->finst_id().hi());
+ tid.__set_lo(request->finst_id().lo());
+
+ Status st = Status::OK();
+ if (request->has_cancel_reason()) {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
+ << ", reason: " << request->cancel_reason();
+ _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
+ } else {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
+ _exec_env->fragment_mgr()->cancel(tid);
+ }
+ // TODO: the logic seems useless, cancel only return Status::OK. remove it
+ st.to_protobuf(result->mutable_status());
+ });
}
-void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* controller,
const PFetchDataRequest* request, PFetchDataResult* result,
google::protobuf::Closure* done) {
- brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
- GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
- _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ DorisMetrics::instance()->fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, result, done]() {
+ brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
+ _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ });
}
void PInternalServiceImpl::fetch_table_schema(google::protobuf::RpcController* controller,
const PFetchTableSchemaRequest* request,
PFetchTableSchemaResult* result,
google::protobuf::Closure* done) {
- VLOG_RPC << "fetch table schema";
- brpc::ClosureGuard closure_guard(done);
- TFileScanRange file_scan_range;
- Status st = Status::OK();
- {
- const uint8_t* buf = (const uint8_t*)(request->file_scan_range().data());
- uint32_t len = request->file_scan_range().size();
- st = deserialize_thrift_msg(buf, &len, false, &file_scan_range);
+ DorisMetrics::instance()->fetch_table_schema->increment(1);
+ _light_work_pool.offer([this, controller, request, result, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, result, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
- _exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
- }
- Status::OK().to_protobuf(response->mutable_status());
+ Status::OK().to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController* controller,
const PUpdateCacheRequest* request,
PCacheResponse* response, google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ DorisMetrics::instance()->update_cache->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->update(request, response);
+ });
}
void PInternalServiceImpl::fetch_cache(google::protobuf::RpcController* controller,
const PFetchCacheRequest* request, PFetchCacheResult* result,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->fetch(request, result);
+ DorisMetrics::instance()->fetch_cache->increment(1);
+ _heavy_work_pool.offer([this, controller, request, result, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->fetch(request, result);
+ });
}
void PInternalServiceImpl::clear_cache(google::protobuf::RpcController* controller,
const PClearCacheRequest* request, PCacheResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->clear(request, response);
+ DorisMetrics::instance()->clear_cache->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
- _exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
- }
- Status::OK().to_protobuf(response->mutable_status());
+ Status::OK().to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController* controller,
const PUpdateCacheRequest* request,
PCacheResponse* response, google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ DorisMetrics::instance()->update_cache->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->update(request, response);
+ });
}
void PInternalServiceImpl::fetch_cache(google::protobuf::RpcController* controller,
const PFetchCacheRequest* request, PFetchCacheResult* result,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->fetch(request, result);
+ DorisMetrics::instance()->fetch_cache->increment(1);
+ _heavy_work_pool.offer([this, controller, request, result, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_heavy_work_pool.offer([this, request, result, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -247,15 +292,18 @@ void PInternalServiceImpl::tablet_writer_cancel(google::protobuf::RpcController*
const PTabletWriterCancelRequest* request,
PTabletWriterCancelResult* response,
google::protobuf::Closure* done) {
- VLOG_RPC << "tablet writer cancel, id=" << request->id() << ", index_id=" << request->index_id()
- << ", sender_id=" << request->sender_id();
- brpc::ClosureGuard closure_guard(done);
- auto st = _exec_env->load_channel_mgr()->cancel(*request);
- if (!st.ok()) {
- LOG(WARNING) << "tablet writer cancel failed, id=" << request->id()
- << ", index_id=" << request->index_id()
- << ", sender_id=" << request->sender_id();
- }
+ DorisMetrics::instance()->tablet_writer_cancel->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
- _exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
- }
- Status::OK().to_protobuf(response->mutable_status());
+ Status::OK().to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController* controller,
const PUpdateCacheRequest* request,
PCacheResponse* response, google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ DorisMetrics::instance()->update_cache->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -286,125 +334,134 @@
}
}
-void PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::cancel_plan_fragment(google::protobuf::RpcController* controller,
const PCancelPlanFragmentRequest* request,
PCancelPlanFragmentResult* result,
google::protobuf::Closure* done) {
- auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start", cntl_base);
- auto scope = OpentelemetryScope {span};
- brpc::ClosureGuard closure_guard(done);
- TUniqueId tid;
- tid.__set_hi(request->finst_id().hi());
- tid.__set_lo(request->finst_id().lo());
-
- Status st = Status::OK();
- if (request->has_cancel_reason()) {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
- << ", reason: " << request->cancel_reason();
- _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
- } else {
- LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
- _exec_env->fragment_mgr()->cancel(tid);
- }
-
- // TODO: the logic seems useless, cancel only return Status::OK. remove it
- st.to_protobuf(result->mutable_status());
+ DorisMetrics::instance()->cancel_plan_fragment->increment(1);
+ _light_work_pool.offer([this, controller, request, result, done]() {
+ auto span = telemetry::start_rpc_server_span("exec_plan_fragment_start", controller);
+ auto scope = OpentelemetryScope {span};
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId tid;
+ tid.__set_hi(request->finst_id().hi());
+ tid.__set_lo(request->finst_id().lo());
+
+ Status st = Status::OK();
+ if (request->has_cancel_reason()) {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid)
+ << ", reason: " << request->cancel_reason();
+ _exec_env->fragment_mgr()->cancel(tid, request->cancel_reason());
+ } else {
+ LOG(INFO) << "cancel fragment, fragment_instance_id=" << print_id(tid);
+ _exec_env->fragment_mgr()->cancel(tid);
+ }
+ // TODO: the logic seems useless, cancel only return Status::OK. remove it
+ st.to_protobuf(result->mutable_status());
+ });
}
-void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::fetch_data(google::protobuf::RpcController* controller,
const PFetchDataRequest* request, PFetchDataResult* result,
google::protobuf::Closure* done) {
- brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
- GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
- _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ DorisMetrics::instance()->fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, result, done]() {
+ brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ GetResultBatchCtx* ctx = new GetResultBatchCtx(cntl, result, done);
+ _exec_env->result_mgr()->fetch_data(request->finst_id(), ctx);
+ });
}
void PInternalServiceImpl::fetch_table_schema(google::protobuf::RpcController* controller,
const PFetchTableSchemaRequest* request,
PFetchTableSchemaResult* result,
google::protobuf::Closure* done) {
- VLOG_RPC << "fetch table schema";
- brpc::ClosureGuard closure_guard(done);
- TFileScanRange file_scan_range;
- Status st = Status::OK();
- {
- const uint8_t* buf = (const uint8_t*)(request->file_scan_range().data());
- uint32_t len = request->file_scan_range().size();
- st = deserialize_thrift_msg(buf, &len, false, &file_scan_range);
+ DorisMetrics::instance()->fetch_table_schema->increment(1);
+ _light_work_pool.offer([this, controller, request, result, done]() {
Review Comment:
warning: lambda capture 'this' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([ controller, request, result, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -693,75 +790,86 @@
const PCheckRPCChannelRequest* request,
PCheckRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->data().size() != request->size()) {
- std::stringstream ss;
- ss << "data size not same, expected: " << request->size()
- << ", actual: " << request->data().size();
- response->mutable_status()->add_error_msgs(ss.str());
- response->mutable_status()->set_status_code(1);
-
- } else {
- Md5Digest digest;
- digest.update(static_cast<const void*>(request->data().c_str()), request->data().size());
- digest.digest();
- if (!iequal(digest.hex(), request->md5())) {
+ DorisMetrics::instance()->check_rpc_channel->increment(1);
+ _light_work_pool.offer([controller, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([ request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -693,75 +790,86 @@
const PCheckRPCChannelRequest* request,
PCheckRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->data().size() != request->size()) {
- std::stringstream ss;
- ss << "data size not same, expected: " << request->size()
- << ", actual: " << request->data().size();
- response->mutable_status()->add_error_msgs(ss.str());
- response->mutable_status()->set_status_code(1);
-
- } else {
- Md5Digest digest;
- digest.update(static_cast<const void*>(request->data().c_str()), request->data().size());
- digest.digest();
- if (!iequal(digest.hex(), request->md5())) {
+ DorisMetrics::instance()->check_rpc_channel->increment(1);
+ _light_work_pool.offer([controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(0);
+ if (request->data().size() != request->size()) {
std::stringstream ss;
- ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
+ ss << "data size not same, expected: " << request->size()
+ << ", actual: " << request->data().size();
response->mutable_status()->add_error_msgs(ss.str());
response->mutable_status()->set_status_code(1);
+
+ } else {
+ Md5Digest digest;
+ digest.update(static_cast<const void*>(request->data().c_str()),
+ request->data().size());
+ digest.digest();
+ if (!iequal(digest.hex(), request->md5())) {
+ std::stringstream ss;
+ ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
+ response->mutable_status()->add_error_msgs(ss.str());
+ response->mutable_status()->set_status_code(1);
+ }
}
- }
+ });
}
void PInternalServiceImpl::reset_rpc_channel(google::protobuf::RpcController* controller,
const PResetRPCChannelRequest* request,
PResetRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->all()) {
- int size = ExecEnv::GetInstance()->brpc_internal_client_cache()->size();
- if (size > 0) {
- std::vector<std::string> endpoints;
- ExecEnv::GetInstance()->brpc_internal_client_cache()->get_all(&endpoints);
- ExecEnv::GetInstance()->brpc_internal_client_cache()->clear();
- *response->mutable_channels() = {endpoints.begin(), endpoints.end()};
- }
- } else {
- for (const std::string& endpoint : request->endpoints()) {
- if (!ExecEnv::GetInstance()->brpc_internal_client_cache()->exist(endpoint)) {
- response->mutable_status()->add_error_msgs(endpoint + ": not found.");
- continue;
+ DorisMetrics::instance()->reset_rpc_channel->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
warning: lambda capture 'controller' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([this, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -693,75 +790,86 @@
const PCheckRPCChannelRequest* request,
PCheckRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->data().size() != request->size()) {
- std::stringstream ss;
- ss << "data size not same, expected: " << request->size()
- << ", actual: " << request->data().size();
- response->mutable_status()->add_error_msgs(ss.str());
- response->mutable_status()->set_status_code(1);
-
- } else {
- Md5Digest digest;
- digest.update(static_cast<const void*>(request->data().c_str()), request->data().size());
- digest.digest();
- if (!iequal(digest.hex(), request->md5())) {
+ DorisMetrics::instance()->check_rpc_channel->increment(1);
+ _light_work_pool.offer([controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(0);
+ if (request->data().size() != request->size()) {
std::stringstream ss;
- ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
+ ss << "data size not same, expected: " << request->size()
+ << ", actual: " << request->data().size();
response->mutable_status()->add_error_msgs(ss.str());
response->mutable_status()->set_status_code(1);
+
+ } else {
+ Md5Digest digest;
+ digest.update(static_cast<const void*>(request->data().c_str()),
+ request->data().size());
+ digest.digest();
+ if (!iequal(digest.hex(), request->md5())) {
+ std::stringstream ss;
+ ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
+ response->mutable_status()->add_error_msgs(ss.str());
+ response->mutable_status()->set_status_code(1);
+ }
}
- }
+ });
}
void PInternalServiceImpl::reset_rpc_channel(google::protobuf::RpcController* controller,
const PResetRPCChannelRequest* request,
PResetRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->all()) {
- int size = ExecEnv::GetInstance()->brpc_internal_client_cache()->size();
- if (size > 0) {
- std::vector<std::string> endpoints;
- ExecEnv::GetInstance()->brpc_internal_client_cache()->get_all(&endpoints);
- ExecEnv::GetInstance()->brpc_internal_client_cache()->clear();
- *response->mutable_channels() = {endpoints.begin(), endpoints.end()};
- }
- } else {
- for (const std::string& endpoint : request->endpoints()) {
- if (!ExecEnv::GetInstance()->brpc_internal_client_cache()->exist(endpoint)) {
- response->mutable_status()->add_error_msgs(endpoint + ": not found.");
- continue;
+ DorisMetrics::instance()->reset_rpc_channel->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
warning: lambda capture 'this' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([ controller, request, response, done]() {
```
##########
be/src/service/internal_service.cpp:
##########
@@ -693,75 +790,86 @@
const PCheckRPCChannelRequest* request,
PCheckRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->data().size() != request->size()) {
- std::stringstream ss;
- ss << "data size not same, expected: " << request->size()
- << ", actual: " << request->data().size();
- response->mutable_status()->add_error_msgs(ss.str());
- response->mutable_status()->set_status_code(1);
-
- } else {
- Md5Digest digest;
- digest.update(static_cast<const void*>(request->data().c_str()), request->data().size());
- digest.digest();
- if (!iequal(digest.hex(), request->md5())) {
+ DorisMetrics::instance()->check_rpc_channel->increment(1);
+ _light_work_pool.offer([controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(0);
+ if (request->data().size() != request->size()) {
std::stringstream ss;
- ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
+ ss << "data size not same, expected: " << request->size()
+ << ", actual: " << request->data().size();
response->mutable_status()->add_error_msgs(ss.str());
response->mutable_status()->set_status_code(1);
+
+ } else {
+ Md5Digest digest;
+ digest.update(static_cast<const void*>(request->data().c_str()),
+ request->data().size());
+ digest.digest();
+ if (!iequal(digest.hex(), request->md5())) {
+ std::stringstream ss;
+ ss << "md5 not same, expected: " << request->md5() << ", actual: " << digest.hex();
+ response->mutable_status()->add_error_msgs(ss.str());
+ response->mutable_status()->set_status_code(1);
+ }
}
- }
+ });
}
void PInternalServiceImpl::reset_rpc_channel(google::protobuf::RpcController* controller,
const PResetRPCChannelRequest* request,
PResetRPCChannelResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- response->mutable_status()->set_status_code(0);
- if (request->all()) {
- int size = ExecEnv::GetInstance()->brpc_internal_client_cache()->size();
- if (size > 0) {
- std::vector<std::string> endpoints;
- ExecEnv::GetInstance()->brpc_internal_client_cache()->get_all(&endpoints);
- ExecEnv::GetInstance()->brpc_internal_client_cache()->clear();
- *response->mutable_channels() = {endpoints.begin(), endpoints.end()};
- }
- } else {
- for (const std::string& endpoint : request->endpoints()) {
- if (!ExecEnv::GetInstance()->brpc_internal_client_cache()->exist(endpoint)) {
- response->mutable_status()->add_error_msgs(endpoint + ": not found.");
- continue;
+ DorisMetrics::instance()->reset_rpc_channel->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ response->mutable_status()->set_status_code(0);
+ if (request->all()) {
+ int size = ExecEnv::GetInstance()->brpc_internal_client_cache()->size();
+ if (size > 0) {
+ std::vector<std::string> endpoints;
+ ExecEnv::GetInstance()->brpc_internal_client_cache()->get_all(&endpoints);
+ ExecEnv::GetInstance()->brpc_internal_client_cache()->clear();
+ *response->mutable_channels() = {endpoints.begin(), endpoints.end()};
}
+ } else {
+ for (const std::string& endpoint : request->endpoints()) {
+ if (!ExecEnv::GetInstance()->brpc_internal_client_cache()->exist(endpoint)) {
+ response->mutable_status()->add_error_msgs(endpoint + ": not found.");
+ continue;
+ }
- if (ExecEnv::GetInstance()->brpc_internal_client_cache()->erase(endpoint)) {
- response->add_channels(endpoint);
- } else {
- response->mutable_status()->add_error_msgs(endpoint + ": reset failed.");
+ if (ExecEnv::GetInstance()->brpc_internal_client_cache()->erase(endpoint)) {
+ response->add_channels(endpoint);
+ } else {
+ response->mutable_status()->add_error_msgs(endpoint + ": reset failed.");
+ }
+ }
+ if (request->endpoints_size() != response->channels_size()) {
+ response->mutable_status()->set_status_code(1);
}
}
- if (request->endpoints_size() != response->channels_size()) {
- response->mutable_status()->set_status_code(1);
- }
- }
+ });
}
-void PInternalServiceImpl::hand_shake(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::hand_shake(google::protobuf::RpcController* controller,
const PHandShakeRequest* request,
PHandShakeResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- if (request->has_hello()) {
- response->set_hello(request->hello());
- }
- response->mutable_status()->set_status_code(0);
+ DorisMetrics::instance()->hand_shake->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
Review Comment:
warning: lambda capture 'this' is not used [clang-diagnostic-unused-lambda-capture]
```suggestion
_light_work_pool.offer([ request, response, done]() {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1434112667
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] hello-stephen commented on pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1426814356
TeamCity pipeline, clickbench performance test result:
the sum of best hot time: 34.68 seconds
stream load tsv: 469 seconds loaded 74807831229 Bytes, about 152 MB/s
stream load json: 37 seconds loaded 2358488459 Bytes, about 60 MB/s
stream load orc: 68 seconds loaded 1101869774 Bytes, about 15 MB/s
stream load parquet: 28 seconds loaded 861443392 Bytes, about 29 MB/s
https://doris-community-test-1308700295.cos.ap-hongkong.myqcloud.com/tmp/20230211162649_clickbench_pr_94553.html
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1437984734
run p0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1438576019
PR approved by at least one committer and no changes requested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1438576096
PR approved by anyone and no changes requested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1438312796
run p0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1434012680
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1437745682
run p0
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on a diff in pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103956391
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
before this pr ,this rpc interface also mixed with each other
i don't think put into thread pool will effect latency
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1428119686
> Please also update this document: https://doris.apache.org/zh-CN/docs/dev/admin-manual/maint-monitor/monitor-metrics/metrics
resolved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] github-actions[bot] commented on pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #16639:
URL: https://github.com/apache/doris/pull/16639#issuecomment-1429064485
clang-tidy review says "All clean, LGTM! :+1:"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] eldenmoon commented on a diff in pull request #16639: [Improvement](brpc) Using a thread pool for RPC service avoiding std::mutex block brpc::bthread
Posted by "eldenmoon (via GitHub)" <gi...@apache.org>.
eldenmoon commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103961000
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
Review Comment:
if the thread pool in hight CPU work load, this interface will be queued up and the latency is infected
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[GitHub] [doris] chenlinzhong commented on a diff in pull request #16639: [Improvement](brpc) replace brpc bthread with pthread
Posted by "chenlinzhong (via GitHub)" <gi...@apache.org>.
chenlinzhong commented on code in PR #16639:
URL: https://github.com/apache/doris/pull/16639#discussion_r1103938276
##########
be/src/service/internal_service.cpp:
##########
@@ -1097,9 +1208,8 @@ void PInternalServiceImpl::multiget_data(google::protobuf::RpcController* contro
const PMultiGetRequest* request,
PMultiGetResponse* response,
google::protobuf::Closure* done) {
- // Submit task to seperate ThreadPool for avoiding block bthread working pthread
- ThreadPool* task_pool = StorageEngine::instance()->get_bg_multiget_threadpool();
- Status submit_st = task_pool->submit_func([request, response, done, this]() {
+ DorisMetrics::instance()->multiget_data->increment(1);
+ _light_work_pool.offer([request, response, done, this]() {
Review Comment:
ok
##########
be/src/service/internal_service.cpp:
##########
@@ -423,200 +480,233 @@ void PInternalServiceImpl::tablet_fetch_data(google::protobuf::RpcController* co
const PTabletKeyLookupRequest* request,
PTabletKeyLookupResponse* response,
google::protobuf::Closure* done) {
- [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- brpc::ClosureGuard guard(done);
- Status st = _tablet_fetch_data(request, response);
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->tablet_fetch_data->increment(1);
+ _heavy_work_pool.offer([this, controller, request, response, done]() {
+ [[maybe_unused]] brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
+ brpc::ClosureGuard guard(done);
+ Status st = _tablet_fetch_data(request, response);
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::get_info(google::protobuf::RpcController* controller,
const PProxyRequest* request, PProxyResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- // PProxyRequest is defined in gensrc/proto/internal_service.proto
- // Currently it supports 2 kinds of requests:
- // 1. get all kafka partition ids for given topic
- // 2. get all kafka partition offsets for given topic and timestamp.
- if (request->has_kafka_meta_request()) {
- const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
- if (!kafka_request.partition_id_for_latest_offsets().empty()) {
- // get latest offsets for specified partition ids
- std::vector<PIntegerPair> partition_offsets;
- Status st = _exec_env->routine_load_task_executor()
- ->get_kafka_latest_offsets_for_partitions(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ DorisMetrics::instance()->get_info->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ // PProxyRequest is defined in gensrc/proto/internal_service.proto
+ // Currently it supports 2 kinds of requests:
+ // 1. get all kafka partition ids for given topic
+ // 2. get all kafka partition offsets for given topic and timestamp.
+ if (request->has_kafka_meta_request()) {
+ const PKafkaMetaProxyRequest& kafka_request = request->kafka_meta_request();
+ if (!kafka_request.partition_id_for_latest_offsets().empty()) {
+ // get latest offsets for specified partition ids
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_latest_offsets_for_partitions(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else if (!kafka_request.offset_times().empty()) {
- // if offset_times() has elements, which means this request is to get offset by timestamp.
- std::vector<PIntegerPair> partition_offsets;
- Status st =
- _exec_env->routine_load_task_executor()->get_kafka_partition_offsets_for_times(
- request->kafka_meta_request(), &partition_offsets);
- if (st.ok()) {
- PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
- for (const auto& entry : partition_offsets) {
- PIntegerPair* res = part_offsets->add_offset_times();
- res->set_key(entry.key());
- res->set_val(entry.val());
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else if (!kafka_request.offset_times().empty()) {
+ // if offset_times() has elements, which means this request is to get offset by timestamp.
+ std::vector<PIntegerPair> partition_offsets;
+ Status st = _exec_env->routine_load_task_executor()
+ ->get_kafka_partition_offsets_for_times(
+ request->kafka_meta_request(), &partition_offsets);
+ if (st.ok()) {
+ PKafkaPartitionOffsets* part_offsets = response->mutable_partition_offsets();
+ for (const auto& entry : partition_offsets) {
+ PIntegerPair* res = part_offsets->add_offset_times();
+ res->set_key(entry.key());
+ res->set_val(entry.val());
+ }
}
- }
- st.to_protobuf(response->mutable_status());
- return;
- } else {
- // get partition ids of topic
- std::vector<int32_t> partition_ids;
- Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
- request->kafka_meta_request(), &partition_ids);
- if (st.ok()) {
- PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
- for (int32_t id : partition_ids) {
- kafka_result->add_partition_ids(id);
+ st.to_protobuf(response->mutable_status());
+ return;
+ } else {
+ // get partition ids of topic
+ std::vector<int32_t> partition_ids;
+ Status st = _exec_env->routine_load_task_executor()->get_kafka_partition_meta(
+ request->kafka_meta_request(), &partition_ids);
+ if (st.ok()) {
+ PKafkaMetaProxyResult* kafka_result = response->mutable_kafka_meta_result();
+ for (int32_t id : partition_ids) {
+ kafka_result->add_partition_ids(id);
+ }
}
+ st.to_protobuf(response->mutable_status());
+ return;
}
- st.to_protobuf(response->mutable_status());
- return;
}
- }
- Status::OK().to_protobuf(response->mutable_status());
+ Status::OK().to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::update_cache(google::protobuf::RpcController* controller,
const PUpdateCacheRequest* request,
PCacheResponse* response, google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->update(request, response);
+ DorisMetrics::instance()->update_cache->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->update(request, response);
+ });
}
void PInternalServiceImpl::fetch_cache(google::protobuf::RpcController* controller,
const PFetchCacheRequest* request, PFetchCacheResult* result,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->fetch(request, result);
+ DorisMetrics::instance()->fetch_cache->increment(1);
+ _heavy_work_pool.offer([this, request, result, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->fetch(request, result);
+ });
}
void PInternalServiceImpl::clear_cache(google::protobuf::RpcController* controller,
const PClearCacheRequest* request, PCacheResponse* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- _exec_env->result_cache()->clear(request, response);
+ DorisMetrics::instance()->clear_cache->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ _exec_env->result_cache()->clear(request, response);
+ });
}
void PInternalServiceImpl::merge_filter(::google::protobuf::RpcController* controller,
const ::doris::PMergeFilterRequest* request,
::doris::PMergeFilterResponse* response,
::google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- auto attachment = static_cast<brpc::Controller*>(controller)->request_attachment();
- butil::IOBufAsZeroCopyInputStream zero_copy_input_stream(attachment);
- Status st = _exec_env->fragment_mgr()->merge_filter(request, &zero_copy_input_stream);
- if (!st.ok()) {
- LOG(WARNING) << "merge meet error" << st.to_string();
- }
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->merge_filter->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ auto attachment = static_cast<brpc::Controller*>(controller)->request_attachment();
+ butil::IOBufAsZeroCopyInputStream zero_copy_input_stream(attachment);
+ Status st = _exec_env->fragment_mgr()->merge_filter(request, &zero_copy_input_stream);
+ if (!st.ok()) {
+ LOG(WARNING) << "merge meet error" << st.to_string();
+ }
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::apply_filter(::google::protobuf::RpcController* controller,
const ::doris::PPublishFilterRequest* request,
::doris::PPublishFilterResponse* response,
::google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- auto attachment = static_cast<brpc::Controller*>(controller)->request_attachment();
- butil::IOBufAsZeroCopyInputStream zero_copy_input_stream(attachment);
- UniqueId unique_id(request->query_id());
- VLOG_NOTICE << "rpc apply_filter recv";
- Status st = _exec_env->fragment_mgr()->apply_filter(request, &zero_copy_input_stream);
- if (!st.ok()) {
- LOG(WARNING) << "apply filter meet error: " << st.to_string();
- }
- st.to_protobuf(response->mutable_status());
+ DorisMetrics::instance()->apply_filter->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ auto attachment = static_cast<brpc::Controller*>(controller)->request_attachment();
+ butil::IOBufAsZeroCopyInputStream zero_copy_input_stream(attachment);
+ UniqueId unique_id(request->query_id());
+ VLOG_NOTICE << "rpc apply_filter recv";
+ Status st = _exec_env->fragment_mgr()->apply_filter(request, &zero_copy_input_stream);
+ if (!st.ok()) {
+ LOG(WARNING) << "apply filter meet error: " << st.to_string();
+ }
+ st.to_protobuf(response->mutable_status());
+ });
}
void PInternalServiceImpl::send_data(google::protobuf::RpcController* controller,
const PSendDataRequest* request, PSendDataResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- TUniqueId fragment_instance_id;
- fragment_instance_id.hi = request->fragment_instance_id().hi();
- fragment_instance_id.lo = request->fragment_instance_id().lo();
-
- auto pipe = _exec_env->fragment_mgr()->get_pipe(fragment_instance_id);
- if (pipe == nullptr) {
- response->mutable_status()->set_status_code(1);
- response->mutable_status()->add_error_msgs("pipe is null");
- } else {
- for (int i = 0; i < request->data_size(); ++i) {
- PDataRow* row = new PDataRow();
- row->CopyFrom(request->data(i));
- pipe->append_and_flush(reinterpret_cast<char*>(&row), sizeof(row),
- sizeof(row) + row->ByteSizeLong());
+ DorisMetrics::instance()->send_data->increment(1);
+ _heavy_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId fragment_instance_id;
+ fragment_instance_id.hi = request->fragment_instance_id().hi();
+ fragment_instance_id.lo = request->fragment_instance_id().lo();
+
+ auto pipe = _exec_env->fragment_mgr()->get_pipe(fragment_instance_id);
+ if (pipe == nullptr) {
+ response->mutable_status()->set_status_code(1);
+ response->mutable_status()->add_error_msgs("pipe is null");
+ } else {
+ for (int i = 0; i < request->data_size(); ++i) {
+ PDataRow* row = new PDataRow();
+ row->CopyFrom(request->data(i));
+ pipe->append_and_flush(reinterpret_cast<char*>(&row), sizeof(row),
+ sizeof(row) + row->ByteSizeLong());
+ }
+ response->mutable_status()->set_status_code(0);
}
- response->mutable_status()->set_status_code(0);
- }
+ });
}
void PInternalServiceImpl::commit(google::protobuf::RpcController* controller,
const PCommitRequest* request, PCommitResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- TUniqueId fragment_instance_id;
- fragment_instance_id.hi = request->fragment_instance_id().hi();
- fragment_instance_id.lo = request->fragment_instance_id().lo();
-
- auto pipe = _exec_env->fragment_mgr()->get_pipe(fragment_instance_id);
- if (pipe == nullptr) {
- response->mutable_status()->set_status_code(1);
- response->mutable_status()->add_error_msgs("pipe is null");
- } else {
- pipe->finish();
- response->mutable_status()->set_status_code(0);
- }
+ DorisMetrics::instance()->commit->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId fragment_instance_id;
+ fragment_instance_id.hi = request->fragment_instance_id().hi();
+ fragment_instance_id.lo = request->fragment_instance_id().lo();
+
+ auto pipe = _exec_env->fragment_mgr()->get_pipe(fragment_instance_id);
+ if (pipe == nullptr) {
+ response->mutable_status()->set_status_code(1);
+ response->mutable_status()->add_error_msgs("pipe is null");
+ } else {
+ pipe->finish();
+ response->mutable_status()->set_status_code(0);
+ }
+ });
}
void PInternalServiceImpl::rollback(google::protobuf::RpcController* controller,
const PRollbackRequest* request, PRollbackResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- TUniqueId fragment_instance_id;
- fragment_instance_id.hi = request->fragment_instance_id().hi();
- fragment_instance_id.lo = request->fragment_instance_id().lo();
-
- auto pipe = _exec_env->fragment_mgr()->get_pipe(fragment_instance_id);
- if (pipe == nullptr) {
- response->mutable_status()->set_status_code(1);
- response->mutable_status()->add_error_msgs("pipe is null");
- } else {
- pipe->cancel("rollback");
- response->mutable_status()->set_status_code(0);
- }
+ DorisMetrics::instance()->rollback->increment(1);
+ _light_work_pool.offer([this, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ TUniqueId fragment_instance_id;
+ fragment_instance_id.hi = request->fragment_instance_id().hi();
+ fragment_instance_id.lo = request->fragment_instance_id().lo();
+
+ auto pipe = _exec_env->fragment_mgr()->get_pipe(fragment_instance_id);
+ if (pipe == nullptr) {
+ response->mutable_status()->set_status_code(1);
+ response->mutable_status()->add_error_msgs("pipe is null");
+ } else {
+ pipe->cancel("rollback");
+ response->mutable_status()->set_status_code(0);
+ }
+ });
}
-void PInternalServiceImpl::fold_constant_expr(google::protobuf::RpcController* cntl_base,
+void PInternalServiceImpl::fold_constant_expr(google::protobuf::RpcController* controller,
const PConstantExprRequest* request,
PConstantExprResult* response,
google::protobuf::Closure* done) {
- brpc::ClosureGuard closure_guard(done);
- brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
+ DorisMetrics::instance()->fold_constant_expr->increment(1);
+ _light_work_pool.offer([this, controller, request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ brpc::Controller* cntl = static_cast<brpc::Controller*>(controller);
- Status st = Status::OK();
- if (request->has_request()) {
- st = _fold_constant_expr(request->request(), response);
- } else {
- // TODO(yangzhengguo) this is just for compatible with old version, this should be removed in the release 0.15
- st = _fold_constant_expr(cntl->request_attachment().to_string(), response);
- }
- if (!st.ok()) {
- LOG(WARNING) << "exec fold constant expr failed, errmsg=" << st;
- }
- st.to_protobuf(response->mutable_status());
+ Status st = Status::OK();
+ if (request->has_request()) {
+ st = _fold_constant_expr(request->request(), response);
+ } else {
+ // TODO(yangzhengguo) this is just for compatible with old version, this should be removed in the release 0.15
Review Comment:
ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org