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 e2a836ec154 branch-4.2: [fix](file-cache) recycle busy blocks that
were marked deleting #68348 (#69141)
e2a836ec154 is described below
commit e2a836ec154c94d069bca68be469802c7d8b62b0
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Oct 16 09:28:56 2026 +0800
branch-4.2: [fix](file-cache) recycle busy blocks that were marked deleting
#68348 (#69141)
Cherry-picked from #68348
Co-authored-by: Xin Liao <[email protected]>
---
be/src/io/cache/block_file_cache.cpp | 107 +++++-
be/src/io/cache/block_file_cache.h | 29 +-
be/test/io/cache/block_file_cache_test.cpp | 558 +++++++++++++++++++++++++++++
3 files changed, 683 insertions(+), 11 deletions(-)
diff --git a/be/src/io/cache/block_file_cache.cpp
b/be/src/io/cache/block_file_cache.cpp
index c5ebc7cdbfb..e1c7fa3aad1 100644
--- a/be/src/io/cache/block_file_cache.cpp
+++ b/be/src/io/cache/block_file_cache.cpp
@@ -406,6 +406,10 @@ BlockFileCache::BlockFileCache(const std::string&
cache_base_path,
_cache_base_path.c_str(), "file_cache_lru_dump_latency_us");
_recycle_keys_length_recorder = std::make_shared<bvar::LatencyRecorder>(
_cache_base_path.c_str(), "file_cache_recycle_keys_length");
+ _deleting_block_count_metrics = std::make_shared<bvar::Status<size_t>>(
+ _cache_base_path.c_str(), "file_cache_deleting_block_count", 0);
+ _recycle_deleting_blocks_metrics = std::make_shared<bvar::Adder<size_t>>(
+ _cache_base_path.c_str(), "file_cache_recycle_deleting_blocks");
_need_update_lru_blocks_length_recorder =
std::make_shared<bvar::LatencyRecorder>(
_cache_base_path.c_str(),
"file_cache_need_update_lru_blocks_length");
_need_update_lru_blocks_produce_metrics =
std::make_shared<bvar::Adder<size_t>>(
@@ -993,6 +997,63 @@ bool BlockFileCache::is_block_deleting(const
FileBlockSPtr& block) const {
return block->_is_deleting;
}
+template <class T>
+ requires IsXLock<T>
+void BlockFileCache::mark_cell_deleting(FileBlockCell& cell, T& /* cache_lock
*/) {
+ cell.file_block->set_deleting();
+ // The block stays in _files until its last reference is dropped. Remember
it so that
+ // run_background_gc() can finish the job if that reference is not
released through
+ // FileBlock::release_cache_reference(). remove() forgets it again when it
erases the cell.
+ _deleting_blocks.emplace(cell.file_block->get_hash_value(),
cell.file_block->offset());
+}
+
+size_t BlockFileCache::recycle_deleting_blocks(size_t batch_limit) {
+ if (batch_limit == 0) {
+ return 0;
+ }
+ size_t recycled = 0;
+ size_t pending = 0;
+ {
+ SCOPED_CACHE_LOCK(_mutex, this);
+ std::vector<FileBlockSPtr> to_remove;
+ for (const auto& [hash, offset] : _deleting_blocks) {
+ if (to_remove.size() >= batch_limit) {
+ break;
+ }
+ auto* cell = get_cell(hash, offset, cache_lock);
+ if (cell == nullptr || !cell->file_block->is_deleting())
[[unlikely]] {
+ // Cannot happen: remove() erases the key together with the
cell.
+ DCHECK(false) << "stale deleting block, hash=" <<
hash.to_string()
+ << " offset=" << offset;
+ continue;
+ }
+ if (!cell->releasable() ||
+ cell->file_block->state_unsafe() ==
FileBlock::State::DOWNLOADING) {
+ // Still in use, or still being downloaded. Look at it again
in the next round.
+ continue;
+ }
+ to_remove.push_back(cell->file_block);
+ }
+ // remove() erases the key of every block it removes, so it must not
run while
+ // _deleting_blocks is being iterated above.
+ for (auto& file_block : to_remove) {
+ std::lock_guard block_lock(file_block->_mutex);
+ // Deleting the file itself happens asynchronously: the storage
must not be touched
+ // while the cache lock is held.
+ remove(file_block, cache_lock, block_lock, false);
+ ++recycled;
+ }
+ pending = _deleting_blocks.size();
+ }
+ if (recycled > 0) {
+ *_recycle_deleting_blocks_metrics << recycled;
+ LOG(INFO) << "recycled deleting blocks, path=" << _cache_base_path <<
" num=" << recycled
+ << " pending=" << pending;
+ }
+ _deleting_block_count_metrics->set_value(pending);
+ return recycled;
+}
+
std::string BlockFileCache::clear_file_cache_async() {
return clear_file_cache_impl(false);
}
@@ -1024,6 +1085,11 @@ std::string BlockFileCache::clear_file_cache_impl(bool
sync_remove) {
}
}
+ // Drop the pending LRU updates before the scan below, not after it:
every block still
+ // queued there carries an extra reference that would make the block
look busy and send
+ // it down the wait-for-recycle path for no reason.
+ clear_need_update_lru_blocks();
+
// Do not erase while walking _files above: remove() may erase the
current map element.
//
// sync_remove only changes how already releasable DOWNLOADED blocks
are deleted from
@@ -1032,8 +1098,9 @@ std::string BlockFileCache::clear_file_cache_impl(bool
sync_remove) {
for (auto& cell : deleting_cells) {
if (!cell->releasable()) {
LOG(INFO) << "cell is not releasable, hash="
+ << cell->file_block->get_hash_value().to_string()
<< " offset=" << cell->file_block->offset();
- cell->file_block->set_deleting();
+ mark_cell_deleting(*cell, cache_lock);
++num_cells_wait_recycle;
continue;
}
@@ -1044,7 +1111,6 @@ std::string BlockFileCache::clear_file_cache_impl(bool
sync_remove) {
++num_cells_to_delete;
}
}
- clear_need_update_lru_blocks();
}
std::stringstream ss;
@@ -1321,7 +1387,7 @@ size_t BlockFileCache::try_release() {
if (cell.releasable()) {
trash.emplace_back(&cell);
} else {
- cell.file_block->set_deleting();
+ mark_cell_deleting(cell, cache_lock);
}
}
}
@@ -1554,7 +1620,7 @@ void BlockFileCache::remove_if_cached(const
UInt128Wrapper& file_key) {
if (cell.releasable()) {
to_remove.push_back(&cell);
} else {
- cell.file_block->set_deleting();
+ mark_cell_deleting(cell, cache_lock);
}
}
remove_file_blocks(to_remove, cache_lock, true, reason);
@@ -1579,7 +1645,7 @@ void BlockFileCache::remove_if_cached_async(const
UInt128Wrapper& file_key) {
if (cell.releasable()) {
to_remove.push_back(&cell);
} else {
- cell.file_block->set_deleting();
+ mark_cell_deleting(cell, cache_lock);
}
}
remove_file_blocks(to_remove, cache_lock, false, reason);
@@ -1805,7 +1871,8 @@ void BlockFileCache::remove(FileBlockSPtr file_block, T&
cache_lock, U& block_lo
auto* cell = get_cell(hash, offset, cache_lock);
file_block->cell = nullptr;
// Holder cleanup can race with prior cache metadata cleanup. In that case,
- // skip the duplicate remove instead of touching a detached or replaced
cell.
+ // skip the duplicate remove instead of touching a detached or replaced
cell. That includes
+ // _deleting_blocks: a key found there now belongs to the block that
replaced this one.
if (cell == nullptr) {
LOG(WARNING) << "remove skipped because cache cell is missing. hash="
<< hash.to_string()
<< " offset=" << offset << " size=" <<
file_block->range().size()
@@ -1835,7 +1902,8 @@ void BlockFileCache::remove(FileBlockSPtr file_block, T&
cache_lock, U& block_lo
: "<null>");
return;
}
- DCHECK(cell->queue_iterator);
+ // No iterator means the block was already taken out of its queue by an
earlier remove()
+ // that could not erase the cell because the block was still downloading.
if (cell->queue_iterator) {
auto& queue = get_queue(file_block->cache_type());
queue.remove(*cell->queue_iterator, cache_lock);
@@ -1891,7 +1959,11 @@ void BlockFileCache::remove(FileBlockSPtr file_block, T&
cache_lock, U& block_lo
}
}
} else if (state == FileBlock::State::DOWNLOADING) {
- file_block->set_deleting();
+ // The block is already out of its LRU queue, but the cell stays in
_files until the
+ // downloader is done with it. Forget the iterator: it points at a
freed queue entry now,
+ // and whoever finally erases this cell must not touch it again.
+ cell->queue_iterator.reset();
+ mark_cell_deleting(*cell, cache_lock);
return;
} else {
*_evict_not_downloaded_size_metrics << file_block->range().size();
@@ -1904,6 +1976,10 @@ void BlockFileCache::remove(FileBlockSPtr file_block, T&
cache_lock, U& block_lo
auto it = _files.find(hash);
if (it != _files.end()) {
it->second.erase(file_block->offset());
+ if (file_block->is_deleting()) {
+ // The key leaves _deleting_blocks together with its cell.
+ _deleting_blocks.erase(AccessKeyAndOffset(hash, offset));
+ }
if (it->second.empty()) {
_files.erase(hash);
}
@@ -2116,8 +2192,12 @@ void BlockFileCache::change_cache_type(const
UInt128Wrapper& hash, size_t offset
auto& file_blocks = iter->second;
if (auto cell_it = file_blocks.find(offset); cell_it !=
file_blocks.end()) {
FileBlockCell& cell = cell_it->second;
+ // No iterator means remove() already took the block out of its
queue while it was
+ // downloading. The block is deleting, so keep it out of every
queue.
+ if (!cell.queue_iterator) {
+ return;
+ }
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,
@@ -2183,7 +2263,7 @@ std::string BlockFileCache::reset_capacity(size_t
new_capacity) {
queue_released += entry_size;
auto* cell = get_cell(entry_key, entry_offset, cache_lock);
if (!cell->releasable()) {
- cell->file_block->set_deleting();
+ mark_cell_deleting(*cell, cache_lock);
continue;
}
to_evict.push_back(cell);
@@ -2421,6 +2501,10 @@ void BlockFileCache::run_background_gc() {
}
}
+ // Blocks that were busy when they were marked deleting, but are not
referenced any
+ // more, are only reclaimed here.
+ recycle_deleting_blocks(batch_limit);
+
while (batch_count < batch_limit && _recycle_keys.try_dequeue(key)) {
int64_t duration_ns = 0;
Status st;
@@ -2506,6 +2590,9 @@ void BlockFileCache::run_background_block_lru_update() {
update_block_lru(block, cache_lock);
}
}
+ // Drop the references now instead of at the top of the next round:
holding them across
+ // the sleep would keep every block in this batch looking busy for a
whole interval.
+ batch.clear();
*_update_lru_blocks_latency_us << (duration_ns / 1000);
*_need_update_lru_blocks_length_recorder <<
_need_update_lru_blocks.size();
}
diff --git a/be/src/io/cache/block_file_cache.h
b/be/src/io/cache/block_file_cache.h
index 9b32b4764b4..6cb9513ff3b 100644
--- a/be/src/io/cache/block_file_cache.h
+++ b/be/src/io/cache/block_file_cache.h
@@ -32,6 +32,7 @@
#include <optional>
#include <thread>
#include <unordered_map>
+#include <unordered_set>
#include <vector>
#include "io/cache/block_file_cache_ttl_mgr.h"
@@ -419,7 +420,8 @@ public:
private:
// Shared scan used by both clear modes. It keeps the FileBlock holder
lifecycle intact:
// releasable blocks are removed immediately, while blocks held by readers
are only marked
- // deleting and are later removed by FileBlocksHolder destruction.
+ // deleting and are later removed by FileBlocksHolder destruction or, when
the last holder
+ // does not release through FileBlock::release_cache_reference(), by
run_background_gc().
std::string clear_file_cache_impl(bool sync_remove);
LRUQueue& get_queue(FileCacheType type);
@@ -536,6 +538,17 @@ private:
void clear_need_update_lru_blocks();
+ // Marks a cell that cannot be erased yet as deleting and remembers it in
_deleting_blocks, so
+ // that the block is still reclaimed when its last reference is dropped
without going through
+ // FileBlock::release_cache_reference(). This is the only place that sets
a block deleting.
+ template <class T>
+ requires IsXLock<T>
+ void mark_cell_deleting(FileBlockCell& cell, T& cache_lock);
+
+ // Removes up to batch_limit blocks that were marked deleting and are not
referenced any
+ // more. Returns how many of them were removed.
+ size_t recycle_deleting_blocks(size_t batch_limit);
+
// info
std::string _cache_base_path;
size_t _capacity = 0;
@@ -581,6 +594,18 @@ private:
// keys for async remove
RecycleFileCacheKeys _recycle_keys;
+ // Blocks that were busy when something asked to delete them. They keep
sitting in _files
+ // until their last reference goes away. Only FileBlocksHolder and
FileBlocksProbeResult
+ // release through FileBlock::release_cache_reference(), which is the one
path that notices
+ // is_deleting and removes the block; every other reference holder just
drops its shared_ptr
+ // and would strand the block here forever. run_background_gc() rechecks
these candidates and
+ // removes the ones that became releasable.
+ //
+ // Invariant: holds exactly the keys of the cells in _files whose block is
deleting. A key is
+ // added by mark_cell_deleting() and erased by remove() together with its
cell, so a block
+ // created later at the same offset is never taken for the one that was
marked.
+ std::unordered_set<AccessKeyAndOffset, KeyAndOffsetHash> _deleting_blocks;
+
std::unique_ptr<LRUQueueRecorder> _lru_recorder;
std::unique_ptr<CacheLRUDumper> _lru_dumper;
std::unique_ptr<BlockFileCacheTtlMgr> _ttl_mgr;
@@ -648,6 +673,8 @@ private:
std::shared_ptr<bvar::LatencyRecorder> _storage_async_remove_latency_us;
std::shared_ptr<bvar::LatencyRecorder> _evict_in_advance_latency_us;
std::shared_ptr<bvar::LatencyRecorder> _recycle_keys_length_recorder;
+ std::shared_ptr<bvar::Status<size_t>> _deleting_block_count_metrics;
+ std::shared_ptr<bvar::Adder<size_t>> _recycle_deleting_blocks_metrics;
std::shared_ptr<bvar::LatencyRecorder> _update_lru_blocks_latency_us;
std::shared_ptr<bvar::LatencyRecorder>
_need_update_lru_blocks_length_recorder;
std::shared_ptr<bvar::Adder<size_t>>
_need_update_lru_blocks_produce_metrics;
diff --git a/be/test/io/cache/block_file_cache_test.cpp
b/be/test/io/cache/block_file_cache_test.cpp
index dbd0b623b97..37cc2068404 100644
--- a/be/test/io/cache/block_file_cache_test.cpp
+++ b/be/test/io/cache/block_file_cache_test.cpp
@@ -3698,6 +3698,564 @@ TEST_F(BlockFileCacheTest,
clear_file_cache_sync_removes_releasable_blocks_synch
}
}
+TEST_F(BlockFileCacheTest,
remove_downloading_block_forgets_its_queue_iterator) {
+ std::string my_cache_path = caches_dir / "remove_downloading_block" / "";
+ if (fs::exists(my_cache_path)) {
+ fs::remove_all(my_cache_path);
+ }
+ fs::create_directories(my_cache_path);
+
+ constexpr size_t block_size = 4;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 64;
+ settings.query_queue_elements = 16;
+ settings.index_queue_size = 64;
+ settings.index_queue_elements = 16;
+ settings.disposable_queue_size = 64;
+ settings.disposable_queue_elements = 16;
+ settings.capacity = 192;
+ settings.max_file_block_size = 64;
+ settings.max_query_cache_size = 64;
+
+ io::BlockFileCache cache(my_cache_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ wait_until_cache_ready(cache);
+
+ io::CacheContext context;
+ context.cache_type = io::FileCacheType::NORMAL;
+ auto key = io::BlockFileCache::hash("remove_downloading_block");
+
+ io::FileBlockSPtr block;
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ auto* cell = cache.add_cell(key, context, 0, block_size,
io::FileBlock::State::DOWNLOADED,
+ cache_lock);
+ ASSERT_NE(cell, nullptr);
+ FileBlockTestAccessor::set_state(*cell->file_block,
io::FileBlock::State::DOWNLOADING);
+ FileBlockTestAccessor::set_downloader_id(*cell->file_block, 1234);
+ block = cell->file_block;
+ }
+ auto& queue = cache.get_queue(io::FileCacheType::NORMAL);
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ ASSERT_EQ(queue.get_elements_num(cache_lock), 1);
+ }
+
+ // A block that is still downloading is taken out of its LRU queue, but
its cell stays in
+ // _files, so the iterator it holds points at a freed queue entry from
here on.
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ std::lock_guard<std::mutex> block_lock(block->_mutex);
+ cache.remove(block, cache_lock, block_lock, false);
+ }
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ auto* cell = cache.get_cell(key, 0, cache_lock);
+ ASSERT_NE(cell, nullptr);
+ EXPECT_TRUE(cell->file_block->is_deleting());
+ EXPECT_EQ(cache._deleting_blocks.count(io::AccessKeyAndOffset(key,
0)), 1);
+ EXPECT_FALSE(cell->queue_iterator.has_value());
+ EXPECT_EQ(queue.get_elements_num(cache_lock), 0);
+ EXPECT_EQ(queue.get_capacity(cache_lock), 0);
+ }
+
+ // The download finishes and the last holder releases the block: this
second remove() is the
+ // one that erases the cell, and it must not touch the queue entry again.
+ FileBlockTestAccessor::set_state(*block, io::FileBlock::State::DOWNLOADED);
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ std::lock_guard<std::mutex> block_lock(block->_mutex);
+ cache.remove(block, cache_lock, block_lock, false);
+ }
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ EXPECT_EQ(cache.get_cell(key, 0, cache_lock), nullptr);
+ EXPECT_TRUE(cache._deleting_blocks.empty());
+ EXPECT_EQ(queue.get_elements_num(cache_lock), 0);
+ EXPECT_EQ(queue.get_capacity(cache_lock), 0);
+ }
+ EXPECT_EQ(cache._cur_cache_size, 0);
+
+ block.reset();
+ if (fs::exists(my_cache_path)) {
+ fs::remove_all(my_cache_path);
+ }
+}
+
+TEST_F(BlockFileCacheTest,
change_cache_type_keeps_removed_downloading_block_out_of_queues) {
+ std::string my_cache_path = caches_dir /
"change_cache_type_removed_downloading_block" / "";
+ if (fs::exists(my_cache_path)) {
+ fs::remove_all(my_cache_path);
+ }
+ fs::create_directories(my_cache_path);
+
+ constexpr size_t block_size = 4;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 64;
+ settings.query_queue_elements = 16;
+ settings.index_queue_size = 64;
+ settings.index_queue_elements = 16;
+ settings.disposable_queue_size = 64;
+ settings.disposable_queue_elements = 16;
+ settings.capacity = 192;
+ settings.max_file_block_size = 64;
+ settings.max_query_cache_size = 64;
+
+ io::BlockFileCache cache(my_cache_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ wait_until_cache_ready(cache);
+
+ io::CacheContext context;
+ context.cache_type = io::FileCacheType::NORMAL;
+ auto key =
io::BlockFileCache::hash("change_cache_type_removed_downloading_block");
+
+ io::FileBlockSPtr block;
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ auto* cell = cache.add_cell(key, context, 0, block_size,
io::FileBlock::State::DOWNLOADED,
+ cache_lock);
+ ASSERT_NE(cell, nullptr);
+ FileBlockTestAccessor::set_state(*cell->file_block,
io::FileBlock::State::DOWNLOADING);
+ FileBlockTestAccessor::set_downloader_id(*cell->file_block, 1234);
+ block = cell->file_block;
+ }
+ // remove() takes a downloading block out of its LRU queue but leaves the
cell in _files.
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ std::lock_guard<std::mutex> block_lock(block->_mutex);
+ cache.remove(block, cache_lock, block_lock, false);
+ }
+
+ // The download finishes, and the block is converted to another cache type
before its holder
+ // lets go of it, as BlockFileCacheTtlMgr does. It must stay out of every
queue.
+ FileBlockTestAccessor::set_state(*block, io::FileBlock::State::DOWNLOADED);
+ auto& normal_queue = cache.get_queue(io::FileCacheType::NORMAL);
+ auto& ttl_queue = cache.get_queue(io::FileCacheType::TTL);
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ cache.change_cache_type(key, 0, io::FileCacheType::TTL, cache_lock);
+ auto* cell = cache.get_cell(key, 0, cache_lock);
+ ASSERT_NE(cell, nullptr);
+ EXPECT_FALSE(cell->queue_iterator.has_value());
+ EXPECT_EQ(normal_queue.get_elements_num(cache_lock), 0);
+ EXPECT_EQ(ttl_queue.get_elements_num(cache_lock), 0);
+ }
+
+ // The holder releases the block: erasing the cell must not touch any
queue either.
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ std::lock_guard<std::mutex> block_lock(block->_mutex);
+ cache.remove(block, cache_lock, block_lock, false);
+ }
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ EXPECT_EQ(cache.get_cell(key, 0, cache_lock), nullptr);
+ EXPECT_EQ(normal_queue.get_elements_num(cache_lock), 0);
+ EXPECT_EQ(normal_queue.get_capacity(cache_lock), 0);
+ EXPECT_EQ(ttl_queue.get_elements_num(cache_lock), 0);
+ EXPECT_EQ(ttl_queue.get_capacity(cache_lock), 0);
+ }
+ EXPECT_EQ(cache._cur_cache_size, 0);
+
+ block.reset();
+ if (fs::exists(my_cache_path)) {
+ fs::remove_all(my_cache_path);
+ }
+}
+
+TEST_F(BlockFileCacheTest,
recycle_deleting_blocks_reclaims_stranded_busy_blocks) {
+ std::string my_cache_path = caches_dir / "recycle_deleting_blocks" / "";
+ if (fs::exists(my_cache_path)) {
+ fs::remove_all(my_cache_path);
+ }
+ fs::create_directories(my_cache_path);
+
+ constexpr int num_blocks = 8;
+ constexpr size_t block_size = 1;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = num_blocks * block_size * 2;
+ settings.query_queue_elements = num_blocks + 16;
+ settings.index_queue_size = num_blocks * block_size * 2;
+ settings.index_queue_elements = num_blocks + 16;
+ settings.disposable_queue_size = num_blocks * block_size * 2;
+ settings.disposable_queue_elements = num_blocks + 16;
+ settings.capacity = num_blocks * block_size * 2;
+ settings.max_file_block_size = 64;
+ settings.max_query_cache_size = num_blocks * block_size * 2;
+
+ int64_t origin_qps_limit = config::file_cache_remove_block_qps_limit;
+ // Give the background GC thread no removal budget: the explicit
recycle_deleting_blocks()
+ // calls below must be the only thing reclaiming blocks, otherwise the
assertions race with
+ // it. The GC interval stays short, so ~BlockFileCache() never waits long
for that thread.
+ config::file_cache_remove_block_qps_limit = 0;
+ Defer restore_qps_limit {[&] { config::file_cache_remove_block_qps_limit =
origin_qps_limit; }};
+
+ io::BlockFileCache cache(my_cache_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ wait_until_cache_ready(cache);
+
+ io::CacheContext context;
+ context.cache_type = io::FileCacheType::NORMAL;
+ auto key = io::BlockFileCache::hash("recycle_deleting_blocks");
+
+ // A plain shared_ptr, standing in for the reference holders that do not
release through
+ // FileBlock::release_cache_reference(): the background LRU update queue,
the block map of
+ // CachedRemoteFileReader, the block vector of BlockFileCacheTtlMgr.
+ std::vector<io::FileBlockSPtr> busy_refs;
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ for (int i = 0; i < num_blocks; ++i) {
+ auto* cell = cache.add_cell(key, context, i * block_size,
block_size,
+ io::FileBlock::State::DOWNLOADED,
cache_lock);
+ ASSERT_NE(cell, nullptr);
+ busy_refs.push_back(cell->file_block);
+ }
+ }
+
+ std::atomic<int> storage_remove_calls {0};
+ auto sp = SyncPoint::get_instance();
+ SyncPoint::CallbackGuard guard;
+ sp->set_call_back(
+ "FSFileCacheStorage::remove",
+ [&storage_remove_calls](auto&& args) {
+ storage_remove_calls.fetch_add(1);
+ try_any_cast_ret<Status>(args)->second = true;
+ },
+ &guard);
+ sp->enable_processing();
+ Defer defer {[sp] {
+ sp->disable_processing();
+ sp->clear_all_call_backs();
+ }};
+
+ // Every block is busy, so clear() can only mark them and wait.
+ auto msg = cache.clear_file_cache_sync();
+ EXPECT_NE(msg.find("num_cells_wait_recycle=" +
std::to_string(num_blocks)), std::string::npos);
+ EXPECT_EQ(storage_remove_calls.load(), 0);
+ EXPECT_EQ(cache._cur_cache_size, num_blocks * block_size);
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ EXPECT_EQ(cache._deleting_blocks.size(), num_blocks);
+ for (int i = 0; i < num_blocks; ++i) {
+ auto* cell = cache.get_cell(key, i * block_size, cache_lock);
+ ASSERT_NE(cell, nullptr);
+ EXPECT_TRUE(cell->file_block->is_deleting());
+ }
+ }
+
+ // Blocks that are still referenced must survive the sweep.
+ EXPECT_EQ(cache.recycle_deleting_blocks(num_blocks), 0);
+ EXPECT_EQ(cache._cur_cache_size, num_blocks * block_size);
+
+ // Dropping the references this way never reaches
FileBlock::release_cache_reference(), so
+ // without the sweep below the blocks would stay in _files and on disk
forever.
+ busy_refs.clear();
+
+ EXPECT_EQ(cache.recycle_deleting_blocks(num_blocks), num_blocks);
+ EXPECT_EQ(cache._cur_cache_size, 0);
+ // The sweep runs under the cache lock, so it hands the files to the async
recycle queue
+ // that run_background_gc() drains right after.
+ EXPECT_EQ(cache._recycle_keys.size_approx(), num_blocks);
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ EXPECT_TRUE(cache._deleting_blocks.empty());
+ for (int i = 0; i < num_blocks; ++i) {
+ EXPECT_EQ(cache.get_cell(key, i * block_size, cache_lock),
nullptr);
+ }
+ }
+
+ // Nothing left to do, and no block is looked at twice.
+ EXPECT_EQ(cache.recycle_deleting_blocks(num_blocks), 0);
+
+ if (fs::exists(my_cache_path)) {
+ fs::remove_all(my_cache_path);
+ }
+}
+
+TEST_F(BlockFileCacheTest, recycle_deleting_blocks_honours_batch_limit) {
+ std::string my_cache_path = caches_dir / "recycle_deleting_blocks_batch" /
"";
+ if (fs::exists(my_cache_path)) {
+ fs::remove_all(my_cache_path);
+ }
+ fs::create_directories(my_cache_path);
+
+ constexpr int num_blocks = 10;
+ constexpr size_t block_size = 1;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = num_blocks * block_size * 2;
+ settings.query_queue_elements = num_blocks + 16;
+ settings.index_queue_size = num_blocks * block_size * 2;
+ settings.index_queue_elements = num_blocks + 16;
+ settings.disposable_queue_size = num_blocks * block_size * 2;
+ settings.disposable_queue_elements = num_blocks + 16;
+ settings.capacity = num_blocks * block_size * 2;
+ settings.max_file_block_size = 64;
+ settings.max_query_cache_size = num_blocks * block_size * 2;
+
+ int64_t origin_qps_limit = config::file_cache_remove_block_qps_limit;
+ // Give the background GC thread no removal budget: the explicit
recycle_deleting_blocks()
+ // calls below must be the only thing reclaiming blocks, otherwise the
assertions race with
+ // it. The GC interval stays short, so ~BlockFileCache() never waits long
for that thread.
+ config::file_cache_remove_block_qps_limit = 0;
+ Defer restore_qps_limit {[&] { config::file_cache_remove_block_qps_limit =
origin_qps_limit; }};
+
+ io::BlockFileCache cache(my_cache_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ wait_until_cache_ready(cache);
+
+ io::CacheContext context;
+ context.cache_type = io::FileCacheType::NORMAL;
+ auto key = io::BlockFileCache::hash("recycle_deleting_blocks_batch");
+
+ std::vector<io::FileBlockSPtr> busy_refs;
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ for (int i = 0; i < num_blocks; ++i) {
+ auto* cell = cache.add_cell(key, context, i * block_size,
block_size,
+ io::FileBlock::State::DOWNLOADED,
cache_lock);
+ ASSERT_NE(cell, nullptr);
+ busy_refs.push_back(cell->file_block);
+ }
+ }
+
+ cache.clear_file_cache_sync();
+ busy_refs.clear();
+
+ EXPECT_EQ(cache.recycle_deleting_blocks(0), 0);
+ EXPECT_EQ(cache.recycle_deleting_blocks(4), 4);
+ EXPECT_EQ(cache.recycle_deleting_blocks(4), 4);
+ EXPECT_EQ(cache.recycle_deleting_blocks(4), 2);
+ EXPECT_EQ(cache.recycle_deleting_blocks(4), 0);
+ EXPECT_EQ(cache._cur_cache_size, 0);
+
+ if (fs::exists(my_cache_path)) {
+ fs::remove_all(my_cache_path);
+ }
+}
+
+TEST_F(BlockFileCacheTest,
recycle_deleting_blocks_ignores_block_recreated_at_same_offset) {
+ std::string my_cache_path = caches_dir /
"recycle_deleting_blocks_recreated" / "";
+ if (fs::exists(my_cache_path)) {
+ fs::remove_all(my_cache_path);
+ }
+ fs::create_directories(my_cache_path);
+
+ constexpr size_t block_size = 4;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 64;
+ settings.query_queue_elements = 16;
+ settings.index_queue_size = 64;
+ settings.index_queue_elements = 16;
+ settings.disposable_queue_size = 64;
+ settings.disposable_queue_elements = 16;
+ settings.capacity = 192;
+ settings.max_file_block_size = 64;
+ settings.max_query_cache_size = 64;
+
+ int64_t origin_qps_limit = config::file_cache_remove_block_qps_limit;
+ // Give the background GC thread no removal budget: the explicit
recycle_deleting_blocks()
+ // calls below must be the only thing reclaiming blocks, otherwise the
assertions race with
+ // it. The GC interval stays short, so ~BlockFileCache() never waits long
for that thread.
+ config::file_cache_remove_block_qps_limit = 0;
+ Defer restore_qps_limit {[&] { config::file_cache_remove_block_qps_limit =
origin_qps_limit; }};
+
+ io::BlockFileCache cache(my_cache_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ wait_until_cache_ready(cache);
+
+ io::CacheContext context;
+ context.cache_type = io::FileCacheType::NORMAL;
+ auto key = io::BlockFileCache::hash("recycle_deleting_blocks_recreated");
+
+ // A reader holds the block while the cache is cleared, so the block is
only marked deleting.
+ io::FileBlocks old_blocks;
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ auto* cell = cache.add_cell(key, context, 0, block_size,
io::FileBlock::State::DOWNLOADED,
+ cache_lock);
+ ASSERT_NE(cell, nullptr);
+ old_blocks.push_back(cell->file_block);
+ }
+ cache.clear_file_cache_sync();
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ EXPECT_EQ(cache._deleting_blocks.size(), 1);
+ }
+
+ // The reader releases it through its holder, which removes the block and
its key.
+ { io::FileBlocksHolder holder(std::move(old_blocks)); }
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ EXPECT_EQ(cache.get_cell(key, 0, cache_lock), nullptr);
+ EXPECT_TRUE(cache._deleting_blocks.empty());
+ }
+
+ // The same offset is read again and cached by a new block before the next
GC round.
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ auto* cell = cache.add_cell(key, context, 0, block_size,
io::FileBlock::State::DOWNLOADED,
+ cache_lock);
+ ASSERT_NE(cell, nullptr);
+ EXPECT_FALSE(cell->file_block->is_deleting());
+ }
+
+ // The sweep must not take the new block for the one that was marked
deleting.
+ EXPECT_EQ(cache.recycle_deleting_blocks(16), 0);
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ EXPECT_NE(cache.get_cell(key, 0, cache_lock), nullptr);
+ }
+ EXPECT_EQ(cache._cur_cache_size, block_size);
+
+ if (fs::exists(my_cache_path)) {
+ fs::remove_all(my_cache_path);
+ }
+}
+
+TEST_F(BlockFileCacheTest, recycle_deleting_blocks_reclaims_empty_blocks) {
+ std::string my_cache_path = caches_dir / "recycle_deleting_blocks_empty" /
"";
+ if (fs::exists(my_cache_path)) {
+ fs::remove_all(my_cache_path);
+ }
+ fs::create_directories(my_cache_path);
+
+ constexpr size_t block_size = 4;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 64;
+ settings.query_queue_elements = 16;
+ settings.index_queue_size = 64;
+ settings.index_queue_elements = 16;
+ settings.disposable_queue_size = 64;
+ settings.disposable_queue_elements = 16;
+ settings.capacity = 192;
+ settings.max_file_block_size = 64;
+ settings.max_query_cache_size = 64;
+
+ int64_t origin_qps_limit = config::file_cache_remove_block_qps_limit;
+ // Give the background GC thread no removal budget: the explicit
recycle_deleting_blocks()
+ // calls below must be the only thing reclaiming blocks, otherwise the
assertions race with
+ // it. The GC interval stays short, so ~BlockFileCache() never waits long
for that thread.
+ config::file_cache_remove_block_qps_limit = 0;
+ Defer restore_qps_limit {[&] { config::file_cache_remove_block_qps_limit =
origin_qps_limit; }};
+
+ io::BlockFileCache cache(my_cache_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ wait_until_cache_ready(cache);
+
+ io::CacheContext context;
+ context.cache_type = io::FileCacheType::NORMAL;
+ auto key = io::BlockFileCache::hash("recycle_deleting_blocks_empty");
+
+ // A block nobody started to download, referenced by something that is not
a holder.
+ io::FileBlockSPtr busy_ref;
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ auto* cell = cache.add_cell(key, context, 0, block_size,
io::FileBlock::State::EMPTY,
+ cache_lock);
+ ASSERT_NE(cell, nullptr);
+ busy_ref = cell->file_block;
+ }
+ cache.clear_file_cache_sync();
+ busy_ref.reset();
+
+ // Nobody will ever download it, so the sweep has to remove it.
+ EXPECT_EQ(cache.recycle_deleting_blocks(16), 1);
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ EXPECT_EQ(cache.get_cell(key, 0, cache_lock), nullptr);
+ EXPECT_TRUE(cache._deleting_blocks.empty());
+ }
+ EXPECT_EQ(cache._cur_cache_size, 0);
+
+ if (fs::exists(my_cache_path)) {
+ fs::remove_all(my_cache_path);
+ }
+}
+
+TEST_F(BlockFileCacheTest,
recycle_deleting_blocks_reclaims_block_removed_while_downloading) {
+ std::string my_cache_path = caches_dir /
"recycle_deleting_blocks_downloading" / "";
+ if (fs::exists(my_cache_path)) {
+ fs::remove_all(my_cache_path);
+ }
+ fs::create_directories(my_cache_path);
+
+ constexpr size_t block_size = 4;
+ io::FileCacheSettings settings;
+ settings.query_queue_size = 64;
+ settings.query_queue_elements = 16;
+ settings.index_queue_size = 64;
+ settings.index_queue_elements = 16;
+ settings.disposable_queue_size = 64;
+ settings.disposable_queue_elements = 16;
+ settings.capacity = 192;
+ settings.max_file_block_size = 64;
+ settings.max_query_cache_size = 64;
+
+ int64_t origin_qps_limit = config::file_cache_remove_block_qps_limit;
+ // Give the background GC thread no removal budget: the explicit
recycle_deleting_blocks()
+ // calls below must be the only thing reclaiming blocks, otherwise the
assertions race with
+ // it. The GC interval stays short, so ~BlockFileCache() never waits long
for that thread.
+ config::file_cache_remove_block_qps_limit = 0;
+ Defer restore_qps_limit {[&] { config::file_cache_remove_block_qps_limit =
origin_qps_limit; }};
+
+ io::BlockFileCache cache(my_cache_path, settings);
+ ASSERT_TRUE(cache.initialize());
+ wait_until_cache_ready(cache);
+
+ io::CacheContext context;
+ context.cache_type = io::FileCacheType::NORMAL;
+ auto key = io::BlockFileCache::hash("recycle_deleting_blocks_downloading");
+
+ io::FileBlockSPtr block;
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ auto* cell = cache.add_cell(key, context, 0, block_size,
io::FileBlock::State::DOWNLOADED,
+ cache_lock);
+ ASSERT_NE(cell, nullptr);
+ FileBlockTestAccessor::set_state(*cell->file_block,
io::FileBlock::State::DOWNLOADING);
+ FileBlockTestAccessor::set_downloader_id(*cell->file_block, 1234);
+ block = cell->file_block;
+ }
+
+ // remove() cannot erase a block that is still downloading, it only marks
it deleting.
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ std::lock_guard<std::mutex> block_lock(block->_mutex);
+ cache.remove(block, cache_lock, block_lock, false);
+ }
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ EXPECT_EQ(cache._deleting_blocks.count(io::AccessKeyAndOffset(key,
0)), 1);
+ }
+
+ // Drop the reference the way a non-holder does. The sweep still leaves
the block to its
+ // downloader.
+ block.reset();
+ EXPECT_EQ(cache.recycle_deleting_blocks(16), 0);
+
+ // Once the download is done nobody would remove the block but the sweep.
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ auto* cell = cache.get_cell(key, 0, cache_lock);
+ ASSERT_NE(cell, nullptr);
+ FileBlockTestAccessor::set_state(*cell->file_block,
io::FileBlock::State::DOWNLOADED);
+ }
+ EXPECT_EQ(cache.recycle_deleting_blocks(16), 1);
+ {
+ std::lock_guard<std::mutex> cache_lock(cache._mutex);
+ auto& queue = cache.get_queue(io::FileCacheType::NORMAL);
+ EXPECT_EQ(cache.get_cell(key, 0, cache_lock), nullptr);
+ EXPECT_TRUE(cache._deleting_blocks.empty());
+ EXPECT_EQ(queue.get_elements_num(cache_lock), 0);
+ EXPECT_EQ(queue.get_capacity(cache_lock), 0);
+ }
+ EXPECT_EQ(cache._cur_cache_size, 0);
+
+ if (fs::exists(my_cache_path)) {
+ fs::remove_all(my_cache_path);
+ }
+}
+
TEST_F(BlockFileCacheTest,
clear_file_cache_sync_factory_rejects_concurrent_sync_clear) {
std::string my_cache_path = caches_dir /
"clear_file_cache_sync_factory_busy" / "";
if (fs::exists(my_cache_path)) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
