This is an automated email from the ASF dual-hosted git repository.
wwbmmm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/brpc.git
The following commit(s) were added to refs/heads/master by this push:
new c4f0a924 Add monitor of timer thread task (#3603)
c4f0a924 is described below
commit c4f0a92478b24e4d2104e462e4cba45167250f34
Author: Bright Chen <[email protected]>
AuthorDate: Sun Oct 18 10:10:57 2026 +0800
Add monitor of timer thread task (#3603)
---
.gitignore | 1 +
src/bthread/timer_thread.cpp | 32 +++++++++++++++++---------------
2 files changed, 18 insertions(+), 15 deletions(-)
diff --git a/.gitignore b/.gitignore
index 739963a2..7a9ade11 100644
--- a/.gitignore
+++ b/.gitignore
@@ -53,6 +53,7 @@ graphify-out/
/protoc-gen-mcpack*/
# Ignore bazel files
+MODULE.bazel.lock
bazel-bin
bazel-out
bazel-testlogs
diff --git a/src/bthread/timer_thread.cpp b/src/bthread/timer_thread.cpp
index b0a5b9f0..9b66fe04 100644
--- a/src/bthread/timer_thread.cpp
+++ b/src/bthread/timer_thread.cpp
@@ -18,7 +18,6 @@
// bthread - An M:N threading library to make applications more concurrent.
-#include <queue> // heap functions
#include <gflags/gflags.h>
#include "butil/scoped_lock.h"
#include "butil/logging.h"
@@ -71,24 +70,23 @@ TimerThreadOptions::TimerThreadOptions()
// A task contains the necessary information for running fn(arg).
// Tasks are created in Bucket::schedule and destroyed in TimerThread::run
struct BAIDU_CACHELINE_ALIGNMENT TimerThread::Task {
- Task* next; // For linking tasks in a Bucket.
- int64_t run_time; // run the task at this realtime
- void (*fn)(void*); // the fn(arg) to run
- void* arg;
+ Task* next{nullptr}; // For linking tasks in a Bucket.
+ int64_t run_time{0}; // run the task at this realtime
+ void (*fn)(void*){nullptr}; // the fn(arg) to run
+ void* arg{nullptr};
// Current TaskId, checked against version in TimerThread::run to test
// if this task is unscheduled.
- TaskId task_id;
+ TaskId task_id{0};
// initial_version: not run yet
// initial_version + 1: running
// initial_version + 2: removed (also the version of next Task reused
// this struct)
- butil::atomic<uint32_t> version;
-
- Task() : version(2/*skip 0*/) {}
+ butil::atomic<uint32_t> version{2/*skip 0*/};
// Run this task and delete this struct.
- // Returns true if fn(arg) did run.
- bool run_and_delete();
+ // If fn(arg) runs, record the task latency into `task_latency` in
+ // microseconds. Returns true if fn(arg) did run.
+ bool run_and_delete(bvar::LatencyRecorder& task_latency);
// Delete this struct if this task was unscheduled.
// Returns true on deletion.
@@ -298,13 +296,15 @@ int TimerThread::unschedule(TaskId task_id) {
return (expected_version == id_version + 1) ? 1 : -1;
}
-bool TimerThread::Task::run_and_delete() {
+bool TimerThread::Task::run_and_delete(bvar::LatencyRecorder& task_latency) {
const uint32_t id_version = version_of_task_id(task_id);
uint32_t expected_version = id_version;
// This CAS is rarely contended, should be fast.
- if (version.compare_exchange_strong(
- expected_version, id_version + 1, butil::memory_order_relaxed)) {
+ if (version.compare_exchange_strong(expected_version, id_version + 1,
+ butil::memory_order_relaxed)) {
+ int64_t start_us = butil::cpuwide_time_us();
fn(arg);
+ task_latency << butil::cpuwide_time_us() - start_us;
// The release fence is paired with acquire fence in
// TimerThread::unschedule to make changes of fn(arg) visible.
version.store(id_version + 2, butil::memory_order_release);
@@ -365,10 +365,12 @@ void TimerThread::run() {
double busy_seconds = 0;
bvar::PassiveStatus<double> busy_seconds_var(deref_value<double>,
&busy_seconds);
bvar::PerSecond<bvar::PassiveStatus<double> >
busy_seconds_second(&busy_seconds_var);
+ bvar::LatencyRecorder task_latency;
if (!_options.bvar_prefix.empty()) {
nscheduled_second.expose_as(_options.bvar_prefix, "scheduled_second");
ntriggered_second.expose_as(_options.bvar_prefix, "triggered_second");
busy_seconds_second.expose_as(_options.bvar_prefix, "usage");
+ task_latency.expose(_options.bvar_prefix, "task");
}
while (!_stop.load(butil::memory_order_relaxed)) {
@@ -450,7 +452,7 @@ void TimerThread::run() {
}
std::pop_heap(tasks.begin(), tasks.end(), task_greater);
tasks.pop_back();
- if (task1->run_and_delete()) {
+ if (task1->run_and_delete(task_latency)) {
++ntriggered;
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
