github-actions[bot] commented on code in PR #68991:
URL: https://github.com/apache/doris/pull/68991#discussion_r4256822568
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -2106,25 +2168,211 @@ std::string
BlockFileCache::dump_single_cache_type_unlocked(const UInt128Wrapper
return cache_type_to_string(cell->second.file_block->cache_type());
}
-void BlockFileCache::change_cache_type(const UInt128Wrapper& hash, size_t
offset,
- FileCacheType new_type,
- std::lock_guard<std::mutex>&
cache_lock) {
- if (auto iter = _files.find(hash); iter != _files.end()) {
- auto& file_blocks = iter->second;
- if (auto cell_it = file_blocks.find(offset); cell_it !=
file_blocks.end()) {
- FileBlockCell& cell = cell_it->second;
- auto& cur_queue = get_queue(cell.file_block->cache_type());
- DCHECK(cell.queue_iterator.has_value());
- cur_queue.remove(*cell.queue_iterator, cache_lock);
- _lru_recorder->record_queue_event(
- cell.file_block->cache_type(), CacheLRULogType::REMOVE,
- cell.file_block->get_hash_value(),
cell.file_block->offset(), cell.size());
- auto& new_queue = get_queue(new_type);
- cell.queue_iterator =
- new_queue.add(hash, offset,
cell.file_block->range().size(), cache_lock);
- _lru_recorder->record_queue_event(new_type, CacheLRULogType::ADD,
-
cell.file_block->get_hash_value(),
- cell.file_block->offset(),
cell.size());
+bool BlockFileCache::is_protected(const FileBlock& block, int64_t now) {
+ return is_protected(block.tablet_id(),
static_cast<int64_t>(block.expiration_time()), now);
+}
+
+bool BlockFileCache::is_protected(int64_t tablet_id, int64_t legacy_deadline,
int64_t now) {
+ bool legacy = false;
+ const int64_t deadline = _policy_registry->protected_until(tablet_id,
legacy_deadline, &legacy);
+ if (deadline <= now) {
+ return false;
+ }
+ if (legacy) {
+ int64_t next = _next_legacy_deadline.load(std::memory_order_relaxed);
+ while (deadline < next && !_next_legacy_deadline.compare_exchange_weak(
+ next, deadline,
std::memory_order_relaxed)) {
+ }
+ }
+ return true;
+}
+
+FileCacheType BlockFileCache::queue_type_for(int64_t tablet_id, int64_t
legacy_deadline,
+ FileCacheType cache_class,
int64_t now) {
+ DCHECK(cache_class != FileCacheType::TTL);
+ return is_protected(tablet_id, legacy_deadline, now) ? FileCacheType::TTL
: cache_class;
+}
+
+void BlockFileCache::move_cell_to_queue(FileBlockCell& cell, FileCacheType
new_type,
+ std::lock_guard<std::mutex>&
cache_lock) {
+ auto& file_block = *cell.file_block;
+ const FileCacheType old_type = file_block.cache_type();
+ DCHECK(old_type != new_type);
+ DCHECK(cell.queue_iterator.has_value());
+ get_queue(old_type).remove(*cell.queue_iterator, cache_lock);
+ _lru_recorder->record_queue_event(old_type, CacheLRULogType::REMOVE,
+ file_block.get_hash_value(),
file_block.offset(),
+ cell.size());
+ cell.queue_iterator = get_queue(new_type).add_front(
+ file_block.get_hash_value(), file_block.offset(), cell.size(),
cache_lock);
+ _lru_recorder->record_queue_event(new_type, CacheLRULogType::ADDTOFRONT,
+ file_block.get_hash_value(),
file_block.offset(),
+ cell.size());
+ file_block._key.meta.type = new_type;
Review Comment:
[P2] Synchronize queue-type writes with block-only readers
This assignment holds only the cache mutex, but
FileBlock::get_info_for_log() reads the same non-atomic field while holding
only the block mutex. The sweep also moves referenced and DOWNLOADING blocks,
so a query or peer-fill thread formatting diagnostics can run concurrently with
this write after ALTER or expiry. Removing change_cache_type_lock() removed the
common block lock and introduces a C++ data race. Take the block mutex in
cache-then-block order for this mutation, or give queue membership its own
consistently synchronized representation.
##########
be/src/io/cache/tablet_cache_policy_registry.cpp:
##########
@@ -0,0 +1,433 @@
+// 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) {
+ // An unknown tablet is judged by the deadline its blocks carry, so
learning or forgetting a
+ // tablet can move its blocks either way.
+ 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);
Review Comment:
[P1] Preserve policy commit order in persisted copies
The policy mutex is released before persist(), so two upserts can commit P0
then P1 in memory but persist P1 then P0: pause the first caller after
set_locked(), let the second finish, then resume the first. _stores_mutex
orders the enqueue calls rather than the registry commits. Memory now holds P1
while every cache path can retain P0; unchanged confirmations return without
repairing the copy, and a clean restart restores the wrong TTL. erase()/prune()
can similarly delete a newer persisted policy. Carry and enforce a per-tablet
generation through the asynchronous writer so older writes/deletes cannot
replace newer commits.
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -2106,25 +2168,211 @@ std::string
BlockFileCache::dump_single_cache_type_unlocked(const UInt128Wrapper
return cache_type_to_string(cell->second.file_block->cache_type());
}
-void BlockFileCache::change_cache_type(const UInt128Wrapper& hash, size_t
offset,
- FileCacheType new_type,
- std::lock_guard<std::mutex>&
cache_lock) {
- if (auto iter = _files.find(hash); iter != _files.end()) {
- auto& file_blocks = iter->second;
- if (auto cell_it = file_blocks.find(offset); cell_it !=
file_blocks.end()) {
- FileBlockCell& cell = cell_it->second;
- auto& cur_queue = get_queue(cell.file_block->cache_type());
- DCHECK(cell.queue_iterator.has_value());
- cur_queue.remove(*cell.queue_iterator, cache_lock);
- _lru_recorder->record_queue_event(
- cell.file_block->cache_type(), CacheLRULogType::REMOVE,
- cell.file_block->get_hash_value(),
cell.file_block->offset(), cell.size());
- auto& new_queue = get_queue(new_type);
- cell.queue_iterator =
- new_queue.add(hash, offset,
cell.file_block->range().size(), cache_lock);
- _lru_recorder->record_queue_event(new_type, CacheLRULogType::ADD,
-
cell.file_block->get_hash_value(),
- cell.file_block->offset(),
cell.size());
+bool BlockFileCache::is_protected(const FileBlock& block, int64_t now) {
+ return is_protected(block.tablet_id(),
static_cast<int64_t>(block.expiration_time()), now);
+}
+
+bool BlockFileCache::is_protected(int64_t tablet_id, int64_t legacy_deadline,
int64_t now) {
+ bool legacy = false;
+ const int64_t deadline = _policy_registry->protected_until(tablet_id,
legacy_deadline, &legacy);
+ if (deadline <= now) {
+ return false;
+ }
+ if (legacy) {
+ int64_t next = _next_legacy_deadline.load(std::memory_order_relaxed);
+ while (deadline < next && !_next_legacy_deadline.compare_exchange_weak(
+ next, deadline,
std::memory_order_relaxed)) {
+ }
+ }
+ return true;
+}
+
+FileCacheType BlockFileCache::queue_type_for(int64_t tablet_id, int64_t
legacy_deadline,
+ FileCacheType cache_class,
int64_t now) {
+ DCHECK(cache_class != FileCacheType::TTL);
+ return is_protected(tablet_id, legacy_deadline, now) ? FileCacheType::TTL
: cache_class;
+}
+
+void BlockFileCache::move_cell_to_queue(FileBlockCell& cell, FileCacheType
new_type,
+ std::lock_guard<std::mutex>&
cache_lock) {
+ auto& file_block = *cell.file_block;
+ const FileCacheType old_type = file_block.cache_type();
+ DCHECK(old_type != new_type);
+ DCHECK(cell.queue_iterator.has_value());
+ get_queue(old_type).remove(*cell.queue_iterator, cache_lock);
+ _lru_recorder->record_queue_event(old_type, CacheLRULogType::REMOVE,
+ file_block.get_hash_value(),
file_block.offset(),
+ cell.size());
+ cell.queue_iterator = get_queue(new_type).add_front(
+ file_block.get_hash_value(), file_block.offset(), cell.size(),
cache_lock);
+ _lru_recorder->record_queue_event(new_type, CacheLRULogType::ADDTOFRONT,
+ file_block.get_hash_value(),
file_block.offset(),
+ cell.size());
+ file_block._key.meta.type = new_type;
+ *_ttl_relocated_blocks << 1;
+}
+
+void BlockFileCache::add_cached_tablet(int64_t tablet_id) {
+ if (tablet_id > 0) {
+ ++_cached_tablets[tablet_id];
+ }
+}
+
+void BlockFileCache::remove_cached_tablet(int64_t tablet_id) {
+ if (tablet_id > 0) {
+ auto it = _cached_tablets.find(tablet_id);
+ DCHECK(it != _cached_tablets.end());
+ if (--it->second == 0) {
+ _cached_tablets.erase(it);
+ }
+ }
+}
+
+void BlockFileCache::relocate_to_ttl_queue(const std::vector<FileBlockCell*>&
cells,
+ std::lock_guard<std::mutex>&
cache_lock) {
+ for (auto* cell : cells) {
+ move_cell_to_queue(*cell, FileCacheType::TTL, cache_lock);
+ }
+}
+
+void BlockFileCache::complete_loaded_cell(FileBlockCell& cell, int64_t
tablet_id,
+ FileCacheType cache_class, uint64_t
legacy_deadline,
+ std::lock_guard<std::mutex>&
cache_lock) {
+ auto& meta = cell.file_block->_key.meta;
+ remove_cached_tablet(meta.tablet_id);
+ add_cached_tablet(tablet_id);
+ meta.tablet_id = tablet_id;
+ meta.cache_class = cache_class_of(cache_class);
+ meta.expiration_time = legacy_deadline;
+ const FileCacheType queue_type = queue_type_for(
+ tablet_id, static_cast<int64_t>(legacy_deadline),
meta.cache_class, UnixSeconds());
+ if (queue_type != meta.type) {
+ move_cell_to_queue(cell, queue_type, cache_lock);
+ }
+}
+
+void BlockFileCache::start_queue_sweeps(int64_t now) {
+ auto& state = _queue_sweep;
+ auto is_active = [&](FileCacheType type) { return
state.active[file_cache_type_index(type)]; };
+ const bool backstop_due =
+ now - state.last_backstop_at >=
config::file_cache_queue_sweep_backstop_interval_s;
+ // A pass over the TTL queue picks up blocks whose tablet lost its TTL, or
whose deadline
+ // passed, since the previous pass started.
+ if (!is_active(FileCacheType::TTL)) {
+ const uint64_t lost_version =
_policy_registry->protection_lost_version();
+ if (backstop_due || lost_version != state.ttl_pass_lost_version ||
+ _policy_registry->has_deadline_in(state.ttl_pass_started_at, now)
||
+ _next_legacy_deadline.load(std::memory_order_relaxed) <= now) {
+ state.active[file_cache_type_index(FileCacheType::TTL)] = true;
+ state.ttl_pass_lost_version = lost_version;
+ state.ttl_pass_started_at = now;
+ state.ttl_pass_legacy_bytes = 0;
+ // The pass finds again those still ahead.
+ _next_legacy_deadline.store(std::numeric_limits<int64_t>::max(),
+ std::memory_order_relaxed);
+ }
+ }
+ // A pass over the class queues picks up blocks whose tablet gained a TTL.
The three queues
+ // start together so that the version they record holds for all of them.
+ constexpr std::array<FileCacheType, 3> class_types {FileCacheType::INDEX,
FileCacheType::NORMAL,
+
FileCacheType::DISPOSABLE};
+ if (std::none_of(class_types.begin(), class_types.end(), is_active)) {
+ const uint64_t gained_version =
_policy_registry->protection_gained_version();
+ if (backstop_due || gained_version != state.class_pass_gained_version)
{
+ for (auto type : class_types) {
+ state.active[file_cache_type_index(type)] = true;
+ }
+ state.class_pass_gained_version = gained_version;
+ }
+ }
+ if (backstop_due &&
+ std::all_of(state.active.begin(), state.active.end(), [](bool active)
{ return active; })) {
+ state.last_backstop_at = now;
+ }
+}
+
+size_t BlockFileCache::sweep_queue(FileCacheType type, size_t budget, int64_t
now,
+ std::lock_guard<std::mutex>& cache_lock) {
+ auto& active = _queue_sweep.active[file_cache_type_index(type)];
+ if (!active) {
+ return 0;
+ }
+ auto& queue = get_queue(type);
+ auto& pass_left = _queue_sweep.pass_left[file_cache_type_index(type)];
+ if (!queue.sweep_cursor.has_value()) {
+ // A pass walks at most as many blocks as the queue holds when it
starts: blocks touched
+ // or added meanwhile go to the hot end, and must not keep the pass
from ending. A block
+ // touched before the cursor reached it may be left for the next pass.
+ queue.sweep_cursor = queue.begin();
+ pass_left = queue.queue.size();
+ }
+ auto& cursor = *queue.sweep_cursor;
+ size_t walked = 0;
+ std::vector<std::pair<FileBlockCell*, FileCacheType>> moves;
+ for (; walked < budget && pass_left > 0 && cursor != queue.end();
++walked, --pass_left) {
+ auto& entry = *cursor;
+ ++cursor;
+ auto* cell = get_cell(entry.hash, entry.offset, cache_lock);
+ DCHECK(cell != nullptr);
+ const auto& block = *cell->file_block;
+ const FileCacheType want =
+ is_protected(block, now) ? FileCacheType::TTL :
block.cache_class();
+ if (want != type) {
+ moves.emplace_back(cell, want);
+ } else if (type == FileCacheType::TTL && block.tablet_id() <= 0) {
+ _queue_sweep.ttl_pass_legacy_bytes += cell->size();
+ }
+ }
+ for (auto& [cell, want] : moves) {
+ move_cell_to_queue(*cell, want, cache_lock);
+ }
+ if (pass_left == 0 || queue.sweep_cursor == queue.end()) {
Review Comment:
[P2] Reschedule blocks skipped by recency touches during a sweep
After TTL removal/expiry, a batch can demote the first 2000 blocks and leave
the cursor on the remaining tail block. Reading that block calls move_to_end(),
which advances the cursor to end even for a tail splice. This branch then
completes the pass without examining the block. The lost-policy version and
deadline event were already consumed, so it can stay in TTL until the default
86400-second backstop. That stale queue still exempts it from advance eviction
and treats it as always hot for warmup. Reclassify on touch or retain a pending
rescan for skipped entries; the current busy-queue test keeps protection
enabled and misses this case.
##########
be/src/io/cache/tablet_cache_policy_registry.cpp:
##########
@@ -0,0 +1,433 @@
+// 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) {
+ // An unknown tablet is judged by the deadline its blocks carry, so
learning or forgetting a
+ // tablet can move its blocks either way.
+ 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) {
+ _noted_tablets.enqueue(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 due_lock(_due_mutex);
+ _due.insert(due.begin(), due.end());
+}
+
+std::vector<int64_t> TabletCachePolicyRegistry::take_due(int64_t now, size_t
budget) {
+ std::vector<int64_t> unknown;
+ int64_t tablet_id = 0;
+ while (_noted_tablets.try_dequeue(tablet_id)) {
Review Comment:
[P2] Bound and deduplicate pending unknown-tablet notifications
Every protection check for an unknown tablet enqueues its id, but this loop
retains every duplicate in unknown until the entire queue is empty; only
afterward does _due deduplicate and the fetch budget take effect. During a slow
metadata RPC or outage, startup and eviction/scanner traffic on multiple cache
paths can build a backlog proportional to blocks/checks rather than tablets.
Draining that backlog consumes unbounded memory/work before any policy fetch,
prune, or stop check; continuously arriving events can prevent fetching
altogether. Deduplicate pending ids and limit notification processing per
refresh so the configured budget also bounds maintenance work.
##########
be/src/io/cache/tablet_cache_policy_registry.cpp:
##########
@@ -0,0 +1,433 @@
+// 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) {
+ // An unknown tablet is judged by the deadline its blocks carry, so
learning or forgetting a
+ // tablet can move its blocks either way.
+ 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) {
+ _noted_tablets.enqueue(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 due_lock(_due_mutex);
+ _due.insert(due.begin(), due.end());
+}
+
+std::vector<int64_t> TabletCachePolicyRegistry::take_due(int64_t now, size_t
budget) {
+ std::vector<int64_t> unknown;
+ int64_t tablet_id = 0;
+ while (_noted_tablets.try_dequeue(tablet_id)) {
+ if (!deadline(tablet_id).has_value()) {
+ unknown.push_back(tablet_id);
+ }
+ }
+ std::lock_guard due_lock(_due_mutex);
+ _due.insert(unknown.begin(), unknown.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);
Review Comment:
[P1] Reject fetch results superseded by a foreground policy update
The refresh thread can read ttl=0 before ALTER and receive that RPC response
after CloudTablet::sync_meta() has fetched and upserted the new ttl=3600. This
unconditional upsert then restores ttl=0 and marks it freshly confirmed, so
eviction can discard blocks protected by the completed ALTER until another
refresh. _refresh_mutex only serializes background refreshes; the foreground
sync uses a separate _sync_meta_lock. Capture a per-tablet update generation
before fetching and apply the result (including NOT_FOUND) only if no newer
update intervened.
##########
be/src/io/cache/tablet_cache_policy_registry.cpp:
##########
@@ -0,0 +1,433 @@
+// 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) {
+ // An unknown tablet is judged by the deadline its blocks carry, so
learning or forgetting a
+ // tablet can move its blocks either way.
+ 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) {
+ _noted_tablets.enqueue(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 due_lock(_due_mutex);
+ _due.insert(due.begin(), due.end());
+}
+
+std::vector<int64_t> TabletCachePolicyRegistry::take_due(int64_t now, size_t
budget) {
+ std::vector<int64_t> unknown;
+ int64_t tablet_id = 0;
+ while (_noted_tablets.try_dequeue(tablet_id)) {
+ if (!deadline(tablet_id).has_value()) {
+ unknown.push_back(tablet_id);
+ }
+ }
+ std::lock_guard due_lock(_due_mutex);
+ _due.insert(unknown.begin(), unknown.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);
+ erase(tablet_id);
+ continue;
+ }
+ // Most likely the meta service is unavailable: the rest waits for the
next round rather
+ // than for one failed call after another. Meanwhile a known tablet
keeps being judged by
+ // its last known policy, an unknown one by the deadline its blocks
were written with.
+ g_tablet_cache_policy_fetch_failed_num << 1;
+ LOG_EVERY_N(WARNING, 100) << "failed to fetch file cache ttl policy of
tablet " << tablet_id
+ << ": " << st;
+ _retry_at[tablet_id] = UnixSeconds() + kFetchRetryIntervalS;
+ std::lock_guard due_lock(_due_mutex);
+ _due.insert(batch.begin() + i + 1, batch.end());
+ break;
+ }
+}
+
+void TabletCachePolicyRegistry::register_cache() {
+ std::lock_guard lock(_thread_mutex);
+ if (_open_caches++ > 0) {
+ return;
+ }
+ {
+ std::lock_guard stop_lock(_stop_mutex);
+ _stop_refresh = false;
+ }
+ _refresh_thread = std::thread(&TabletCachePolicyRegistry::refresh_loop,
this);
+}
+
+void TabletCachePolicyRegistry::unregister_cache() {
+ std::lock_guard lock(_thread_mutex);
+ DCHECK_GT(_open_caches, 0);
+ if (--_open_caches > 0) {
+ return;
+ }
+ {
+ std::lock_guard stop_lock(_stop_mutex);
+ _stop_refresh = true;
+ }
+ _stop_cv.notify_all();
+ _refresh_thread.join();
+}
+
+void TabletCachePolicyRegistry::refresh_loop() {
+ Thread::set_self_name("file_cache_tablet_policy_refresh");
+ while (true) {
+ {
+ std::unique_lock stop_lock(_stop_mutex);
+ if (_stop_cv.wait_for(stop_lock, kRefreshInterval, [this] { return
_stop_refresh; })) {
+ return;
+ }
+ }
+
refresh(static_cast<size_t>(config::file_cache_tablet_policy_refresh_batch));
+ }
+}
+
+void TabletCachePolicyRegistry::add_store(CacheBlockMetaStore* store) {
+ std::lock_guard lock(_stores_mutex);
+ _stores.push_back(store);
Review Comment:
[P2] Seed new cache paths with already-known policies
When startup adds an empty cache path B alongside path A, A can restore a
policy that never gets copied into B: restore() does not persist it,
add_store() only registers the pointer, and unchanged upserts skip persistence.
This matters for peer fills, whose context now carries the tablet id but still
has expiration_time=0; B can therefore hold TTL-protected blocks with a zero
persisted fallback deadline. If a later configured restart retains B while
removing/replacing A, those blocks lose protection until a remote fetch despite
the previously known policy. Copy the merged restored policies to every store,
and synchronize store registration with policy updates.
--
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]
