This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new c3527a8f161 branch-4.2: [fix](be) Avoid remote IO under cache shard
locks (#69324) (#69708)
c3527a8f161 is described below
commit c3527a8f16167fbb4c7e357a55d96e98ab9cfe8d
Author: daidai <[email protected]>
AuthorDate: Thu Oct 29 16:29:43 2026 +0800
branch-4.2: [fix](be) Avoid remote IO under cache shard locks (#69324)
(#69708)
pick #69324
---
be/src/format/format_common.h | 77 +++++++++++----
be/test/format/format_common_test.cpp | 174 ++++++++++++++++++++++++++++++++++
2 files changed, 231 insertions(+), 20 deletions(-)
diff --git a/be/src/format/format_common.h b/be/src/format/format_common.h
index bf18688dead..ae227a37a46 100644
--- a/be/src/format/format_common.h
+++ b/be/src/format/format_common.h
@@ -17,6 +17,10 @@
#pragma once
+#include <memory>
+#include <typeindex>
+
+#include "core/data_type/primitive_type.h"
#include "core/types.h"
#include "exec/common/int_exp.h"
@@ -72,40 +76,47 @@ template <typename KType>
class KVCache {
public:
KVCache() = default;
+ ~KVCache() = default;
- ~KVCache() {
- for (auto& kv : _storage) {
- _delete_fn[kv.first](kv.second);
- }
- }
-
+ template <class T>
void erase(const KType& key) {
- std::lock_guard<std::mutex> lock(_lock);
- auto it = _storage.find(key);
- if (it != _storage.end()) {
- _delete_fn[key](_storage[key]);
- _storage.erase(key);
- _delete_fn.erase(key);
+ std::shared_ptr<Entry> erased_entry;
+ {
+ std::lock_guard<std::mutex> lock(_lock);
+ auto key_it = _entries.find(key);
+ if (key_it == _entries.end()) {
+ return;
+ }
+ auto type_it = key_it->second.find(typeid(T));
+ if (type_it == key_it->second.end()) {
+ return;
+ }
+ erased_entry = std::move(type_it->second);
+ key_it->second.erase(type_it);
+ if (key_it->second.empty()) {
+ _entries.erase(key_it);
+ }
}
+ // Destroy the cached value outside the shard lock.
}
template <class T>
T* get(const KType& key, const std::function<T*()> create_func, bool*
cache_hit = nullptr) {
- std::lock_guard<std::mutex> lock(_lock);
- auto it = _storage.find(key);
- if (it != _storage.end()) {
+ std::shared_ptr<Entry> entry = _get_or_create_entry<T>(key);
+ std::lock_guard<std::mutex> entry_lock(entry->_lock);
+ if (entry->_value != nullptr) {
if (cache_hit != nullptr) {
*cache_hit = true;
}
- return reinterpret_cast<T*>(it->second);
+ return static_cast<T*>(entry->_value);
} else {
if (cache_hit != nullptr) {
*cache_hit = false;
}
T* rawPtr = create_func();
if (rawPtr != nullptr) {
- _delete_fn[key] = [](void* obj) { delete
reinterpret_cast<T*>(obj); };
- _storage[key] = rawPtr;
+ entry->_delete_fn = [](void* obj) { delete
static_cast<T*>(obj); };
+ entry->_value = rawPtr;
}
return rawPtr;
}
@@ -114,9 +125,35 @@ public:
private:
using DeleteFn = void (*)(void*);
+ struct Entry {
+ ~Entry() {
+ if (_value != nullptr) {
+ _delete_fn(_value);
+ }
+ }
+
+ // The shard lock only protects the map. Factory calls hold this entry
lock, so an
+ // unrelated key can keep loading while another key is doing remote
I/O.
+ std::mutex _lock;
+ void* _value = nullptr;
+ DeleteFn _delete_fn = nullptr;
+ };
+
+ template <class T>
+ std::shared_ptr<Entry> _get_or_create_entry(const KType& key) {
+ std::lock_guard<std::mutex> lock(_lock);
+ auto key_it = _entries.try_emplace(key).first;
+ auto type_it = key_it->second.find(typeid(T));
+ if (type_it != key_it->second.end()) {
+ return type_it->second;
+ }
+ auto entry = std::make_shared<Entry>();
+ key_it->second.emplace(typeid(T), entry);
+ return entry;
+ }
+
std::mutex _lock;
- std::unordered_map<KType, DeleteFn> _delete_fn;
- std::unordered_map<KType, void*> _storage;
+ std::unordered_map<KType, std::unordered_map<std::type_index,
std::shared_ptr<Entry>>> _entries;
};
class ShardedKVCache {
diff --git a/be/test/format/format_common_test.cpp
b/be/test/format/format_common_test.cpp
new file mode 100644
index 00000000000..051b5b36b55
--- /dev/null
+++ b/be/test/format/format_common_test.cpp
@@ -0,0 +1,174 @@
+// 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 "format/format_common.h"
+
+#include <gtest/gtest.h>
+
+#include <atomic>
+#include <chrono>
+#include <future>
+#include <memory>
+#include <thread>
+#include <vector>
+
+namespace doris {
+
+TEST(KVCacheTest, DifferentKeysOnSameShardLoadConcurrently) {
+ ShardedKVCache cache(1);
+ std::promise<void> first_entered;
+ std::promise<void> release_first;
+ std::promise<void> second_entered;
+ auto first_entered_future = first_entered.get_future();
+ auto release_first_future = release_first.get_future().share();
+ auto second_entered_future = second_entered.get_future();
+
+ std::thread first([&]() {
+ cache.get<int>("first", [&]() -> int* {
+ first_entered.set_value();
+ release_first_future.wait();
+ return new int(1);
+ });
+ });
+ first_entered_future.wait();
+
+ std::thread second([&]() {
+ cache.get<int>("second", [&]() -> int* {
+ second_entered.set_value();
+ return new int(2);
+ });
+ });
+
+ bool second_loaded_before_first =
+ second_entered_future.wait_for(std::chrono::seconds(1)) ==
std::future_status::ready;
+ release_first.set_value();
+ first.join();
+ second.join();
+
+ EXPECT_TRUE(second_loaded_before_first);
+}
+
+TEST(KVCacheTest, SameKeyLoadsOnceForConcurrentCallers) {
+ constexpr int num_callers = 8;
+ ShardedKVCache cache(1);
+ std::atomic<int> factory_calls {0};
+ std::promise<void> factory_entered;
+ std::promise<void> release_factory;
+ auto factory_entered_future = factory_entered.get_future();
+ auto release_factory_future = release_factory.get_future().share();
+ std::vector<int*> values(num_callers);
+ std::vector<std::thread> callers;
+
+ for (int i = 0; i < num_callers; ++i) {
+ callers.emplace_back([&, i]() {
+ values[i] = cache.get<int>("shared", [&]() -> int* {
+ if (++factory_calls == 1) {
+ factory_entered.set_value();
+ }
+ release_factory_future.wait();
+ return new int(7);
+ });
+ });
+ }
+
+ factory_entered_future.wait();
+ release_factory.set_value();
+ for (auto& caller : callers) {
+ caller.join();
+ }
+
+ EXPECT_EQ(factory_calls.load(), 1);
+ ASSERT_NE(values.front(), nullptr);
+ for (int* value : values) {
+ EXPECT_EQ(value, values.front());
+ }
+}
+
+TEST(KVCacheTest, FailedFactoryDoesNotPoisonEntry) {
+ ShardedKVCache cache(1);
+ std::atomic<int> factory_calls {0};
+ std::promise<void> failed_factory_entered;
+ std::promise<void> release_failed_factory;
+ auto failed_factory_entered_future = failed_factory_entered.get_future();
+ auto release_failed_factory_future =
release_failed_factory.get_future().share();
+ bool first_cache_hit = true;
+ bool second_cache_hit = true;
+ int* second_value = nullptr;
+
+ std::thread first([&]() {
+ EXPECT_EQ(cache.get<int>(
+ "retry",
+ [&]() -> int* {
+ if (++factory_calls == 1) {
+ failed_factory_entered.set_value();
+ }
+ release_failed_factory_future.wait();
+ return nullptr;
+ },
+ &first_cache_hit),
+ nullptr);
+ });
+ failed_factory_entered_future.wait();
+
+ std::thread second([&]() {
+ second_value = cache.get<int>(
+ "retry",
+ [&]() -> int* {
+ ++factory_calls;
+ return new int(9);
+ },
+ &second_cache_hit);
+ });
+
+ release_failed_factory.set_value();
+ first.join();
+ second.join();
+
+ EXPECT_EQ(factory_calls.load(), 2);
+ EXPECT_FALSE(first_cache_hit);
+ EXPECT_FALSE(second_cache_hit);
+ ASSERT_NE(second_value, nullptr);
+ EXPECT_EQ(*second_value, 9);
+}
+
+TEST(KVCacheTest, SameStringKeyIsIsolatedByValueType) {
+ struct FirstValue {
+ int value = 1;
+ };
+ struct SecondValue {
+ int value = 2;
+ };
+
+ ShardedKVCache cache(1);
+ auto* first =
+ cache.get<FirstValue>("same-key", []() -> FirstValue* { return new
FirstValue(); });
+ auto* second =
+ cache.get<SecondValue>("same-key", []() -> SecondValue* { return
new SecondValue(); });
+
+ ASSERT_NE(first, nullptr);
+ ASSERT_NE(second, nullptr);
+ EXPECT_EQ(first->value, 1);
+ EXPECT_EQ(second->value, 2);
+ EXPECT_EQ(cache.get<FirstValue>("same-key",
+ []() -> FirstValue* {
+ ADD_FAILURE() << "cached FirstValue
was rebuilt";
+ return nullptr;
+ }),
+ first);
+}
+
+} // namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
