github-actions[bot] commented on code in PR #68991:
URL: https://github.com/apache/doris/pull/68991#discussion_r4273711125
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -2106,25 +2172,225 @@ 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. A pass under way is never
restarted; what
+ // changes meanwhile starts the next one.
+ 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;
+ // What the next pass is compared with.
+ 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;
+ }
+ }
+ // The backstop counts as done once every queue has a pass; a queue still
busy with an older
+ // pass gets its own when that one ends.
+ 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();
+ }
+ // The cursor lives on the queue, so the pass resumes there after the
cache lock was released.
+ 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;
+ // Past the entry before it may move.
+ ++cursor;
+ auto* cell = get_cell(entry.hash, entry.offset, cache_lock);
+ DCHECK(cell != nullptr);
+ const auto& block = *cell->file_block;
+ // Where the block belongs now: the TTL queue while protected, else
its class queue.
+ 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();
+ }
+ }
+ // Moved after the walk, all of them behind the cursor.
+ for (auto& [cell, want] : moves) {
+ move_cell_to_queue(*cell, want, cache_lock);
+ }
+ // The pass is over; start_queue_sweeps() decides whether another one is
due.
+ if (pass_left == 0 || queue.sweep_cursor == queue.end()) {
+ queue.sweep_cursor.reset();
+ active = false;
+ if (type == FileCacheType::TTL) {
+
_ttl_legacy_v2_protected_bytes->set_value(_queue_sweep.ttl_pass_legacy_bytes);
+ }
+ }
+ return walked;
+}
+
+// How often a cache path tells the policy registry which tablets it holds
blocks of.
+constexpr int64_t kReportCachedTabletsIntervalS = 60;
+
+void BlockFileCache::run_background_queue_sweep() {
+ Thread::set_self_name("run_background_queue_sweep");
+ while (!_close) {
+ {
+ std::unique_lock close_lock(_close_mtx);
+ _close_cv.wait_for(close_lock, std::chrono::milliseconds(
+
config::file_cache_queue_sweep_interval_ms));
+ if (_close) {
+ break;
+ }
+ }
+ // Blocks being loaded are put in the right queue as they are loaded.
+ if (!_async_open_done) {
Review Comment:
[P2] Apply live protection checks during asynchronous startup
A policy can gain an active TTL after its downloaded NORMAL/INDEX cells have
been loaded but while other metadata batches are still loading (for example, a
foreground tablet load/meta sync corrects a restored zero policy). This guard
prevents relocating those cells until loading finishes. Under disk/inode
pressure, a concurrent cache miss instead takes `try_reserve()`'s early
`try_reserve_during_async_load()` branch, which removes the first releasable
class-queue cells without consulting `is_protected()`. It therefore evicts an
actively protected cold block even when later unprotected blocks could provide
the needed space, and bypasses protected-eviction accounting. Apply the same
live protection/relocation checks to startup reservation, and cover a policy
gain between loader batches under disk pressure.
--
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]
