This is an automated email from the ASF dual-hosted git repository.
chenBright 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 7c524731 Cache AgentId in native bvar recorders and reducers (#3709)
7c524731 is described below
commit 7c5247319076c1395e39545f38d8a2e305c64fd3
Author: Bright Chen <[email protected]>
AuthorDate: Wed Nov 4 21:46:29 2026 +0800
Cache AgentId in native bvar recorders and reducers (#3709)
---
src/bvar/detail/combiner.h | 9 ++
src/bvar/recorder.h | 16 ++--
src/bvar/reducer.h | 12 ++-
test/bvar_cached_agent_id_unittest.cpp | 159 +++++++++++++++++++++++++++++++++
4 files changed, 188 insertions(+), 8 deletions(-)
diff --git a/src/bvar/detail/combiner.h b/src/bvar/detail/combiner.h
index 7f531d94..8fdf8e82 100644
--- a/src/bvar/detail/combiner.h
+++ b/src/bvar/detail/combiner.h
@@ -340,6 +340,15 @@ friend class GlobalValue<self_type>;
call_op_returning_void(_op, _global_result, prev);
}
+ AgentId agent_id() const { return _id; }
+
+ // The caller must keep the combiner owning id alive. An allocated TLS
+ // slot may still be unbound, or belong to an expired combiner after reuse.
+ inline static Agent* get_bound_tls_agent(AgentId id) {
+ Agent* agent = AgentGroup::get_tls_agent(id);
+ return agent && !agent->combiner.expired() ? agent : nullptr;
+ }
+
// We need this function to be as fast as possible.
Agent* get_or_create_tls_agent() {
Agent* agent = AgentGroup::get_tls_agent(_id);
diff --git a/src/bvar/recorder.h b/src/bvar/recorder.h
index 147fcb72..6220d553 100644
--- a/src/bvar/recorder.h
+++ b/src/bvar/recorder.h
@@ -124,7 +124,7 @@ public:
typedef typename combiner_type::self_shared_type shared_combiner_type;
typedef combiner_type::Agent agent_type;
- IntRecorder() : _combiner(std::make_shared<combiner_type>()),
_sampler(nullptr) {}
+ IntRecorder() = default;
IntRecorder(const butil::StringPiece& name) : IntRecorder() {
expose(name);
@@ -263,8 +263,10 @@ private:
}
private:
- shared_combiner_type _combiner;
- sampler_type* _sampler;
+ shared_combiner_type _combiner{std::make_shared<combiner_type>()};
+ // `_combiner` keeps this ID allocated for the lifetime of the recorder.
+ const detail::AgentId _agent_id{_combiner->agent_id()};
+ sampler_type* _sampler{nullptr};
std::string _debug_name;
};
@@ -291,8 +293,12 @@ inline IntRecorder& IntRecorder::operator<<(int64_t
sample) {
<< (void*)this << ") " << reason;
}
}
- agent_type* agent = _combiner->get_or_create_tls_agent();
- if (BAIDU_UNLIKELY(!agent)) {
+
+ agent_type* agent = combiner_type::get_bound_tls_agent(_agent_id);
+ if (BAIDU_UNLIKELY(agent == nullptr)) {
+ agent = _combiner->get_or_create_tls_agent();
+ }
+ if (BAIDU_UNLIKELY(agent == nullptr)) {
LOG(FATAL) << "Fail to create agent";
return *this;
}
diff --git a/src/bvar/reducer.h b/src/bvar/reducer.h
index c49d1e4c..a7d51f49 100644
--- a/src/bvar/reducer.h
+++ b/src/bvar/reducer.h
@@ -250,7 +250,8 @@ public:
explicit Reducer(typename butil::add_cr_non_integral<T>::type identity =
T(),
const Op& op = Op(), const InvOp& inv_op = InvOp())
: _combiner(std::make_shared<combiner_type>(identity, identity, op))
- , _sampler(nullptr) , _series_sampler(nullptr) , _inv_op(inv_op) {}
+ , _agent_id(_combiner->agent_id()) , _sampler(nullptr)
+ , _series_sampler(nullptr) , _inv_op(inv_op) {}
~Reducer() override {
// Calling hide() manually is a MUST required by Variable.
@@ -351,6 +352,8 @@ protected:
private:
shared_combiner_type _combiner;
+ // `_combiner` keeps this ID allocated for the lifetime of the reducer.
+ const detail::AgentId _agent_id;
sampler_type* _sampler;
SeriesSampler* _series_sampler;
InvOp _inv_op;
@@ -360,8 +363,11 @@ template <typename T, typename Op, typename InvOp>
inline Reducer<T, Op, InvOp>& Reducer<T, Op, InvOp>::operator<<(
typename butil::add_cr_non_integral<T>::type value) {
// It's wait-free for most time
- agent_type* agent = _combiner->get_or_create_tls_agent();
- if (__builtin_expect(!agent, 0)) {
+ agent_type* agent = combiner_type::get_bound_tls_agent(_agent_id);
+ if (BAIDU_UNLIKELY(agent == nullptr)) {
+ agent = _combiner->get_or_create_tls_agent();
+ }
+ if (BAIDU_UNLIKELY(agent == nullptr)) {
LOG(FATAL) << "Fail to create agent";
return *this;
}
diff --git a/test/bvar_cached_agent_id_unittest.cpp
b/test/bvar_cached_agent_id_unittest.cpp
new file mode 100644
index 00000000..9e08222c
--- /dev/null
+++ b/test/bvar_cached_agent_id_unittest.cpp
@@ -0,0 +1,159 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <algorithm>
+#include <limits>
+#include <memory>
+#include <thread>
+
+#include <gtest/gtest.h>
+
+#include "bvar/recorder.h"
+#include "bvar/reducer.h"
+
+namespace {
+
+struct SelectExtremum {
+ explicit SelectExtremum(bool select_min) : minimum(select_min) {}
+ void operator()(int& lhs, int rhs) const {
+ lhs = minimum ? std::min(lhs, rhs) : std::max(lhs, rhs);
+ }
+ bool minimum;
+};
+
+TEST(CachedAgentIdTest, unbound_slot_and_stateful_operator) {
+ typedef bvar::Reducer<int, SelectExtremum> Reducer;
+ typedef Reducer::combiner_type Combiner;
+ Reducer first(std::numeric_limits<int>::max(), SelectExtremum(true));
+ Reducer second(std::numeric_limits<int>::lowest(), SelectExtremum(false));
+ const bvar::detail::AgentId id = first.share_combiner()->agent_id();
+ ASSERT_EQ(nullptr, Combiner::get_bound_tls_agent(-1));
+ ASSERT_NE(nullptr, Combiner::AgentGroup::get_or_create_tls_agent(id));
+ ASSERT_EQ(nullptr, Combiner::get_bound_tls_agent(id));
+ first << 12 << 42;
+ second << 12 << 42;
+ ASSERT_NE(nullptr, Combiner::get_bound_tls_agent(id));
+ ASSERT_EQ(12, first.get_value());
+ ASSERT_EQ(42, second.get_value());
+ ASSERT_EQ(12, first.reset());
+ first << 3;
+ ASSERT_EQ(3, first.get_value());
+}
+
+TEST(CachedAgentIdTest, reducer_reuses_expired_slot) {
+ typedef bvar::Reducer<int, SelectExtremum> Reducer;
+ typedef Reducer::combiner_type Combiner;
+ bvar::detail::AgentId id;
+ {
+ Reducer old(std::numeric_limits<int>::lowest(), SelectExtremum(false));
+ id = old.share_combiner()->agent_id();
+ old << 80;
+ ASSERT_EQ(80, old.get_value());
+ }
+ Reducer fresh(std::numeric_limits<int>::max(), SelectExtremum(true));
+ ASSERT_EQ(id, fresh.share_combiner()->agent_id());
+ ASSERT_NE(nullptr, Combiner::AgentGroup::get_tls_agent(id));
+ ASSERT_EQ(nullptr, Combiner::get_bound_tls_agent(id));
+ fresh << 100;
+ ASSERT_EQ(100, fresh.get_value());
+ fresh << -20;
+ ASSERT_EQ(-20, fresh.get_value());
+}
+
+TEST(CachedAgentIdTest, retained_reducer_combiner_keeps_id_alive) {
+ typedef bvar::Reducer<int, SelectExtremum> Reducer;
+ typedef Reducer::combiner_type Combiner;
+ Combiner::self_shared_type held;
+ bvar::detail::AgentId id;
+ {
+ Reducer old(std::numeric_limits<int>::lowest(), SelectExtremum(false));
+ held = old.share_combiner();
+ id = held->agent_id();
+ old << 80;
+ }
+ Reducer other(std::numeric_limits<int>::max(), SelectExtremum(true));
+ ASSERT_NE(id, other.share_combiner()->agent_id());
+ other << 5;
+ ASSERT_EQ(80, held->combine_agents());
+ ASSERT_EQ(5, other.get_value());
+ held.reset();
+ Reducer fresh(std::numeric_limits<int>::lowest(), SelectExtremum(false));
+ ASSERT_EQ(id, fresh.share_combiner()->agent_id());
+ fresh << 6;
+ ASSERT_EQ(6, fresh.get_value());
+}
+
+#if !WITH_BABYLON_COUNTER
+TEST(CachedAgentIdTest, recorder_reuses_slot_after_last_owner_releases) {
+ typedef bvar::IntRecorder::combiner_type Combiner;
+ Combiner::self_shared_type held;
+ bvar::detail::AgentId id;
+ {
+ bvar::IntRecorder old;
+ held = old.share_combiner();
+ id = held->agent_id();
+ old << 80;
+ }
+ bvar::IntRecorder other;
+ ASSERT_NE(id, other.share_combiner()->agent_id());
+ other << 5;
+ ASSERT_EQ(80, held->combine_agents().sum);
+ ASSERT_EQ(1, held->combine_agents().num);
+ held.reset();
+ bvar::IntRecorder fresh;
+ ASSERT_EQ(id, fresh.share_combiner()->agent_id());
+ ASSERT_NE(nullptr, Combiner::AgentGroup::get_tls_agent(id));
+ ASSERT_EQ(nullptr, Combiner::get_bound_tls_agent(id));
+ fresh << 6 << 7;
+ ASSERT_EQ(13, fresh.get_value().sum);
+ ASSERT_EQ(2, fresh.get_value().num);
+ ASSERT_EQ(13, fresh.reset().sum);
+ fresh << -4;
+ ASSERT_EQ(-4, fresh.get_value().sum);
+ ASSERT_EQ(1, fresh.get_value().num);
+}
+
+TEST(CachedAgentIdTest, fresh_threads_initialize_and_commit_on_exit) {
+ typedef bvar::IntRecorder::combiner_type RecorderCombiner;
+ typedef bvar::Adder<int64_t>::combiner_type AdderCombiner;
+ bvar::IntRecorder recorder;
+ bvar::Adder<int64_t> adder;
+ const auto recorder_id = recorder.share_combiner()->agent_id();
+ const auto adder_id = adder.share_combiner()->agent_id();
+ std::thread workers[8];
+ for (auto& worker : workers) {
+ worker = std::thread([&] {
+ EXPECT_EQ(nullptr,
+ RecorderCombiner::get_bound_tls_agent(recorder_id));
+ EXPECT_EQ(nullptr,
+ AdderCombiner::get_bound_tls_agent(adder_id));
+ for (int i = 0; i < 10000; ++i) {
+ recorder << 3;
+ adder << 2;
+ }
+ });
+ }
+ for (auto& worker : workers) {
+ worker.join();
+ }
+ ASSERT_EQ(80000, recorder.get_value().num);
+ ASSERT_EQ(240000, recorder.get_value().sum);
+ ASSERT_EQ(160000, adder.get_value());
+}
+#endif // !WITH_BABYLON_COUNTER
+
+} // namespace
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
