github-actions[bot] commented on code in PR #68991:
URL: https://github.com/apache/doris/pull/68991#discussion_r4272525603
##########
be/src/io/cache/fs_file_cache_storage.cpp:
##########
@@ -789,8 +790,11 @@ bool FSFileCacheStorage::handle_already_loaded_block(
}
auto block = cell_it->second.file_block;
- if (tablet_id != 0 && block->tablet_id() == 0) {
- block->set_tablet_id(tablet_id);
+ if (loaded.tablet_id != 0 && block->tablet_id() == 0) {
+ // Restored from an LRU dump, which records neither the tablet nor the
class; the queue
+ // the dump put it in is only a guess until both are known.
+ mgr->complete_loaded_cell(cell_it->second, loaded.tablet_id,
loaded.cache_type,
Review Comment:
[P2] Restore the real deadline for tablet-zero LRU entries
A TTL entry restored from an LRU dump starts with tablet_id=0 and the
synthetic expiration_time=10800. For a valid tablet-zero block (including
legacy v2 cache data) whose loaded metadata has a future deadline, this guard
skips installing the real deadline and class. Once loading finishes, the new
sweep judges is_protected(0,10800,now) false and moves the block to NORMAL,
making it evictable before its actual deadline. Previously its restored TTL
queue membership kept it protected. Distinguish an incomplete LRU placeholder
from resolved tablet-zero metadata and complete it regardless of the real
tablet id; cover a future-deadline tablet-zero block with LRU dumping enabled.
This is separate from missing per-tablet policy copies, since tablet-zero data
intentionally uses only its block deadline.
##########
be/src/io/cache/tablet_cache_policy_registry.cpp:
##########
@@ -0,0 +1,434 @@
+// 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 "io/cache/tablet_cache_policy_registry.h"
+
+#include <bvar/bvar.h>
+
+#include <algorithm>
+#include <chrono>
+#include <limits>
+
+#include "common/config.h"
+#include "common/logging.h"
+#include "cpp/sync_point.h"
+#include "io/cache/cache_block_meta_store.h"
+#include "runtime/exec_env.h"
+#include "storage/tablet/tablet_meta.h"
+#include "util/thread.h"
+#include "util/time.h"
+
+namespace doris::io {
+
+bvar::Adder<uint64_t>
g_tablet_cache_policy_fetch_num("file_cache_tablet_policy_fetch_num");
+bvar::Adder<uint64_t> g_tablet_cache_policy_fetch_failed_num(
+ "file_cache_tablet_policy_fetch_failed_num");
+bvar::PassiveStatus<size_t> g_tablet_cache_policy_num(
+ "file_cache_tablet_policy_num",
+ [](void*) { return TabletCachePolicyRegistry::instance()->size(); },
nullptr);
+
+namespace {
+
+// How often the record is scanned for policies nobody needs any more.
+constexpr int64_t kPruneIntervalS = 60;
+// How long a tablet whose policy could not be fetched is left alone.
+constexpr int64_t kFetchRetryIntervalS = 60;
+// How many reported tablets are looked up per hold of the exclusive lock.
+constexpr size_t kReportBatchSize = 1024;
+// How often the background refresh runs.
+constexpr std::chrono::seconds kRefreshInterval {1};
+
+Status fetch_from_tablet_meta(int64_t tablet_id, TabletCachePolicy* policy) {
+ TabletMetaSharedPtr tablet_meta;
+ RETURN_IF_ERROR(ExecEnv::get_tablet_meta(tablet_id, &tablet_meta, false));
+ DORIS_CHECK(tablet_meta != nullptr);
+ policy->ttl_seconds = tablet_meta->ttl_seconds();
+ policy->ctime = tablet_meta->creation_time();
+ return Status::OK();
+}
+
+} // namespace
+
+int64_t TabletCachePolicy::deadline() const {
+ if (ttl_seconds <= 0 || ctime <= 0) {
+ return 0;
+ }
+ // Same rule as TabletMeta::file_cache_ttl_expiration_time(): a huge ttl
must not wrap into
+ // a deadline in the past.
+ if (ctime > std::numeric_limits<int64_t>::max() - ttl_seconds) {
+ return std::numeric_limits<int64_t>::max();
+ }
+ return ctime + ttl_seconds;
+}
+
+TabletCachePolicyRegistry* TabletCachePolicyRegistry::instance() {
+ // Never destroyed: a cache path may still be running when the process
exits.
+ static auto* registry = new TabletCachePolicyRegistry();
+ return registry;
+}
+
+std::optional<int64_t> TabletCachePolicyRegistry::deadline(int64_t tablet_id)
const {
+ std::shared_lock lock(_mutex);
+ auto it = _entries.find(tablet_id);
+ if (it == _entries.end()) {
+ return std::nullopt;
+ }
+ return it->second.policy.deadline();
+}
+
+int64_t TabletCachePolicyRegistry::protected_until(int64_t tablet_id, int64_t
legacy_deadline,
+ bool* legacy) {
+ // tablet_id 0 is data outside any Doris tablet (external tables, v2 cache
files that never
+ // recorded one): it only has the deadline it was written with.
+ if (tablet_id > 0) {
+ if (const std::optional<int64_t> known = deadline(tablet_id);
known.has_value()) {
+ *legacy = false;
+ return *known;
+ }
+ note_tablet(tablet_id);
+ }
+ *legacy = true;
+ return legacy_deadline;
+}
+
+bool TabletCachePolicyRegistry::is_protected(int64_t tablet_id, int64_t
legacy_deadline,
+ int64_t now) {
+ bool legacy = false;
+ return protected_until(tablet_id, legacy_deadline, &legacy) > now;
+}
+
+std::optional<TabletCachePolicy> TabletCachePolicyRegistry::set_locked(
+ int64_t tablet_id, const TabletCachePolicy& policy, int64_t
refreshed_at_s) {
+ auto [it, inserted] =
+ _entries.try_emplace(tablet_id, Entry {policy, refreshed_at_s,
refreshed_at_s});
+ std::optional<TabletCachePolicy> before;
+ if (!inserted) {
+ before = it->second.policy;
+ if (const int64_t old_deadline = before->deadline(); old_deadline > 0)
{
+ _deadlines.erase(_deadlines.find(old_deadline));
+ }
+ it->second.policy = policy;
+ it->second.refreshed_at_s = refreshed_at_s;
+ }
+ if (const int64_t new_deadline = policy.deadline(); new_deadline > 0) {
+ _deadlines.insert(new_deadline);
+ }
+ return before;
+}
+
+std::optional<TabletCachePolicy>
TabletCachePolicyRegistry::erase_locked(int64_t tablet_id) {
+ auto it = _entries.find(tablet_id);
+ if (it == _entries.end()) {
+ return std::nullopt;
+ }
+ const TabletCachePolicy before = it->second.policy;
+ if (const int64_t old_deadline = before.deadline(); old_deadline > 0) {
+ _deadlines.erase(_deadlines.find(old_deadline));
+ }
+ _entries.erase(it);
+ return before;
+}
+
+void TabletCachePolicyRegistry::bump_versions(const
std::optional<TabletCachePolicy>& before,
+ const
std::optional<TabletCachePolicy>& after) {
+ // A block is protected while the deadline is ahead, so a later deadline
can only make blocks
+ // protected and an earlier one only end their protection. An unknown
tablet is judged by the
+ // deadline each block carries, which may be earlier or later, so learning
or forgetting a
+ // tablet can move its blocks either way. The -1 is never compared: a
missing side bumps both.
+ const int64_t old_deadline = before.has_value() ? before->deadline() : -1;
+ const int64_t new_deadline = after.has_value() ? after->deadline() : -1;
+ if (!before.has_value() || !after.has_value() || new_deadline >
old_deadline) {
+ _protection_gained_version.fetch_add(1, std::memory_order_acq_rel);
+ }
+ if (!before.has_value() || !after.has_value() || new_deadline <
old_deadline) {
+ _protection_lost_version.fetch_add(1, std::memory_order_acq_rel);
+ }
+}
+
+void TabletCachePolicyRegistry::persist(int64_t tablet_id,
+ const
std::optional<TabletCachePolicy>& policy) {
+ std::lock_guard lock(_stores_mutex);
+ for (auto* store : _stores) {
+ if (policy.has_value()) {
+ store->put_tablet_policy(tablet_id, *policy);
+ } else {
+ store->delete_tablet_policy(tablet_id);
+ }
+ }
+}
+
+void TabletCachePolicyRegistry::upsert(int64_t tablet_id, const
TabletCachePolicy& policy) {
+ DORIS_CHECK(tablet_id > 0);
+ const int64_t now = UnixSeconds();
+ {
+ // Unchanged and recently confirmed is the common case: every meta
refresh reports the
+ // policy. Only take the exclusive lock to record a change, or to push
the refresh time
+ // forward once it is half way to being due.
+ std::shared_lock lock(_mutex);
+ auto it = _entries.find(tablet_id);
+ if (it != _entries.end() && it->second.policy == policy &&
+ it->second.refreshed_at_s >
+ now - config::file_cache_tablet_policy_refresh_interval_s
/ 2) {
+ return;
+ }
+ }
+ std::optional<TabletCachePolicy> before;
+ {
+ std::unique_lock lock(_mutex);
+ before = set_locked(tablet_id, policy, now);
+ }
+ if (before == policy) {
+ return;
+ }
+ bump_versions(before, policy);
+ persist(tablet_id, policy);
+}
+
+void TabletCachePolicyRegistry::erase(int64_t tablet_id) {
+ std::optional<TabletCachePolicy> before;
+ {
+ std::unique_lock lock(_mutex);
+ before = erase_locked(tablet_id);
+ }
+ if (!before.has_value()) {
+ return;
+ }
+ bump_versions(before, std::nullopt);
+ persist(tablet_id, std::nullopt);
+}
+
+void TabletCachePolicyRegistry::restore(int64_t tablet_id, const
TabletCachePolicy& policy) {
+ std::optional<TabletCachePolicy> before;
+ {
+ std::unique_lock lock(_mutex);
+ if (_entries.contains(tablet_id)) {
+ // Already known from the live meta or from another cache path's
copy.
+ return;
+ }
+ // Never refreshed by this process, so it is due as soon as a cache
path reports it;
+ // counted as reported now so that it is not forgotten before the
cache has loaded.
+ before = set_locked(tablet_id, policy, 0);
+ _entries[tablet_id].reported_at_s = UnixSeconds();
+ }
+ bump_versions(before, policy);
+}
+
+void TabletCachePolicyRegistry::note_tablet(int64_t tablet_id) {
+ std::lock_guard pending_lock(_pending_mutex);
+ _noted.insert(tablet_id);
+}
+
+bool TabletCachePolicyRegistry::has_deadline_in(int64_t from, int64_t to)
const {
+ std::shared_lock lock(_mutex);
+ auto it = _deadlines.upper_bound(from);
+ return it != _deadlines.end() && *it <= to;
+}
+
+void TabletCachePolicyRegistry::report_cached_tablets(const
std::vector<int64_t>& tablet_ids) {
+ const int64_t now = UnixSeconds();
+ const int64_t stale_before = now -
config::file_cache_tablet_policy_refresh_interval_s;
+ std::vector<int64_t> due;
+ // In batches: a path may hold blocks of many tablets, and the readers
waiting on this lock
+ // hold their cache lock.
+ for (size_t begin = 0; begin < tablet_ids.size(); begin +=
kReportBatchSize) {
+ const size_t end = std::min(tablet_ids.size(), begin +
kReportBatchSize);
+ std::unique_lock lock(_mutex);
+ for (size_t i = begin; i < end; ++i) {
+ const int64_t tablet_id = tablet_ids[i];
+ auto it = _entries.find(tablet_id);
+ if (it == _entries.end()) {
+ due.push_back(tablet_id);
+ continue;
+ }
+ it->second.reported_at_s = now;
+ if (it->second.refreshed_at_s <= stale_before) {
+ due.push_back(tablet_id);
+ }
+ }
+ }
+ std::lock_guard pending_lock(_pending_mutex);
+ _due.insert(due.begin(), due.end());
+}
+
+std::vector<int64_t> TabletCachePolicyRegistry::take_due(int64_t now, size_t
budget) {
+ std::unordered_set<int64_t> noted;
+ {
+ std::lock_guard pending_lock(_pending_mutex);
+ noted.swap(_noted);
+ }
+ // A noted tablet may have been learned since.
+ std::erase_if(noted, [this](int64_t tablet_id) { return
deadline(tablet_id).has_value(); });
+ std::lock_guard pending_lock(_pending_mutex);
+ _due.insert(noted.begin(), noted.end());
+ std::vector<int64_t> batch;
+ for (auto it = _due.begin(); it != _due.end() && batch.size() < budget; it
= _due.erase(it)) {
+ auto retry = _retry_at.find(*it);
+ if (retry == _retry_at.end() || retry->second <= now) {
+ batch.push_back(*it);
+ }
+ }
+ return batch;
+}
+
+void TabletCachePolicyRegistry::prune(int64_t now) {
+ if (now < _next_prune_s) {
+ return;
+ }
+ _next_prune_s = now + kPruneIntervalS;
+ std::erase_if(_retry_at, [now](const auto& retry) { return retry.second <=
now; });
+ // A tablet no cache path holds and nothing refreshes (unloaded, or
dropped) has no blocks
+ // to judge; it is learned again if its blocks come back.
+ const int64_t forget_before = now - 2 *
config::file_cache_tablet_policy_refresh_interval_s;
+ auto is_unused = [forget_before](const Entry& entry) {
+ return std::max(entry.refreshed_at_s, entry.reported_at_s) <
forget_before;
+ };
+ std::vector<int64_t> unused;
+ {
+ std::shared_lock lock(_mutex);
+ for (const auto& [id, entry] : _entries) {
+ if (is_unused(entry)) {
+ unused.push_back(id);
+ }
+ }
+ }
+ TEST_SYNC_POINT_CALLBACK("TabletCachePolicyRegistry::prune:collected",
this, &unused);
+ for (int64_t id : unused) {
+ std::optional<TabletCachePolicy> before;
+ {
+ std::unique_lock lock(_mutex);
+ // Checked again: it may have been refreshed or reported since.
+ if (auto it = _entries.find(id); it != _entries.end() &&
is_unused(it->second)) {
+ before = erase_locked(id);
+ }
+ }
+ if (before.has_value()) {
+ bump_versions(before, std::nullopt);
+ persist(id, std::nullopt);
+ }
+ }
+}
+
+void TabletCachePolicyRegistry::refresh(size_t budget) {
+ std::lock_guard refresh_lock(_refresh_mutex);
+ const int64_t now = UnixSeconds();
+ const std::vector<int64_t> batch = take_due(now, budget);
+ prune(now);
+ const Fetcher& fetch = _fetcher ? _fetcher :
Fetcher(fetch_from_tablet_meta);
+ for (size_t i = 0; i < batch.size(); ++i) {
+ const int64_t tablet_id = batch[i];
+ TabletCachePolicy policy;
+ Status st = fetch(tablet_id, &policy);
+ g_tablet_cache_policy_fetch_num << 1;
+ if (st.ok()) {
+ _retry_at.erase(tablet_id);
+ upsert(tablet_id, policy);
+ continue;
+ }
+ if (st.is<ErrorCode::NOT_FOUND>()) {
+ _retry_at.erase(tablet_id);
Review Comment:
[P2] Keep backoff state for tablets confirmed absent
After a restart, cached blocks can still belong to a dropped tablet with no
loaded CloudTablet available to clean them up. This branch forgets both its
policy and retry deadline. The next eviction, sweep or file_cache_info check
sees an unknown tablet and notes it again, so the next one-second refresh sends
another GetTablet RPC; periodic cached-tablet reports also keep rediscovering
it. With repeated checks, absent tablets can keep consuming the refresh budget
and delay useful policy fetches. Retain bounded negative-result/backoff state
for NOT_FOUND and invalidate it when a foreground update learns the tablet. The
existing test never re-notes/reports the absent id before its second refresh.
This concerns redundant RPCs across rounds, separately from the fixed
duplicate-notification backlog.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
