Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[feat](metrics) Unify metrics of thread pool #43144

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
+415 −166
Open

[feat](metrics) Unify metrics of thread pool #43144

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
d7e5f63
NED
zhiqiang-hhhh Dec 24, 2024
3d0859e
FMT
zhiqiang-hhhh Dec 24, 2024
7002be2
FIX STATIC
zhiqiang-hhhh Dec 25, 2024
27406d5
NED
zhiqiang-hhhh Dec 25, 2024
102b19c
TEST
zhiqiang-hhhh Dec 25, 2024
ad26051
COM
zhiqiang-hhhh Dec 25, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 0 additions & 16 deletions be/src/olap/memtable_flush_executor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,9 +37,6 @@
namespace doris {
using namespace ErrorCode;

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(flush_thread_pool_queue_size, MetricUnit::NOUNIT);
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(flush_thread_pool_thread_num, MetricUnit::NOUNIT);

bvar::Adder<int64_t> g_flush_task_num("memtable_flush_task_num");

class MemtableFlushTask final : public Runnable {
Expand Down Expand Up @@ -239,7 +236,6 @@ void MemTableFlushExecutor::init(int num_disk) {
.set_min_threads(min_threads)
.set_max_threads(max_threads)
.build(&_high_prio_flush_pool));
_register_metrics();
}

// NOTE: we use SERIAL mode here to ensure all mem-tables from one tablet are flushed in order.
Expand All @@ -263,16 +259,4 @@ Status MemTableFlushExecutor::create_flush_token(std::shared_ptr<FlushToken>& fl
}
}

void MemTableFlushExecutor::_register_metrics() {
REGISTER_HOOK_METRIC(flush_thread_pool_queue_size,
[this]() { return _flush_pool->get_queue_size(); });
REGISTER_HOOK_METRIC(flush_thread_pool_thread_num,
[this]() { return _flush_pool->num_threads(); })
}

void MemTableFlushExecutor::_deregister_metrics() {
DEREGISTER_HOOK_METRIC(flush_thread_pool_queue_size);
DEREGISTER_HOOK_METRIC(flush_thread_pool_thread_num);
}

} // namespace doris
4 changes: 0 additions & 4 deletions be/src/olap/memtable_flush_executor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,6 @@ class MemTableFlushExecutor {
public:
MemTableFlushExecutor() = default;
~MemTableFlushExecutor() {
_deregister_metrics();
_flush_pool->shutdown();
_high_prio_flush_pool->shutdown();
}
Expand All @@ -141,9 +140,6 @@ class MemTableFlushExecutor {
std::shared_ptr<WorkloadGroup> wg_sptr);

private:
void _register_metrics();
static void _deregister_metrics();

std::unique_ptr<ThreadPool> _flush_pool;
std::unique_ptr<ThreadPool> _high_prio_flush_pool;
};
Expand Down
4 changes: 3 additions & 1 deletion be/src/runtime/exec_env.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,9 @@

namespace doris {

ExecEnv::ExecEnv() = default;
ExecEnv::ExecEnv() {
_doris_metrics = std::make_unique<DorisMetrics>();
}

ExecEnv::~ExecEnv() {
destroy();
Expand Down
8 changes: 5 additions & 3 deletions be/src/runtime/exec_env.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@ class MemoryPool;
}

namespace doris {
class DorisMetrics;
namespace vectorized {
class VDataStreamMgr;
class ScannerScheduler;
Expand Down Expand Up @@ -349,6 +350,8 @@ class ExecEnv {

bool check_auth_token(const std::string& auth_token);

DorisMetrics* doris_metrics() { return _doris_metrics.get(); }

private:
ExecEnv();

Expand All @@ -360,9 +363,6 @@ class ExecEnv {
Status _init_mem_env();
Status _check_deploy_mode();

void _register_metrics();
void _deregister_metrics();

inline static std::atomic_bool _s_ready {false};
inline static std::atomic_bool _s_tracking_memory {false};
std::vector<StorePath> _store_paths;
Expand Down Expand Up @@ -488,6 +488,8 @@ class ExecEnv {

orc::MemoryPool* _orc_memory_pool = nullptr;
arrow::MemoryPool* _arrow_memory_pool = nullptr;

std::unique_ptr<DorisMetrics> _doris_metrics;
};

template <>
Expand Down
23 changes: 3 additions & 20 deletions be/src/runtime/exec_env_init.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -139,10 +139,6 @@ namespace doris {
class PBackendService_Stub;
class PFunctionService_Stub;

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(scanner_thread_pool_queue_size, MetricUnit::NOUNIT);
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(send_batch_thread_pool_thread_num, MetricUnit::NOUNIT);
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(send_batch_thread_pool_queue_size, MetricUnit::NOUNIT);

static void init_doris_metrics(const std::vector<StorePath>& store_paths) {
bool init_system_metrics = config::enable_system_metrics;
std::set<std::string> disk_devices;
Expand Down Expand Up @@ -336,7 +332,6 @@ Status ExecEnv::_init(const std::vector<StorePath>& store_paths,
RETURN_IF_ERROR(_load_channel_mgr->init(MemInfo::mem_limit()));
RETURN_IF_ERROR(_wal_manager->init());
_heartbeat_flags = new HeartbeatFlags();
_register_metrics();

_tablet_schema_cache =
TabletSchemaCache::create_global_schema_cache(config::tablet_schema_cache_capacity);
Expand Down Expand Up @@ -668,20 +663,6 @@ Status ExecEnv::_check_deploy_mode() {
return Status::OK();
}

void ExecEnv::_register_metrics() {
REGISTER_HOOK_METRIC(send_batch_thread_pool_thread_num,
[this]() { return _send_batch_thread_pool->num_threads(); });

REGISTER_HOOK_METRIC(send_batch_thread_pool_queue_size,
[this]() { return _send_batch_thread_pool->get_queue_size(); });
}

void ExecEnv::_deregister_metrics() {
DEREGISTER_HOOK_METRIC(scanner_thread_pool_queue_size);
DEREGISTER_HOOK_METRIC(send_batch_thread_pool_thread_num);
DEREGISTER_HOOK_METRIC(send_batch_thread_pool_queue_size);
}

// TODO(zhiqiang): Need refactor all thread pool. Each thread pool must have a Stop method.
// We need to stop all threads before releasing resource.
void ExecEnv::destroy() {
Expand Down Expand Up @@ -736,7 +717,6 @@ void ExecEnv::destroy() {
SAFE_SHUTDOWN(_s3_file_system_thread_pool);
SAFE_SHUTDOWN(_send_batch_thread_pool);

_deregister_metrics();
SAFE_DELETE(_load_channel_mgr);

SAFE_DELETE(_spill_stream_mgr);
Expand Down Expand Up @@ -825,6 +805,9 @@ void ExecEnv::destroy() {
SAFE_DELETE(_process_profile);
SAFE_DELETE(_heap_profiler);

// DorisMetrics should be destoried after all other thread stopped.
_doris_metrics.reset(nullptr);

_s_tracking_memory = false;

LOG(INFO) << "Doris exec envorinment is destoried.";
Expand Down
10 changes: 1 addition & 9 deletions be/src/runtime/fragment_mgr.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,8 +97,7 @@ namespace doris {

DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(fragment_instance_count, MetricUnit::NOUNIT);
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(timeout_canceled_fragment_count, MetricUnit::NOUNIT);
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(fragment_thread_pool_queue_size, MetricUnit::NOUNIT);
DEFINE_GAUGE_METRIC_PROTOTYPE_2ARG(fragment_thread_pool_num_active_threads, MetricUnit::NOUNIT);

bvar::LatencyRecorder g_fragmentmgr_prepare_latency("doris_FragmentMgr", "prepare");

bvar::Adder<uint64_t> g_fragment_executing_count("fragment_executing_count");
Expand Down Expand Up @@ -243,20 +242,13 @@ FragmentMgr::FragmentMgr(ExecEnv* exec_env)
.set_max_threads(config::fragment_mgr_asynic_work_pool_thread_num_max)
.set_max_queue_size(config::fragment_mgr_asynic_work_pool_queue_size)
.build(&_thread_pool);

REGISTER_HOOK_METRIC(fragment_thread_pool_queue_size,
[this]() { return _thread_pool->get_queue_size(); });
REGISTER_HOOK_METRIC(fragment_thread_pool_num_active_threads,
[this]() { return _thread_pool->num_active_threads(); });
CHECK(s.ok()) << s.to_string();
}

FragmentMgr::~FragmentMgr() = default;

void FragmentMgr::stop() {
DEREGISTER_HOOK_METRIC(fragment_instance_count);
DEREGISTER_HOOK_METRIC(fragment_thread_pool_queue_size);
DEREGISTER_HOOK_METRIC(fragment_thread_pool_num_active_threads);
_stop_background_threads_latch.count_down();
if (_cancel_thread) {
_cancel_thread->join();
Expand Down
4 changes: 2 additions & 2 deletions be/src/runtime/workload_group/workload_group.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -487,7 +487,7 @@ void WorkloadGroup::upsert_thread_pool_no_lock(WorkloadGroupInfo* wg_info,
if (_scan_task_sched == nullptr) {
std::unique_ptr<vectorized::SimplifiedScanScheduler> scan_scheduler =
std::make_unique<vectorized::SimplifiedScanScheduler>("Scan_" + wg_name,
cg_cpu_ctl_ptr);
cg_cpu_ctl_ptr, wg_name);
Status ret = scan_scheduler->start(config::doris_scanner_thread_pool_thread_num,
config::doris_scanner_thread_pool_thread_num,
config::doris_scanner_thread_pool_queue_size);
Expand All @@ -507,7 +507,7 @@ void WorkloadGroup::upsert_thread_pool_no_lock(WorkloadGroupInfo* wg_info,
vectorized::ScannerScheduler::get_remote_scan_thread_queue_size();
std::unique_ptr<vectorized::SimplifiedScanScheduler> remote_scan_scheduler =
std::make_unique<vectorized::SimplifiedScanScheduler>("RScan_" + wg_name,
cg_cpu_ctl_ptr);
cg_cpu_ctl_ptr, wg_name);
Status ret = remote_scan_scheduler->start(remote_max_thread_num,
config::doris_scanner_min_thread_pool_thread_num,
remote_scan_thread_queue_size);
Expand Down
6 changes: 5 additions & 1 deletion be/src/service/doris_main.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -425,6 +425,11 @@ int main(int argc, char** argv) {
<< doris::config::spill_storage_root_path;
exit(-1);
}

// ExecEnv must be created before calleing doris::check_datapath_rw
// because will create metrics, it need an initialized DorisMetrics.
doris::ExecEnv* exec_env(doris::ExecEnv::GetInstance());

std::set<std::string> broken_paths;
doris::parse_conf_broken_store_paths(doris::config::broken_storage_path, &broken_paths);

Expand Down Expand Up @@ -526,7 +531,6 @@ int main(int argc, char** argv) {
doris::ThreadLocalHandle::create_thread_local_if_not_exits();

// init exec env
auto* exec_env(doris::ExecEnv::GetInstance());
status = doris::ExecEnv::init(doris::ExecEnv::GetInstance(), paths, spill_paths, broken_paths);
if (status != Status::OK()) {
std::cerr << "failed to init doris storage engine, res=" << status;
Expand Down
6 changes: 0 additions & 6 deletions be/src/util/doris_metrics.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,9 +191,6 @@ DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(query_ctx_cnt, MetricUnit::NOUNIT);
DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(scanner_ctx_cnt, MetricUnit::NOUNIT);
DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(scanner_cnt, MetricUnit::NOUNIT);
DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(scanner_task_cnt, MetricUnit::NOUNIT);
DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(scanner_task_queued, MetricUnit::NOUNIT);
DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(scanner_task_running, MetricUnit::NOUNIT);
DEFINE_COUNTER_METRIC_PROTOTYPE_2ARG(scanner_task_submit_failed, MetricUnit::NOUNIT);

const std::string DorisMetrics::_s_registry_name = "doris_be";
const std::string DorisMetrics::_s_hook_name = "doris_metrics";
Expand Down Expand Up @@ -319,9 +316,6 @@ DorisMetrics::DorisMetrics() : _metric_registry(_s_registry_name) {
INT_COUNTER_METRIC_REGISTER(_server_metric_entity, scanner_ctx_cnt);
INT_COUNTER_METRIC_REGISTER(_server_metric_entity, scanner_cnt);
INT_COUNTER_METRIC_REGISTER(_server_metric_entity, scanner_task_cnt);
INT_COUNTER_METRIC_REGISTER(_server_metric_entity, scanner_task_queued);
INT_COUNTER_METRIC_REGISTER(_server_metric_entity, scanner_task_running);
INT_COUNTER_METRIC_REGISTER(_server_metric_entity, scanner_task_submit_failed);
}

void DorisMetrics::initialize(bool init_system_metrics, const std::set<std::string>& disk_devices,
Expand Down
36 changes: 5 additions & 31 deletions be/src/util/doris_metrics.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
#include <string>
#include <vector>

#include "runtime/exec_env.h"
#include "util/jvm_metrics.h"
#include "util/metrics.h"
#include "util/system_metrics.h"
Expand All @@ -47,6 +48,10 @@ namespace doris {

class DorisMetrics {
public:
DorisMetrics();

static DorisMetrics* instance() { return ExecEnv::GetInstance()->doris_metrics(); }

IntCounter* fragment_requests_total = nullptr;
IntCounter* fragment_request_duration_us = nullptr;
IntCounter* query_scan_bytes = nullptr;
Expand Down Expand Up @@ -197,13 +202,6 @@ class DorisMetrics {
UIntGauge* query_cache_sql_total_count = nullptr;
UIntGauge* query_cache_partition_total_count = nullptr;

UIntGauge* scanner_thread_pool_queue_size = nullptr;
UIntGauge* add_batch_task_queue_size = nullptr;
UIntGauge* send_batch_thread_pool_thread_num = nullptr;
UIntGauge* send_batch_thread_pool_queue_size = nullptr;
UIntGauge* fragment_thread_pool_queue_size = nullptr;
UIntGauge* fragment_thread_pool_num_active_threads = nullptr;

// Upload metrics
UIntGauge* upload_total_byte = nullptr;
IntCounter* upload_rowset_count = nullptr;
Expand All @@ -224,18 +222,6 @@ class DorisMetrics {
UIntGauge* arrow_flight_work_pool_max_queue_size = nullptr;
UIntGauge* arrow_flight_work_max_threads = nullptr;

UIntGauge* flush_thread_pool_queue_size = nullptr;
UIntGauge* flush_thread_pool_thread_num = nullptr;

UIntGauge* local_scan_thread_pool_queue_size = nullptr;
UIntGauge* local_scan_thread_pool_thread_num = nullptr;
UIntGauge* remote_scan_thread_pool_queue_size = nullptr;
UIntGauge* remote_scan_thread_pool_thread_num = nullptr;
UIntGauge* limited_scan_thread_pool_queue_size = nullptr;
UIntGauge* limited_scan_thread_pool_thread_num = nullptr;
UIntGauge* group_local_scan_thread_pool_queue_size = nullptr;
UIntGauge* group_local_scan_thread_pool_thread_num = nullptr;

IntCounter* num_io_bytes_read_total = nullptr;
IntCounter* num_io_bytes_read_from_cache = nullptr;
IntCounter* num_io_bytes_read_from_remote = nullptr;
Expand All @@ -244,14 +230,6 @@ class DorisMetrics {
IntCounter* scanner_ctx_cnt = nullptr;
IntCounter* scanner_cnt = nullptr;
IntCounter* scanner_task_cnt = nullptr;
IntCounter* scanner_task_queued = nullptr;
IntCounter* scanner_task_submit_failed = nullptr;
IntCounter* scanner_task_running = nullptr;

static DorisMetrics* instance() {
static DorisMetrics instance;
return &instance;
}

// not thread-safe, call before calling metrics
void initialize(
Expand All @@ -266,14 +244,10 @@ class DorisMetrics {
void init_jvm_metrics(JNIEnv* env);

private:
// Don't allow constructor
DorisMetrics();

void _update();
void _update_process_thread_num();
void _update_process_fd_num();

private:
static const std::string _s_registry_name;
static const std::string _s_hook_name;

Expand Down
Loading
Loading