This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 4c13cb1da9d [fix](file-cache) recycle busy blocks that were marked
deleting (#68348)
4c13cb1da9d is described below
commit 4c13cb1da9d4599bfffb720357795f8ef44c1a50
Author: Xin Liao <[email protected]>
AuthorDate: Thu Oct 15 20:46:42 2026 +0800
[fix](file-cache) recycle busy blocks that were marked deleting (#68348)
### What problem does this PR solve?
Related Jira: CORE-6174
When `clear_file_cache_sync/async`, `remove_if_cached[_async]`,
`try_release` or `reset_capacity` run into a block that is still
referenced, they do not remove it. They only call
`FileBlock::set_deleting()` and leave the block in `_files`, expecting
the last reference holder to remove it when it releases.
The only code that honours that expectation is
`FileBlock::release_cache_reference()`, which runs from
`~FileBlocksHolder` and `~FileBlocksProbeResult`. Every other reference
holder simply drops its `shared_ptr`:
- the `batch` of `run_background_block_lru_update()`, which is declared
outside the loop and therefore keeps its references alive across a whole
sleep interval (5s by default);
- `clear_need_update_lru_blocks()`, which `clear_file_cache_impl()`
called *after* the scan, so a block queued there was first counted as
busy and then had its reference dropped;
- the `_cache_file_readers` map of `CachedRemoteFileReader`;
- the block vector of `BlockFileCacheTtlMgr::reconcile_tablet_blocks()`.
If one of those held the last extra reference, the block ended up
stranded: marked deleting, no longer referenced, but still in `_files`,
still in its LRU queue and still on disk. Nothing ever removed it,
because no background thread looks for deleting blocks —
`run_background_gc()` only drains `_recycle_keys`, which is filled by
`remove()`. The block only disappeared if it happened to be read again
(`get_impl()` does not filter deleting cells, so a later holder release
cleans it up), or if eviction or `try_release()` picked it up.
So a synchronous clear could return `status=OK` with
`num_cells_wait_recycle=1` and leave that block in `file_cache_info` and
on disk indefinitely. That is what a regression test caught: no query
touched the table after the clear, and the same 2599-byte TTL block was
still there more than 10 minutes later.
### Release note
[fix](file-cache) Fix file cache blocks that were busy during a clear
never being reclaimed
### What is changed and how it works?
Make the reclaim independent of which code path drops the last
reference:
- New `_deleting_blocks` set holds exactly the keys of the cells in
`_files` whose block is deleting. A key is added by the new
`mark_cell_deleting()` helper, which is now the only place that sets a
block deleting (every busy-block call site and the downloading branch of
`remove()` go through it), and is erased by `remove()` together with its
cell. Keeping the set in step with `_files`, instead of trimming it
lazily, means a block cached again at the same offset after the holder
removed the marked one is never taken for it, and the set never holds
more than the deleting cells that are still cached.
- New `recycle_deleting_blocks()` rechecks those candidates and removes
the ones that became releasable, skipping only blocks that are still
being downloaded (an `EMPTY` block that nobody downloads is removed
too). `run_background_gc()` calls it every round, bounded by the
existing `file_cache_remove_block_qps_limit`, so a stranded block is now
reclaimed within one GC interval (100ms by default) instead of never.
Removal is asynchronous, so the storage is not touched while the cache
lock is held.
- Two bvars per cache path: `file_cache_recycle_deleting_blocks` counts
the blocks the sweep had to reclaim, i.e. blocks whose last reference
was dropped outside the holder path, and the gauge
`file_cache_deleting_block_count` shows how many deleting blocks are
still waiting for their references to go away.
The sweep also uncovered a latent use-after-free in `remove()` itself.
When it is called on a block whose state is `DOWNLOADING`, it takes the
block out of its LRU queue and then returns early without erasing the
cell, leaving `cell->queue_iterator` pointing at a freed `std::list`
node. Nothing ever reset it, so the `remove()` that finally erased that
cell — from `release_cache_reference()` once the download finished —
read freed memory in `LRUQueue::remove()` and decremented the queue size
a second time. `move_to_end()`, `resize()` and `change_cache_type()`
would have done the same. The iterator is now reset when `remove()`
bails out, and an empty iterator simply means "already out of the
queue": the `DCHECK` in `remove()` that forbade it is gone, and
`change_cache_type()`, the one caller that still dereferenced it
unconditionally, now leaves such a cell out of every queue. ASAN in the
first CI run on this PR reported exactly this, because the sweep is the
first thing that reliably re-enters `remove()` for such a cell.
Two related cleanups, both of which reduce how often a block is treated
as busy in the first place:
- `clear_file_cache_impl()` now drops the pending LRU updates *before*
the scan rather than after it.
- `run_background_block_lru_update()` clears its batch right after
applying it instead of at the top of the next round.
And, since the existing log line made this impossible to diagnose from a
BE log, `clear_file_cache_impl()` now logs the hash of the blocks it has
to wait for, which was missing entirely.
Seven unit tests:
- three cover the sweep itself: blocks are dropped the way a non-holder
reference holder drops them, and the tests assert that they are then
reclaimed, that still-referenced blocks survive, that the batch limit is
honoured, and that `EMPTY` blocks are reclaimed too;
- one covers a block cached again at the same offset after the holder
removed the marked one, which the sweep must leave alone;
- three cover a block removed while it is downloading: removing it
twice, which fails under ASAN without the iterator fix, converting it to
another cache type before its holder releases it, and reclaiming it
through the sweep once the download is done.
---
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 33bb9cd1055..0e4b51ad161 100644
--- a/be/src/io/cache/block_file_cache.cpp
+++ b/be/src/io/cache/block_file_cache.cpp
@@ -403,6 +403,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>>(
@@ -990,6 +994,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);
}
@@ -1021,6 +1082,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
@@ -1029,8 +1095,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;
}
@@ -1041,7 +1108,6 @@ std::string BlockFileCache::clear_file_cache_impl(bool
sync_remove) {
++num_cells_to_delete;
}
}
- clear_need_update_lru_blocks();
}
std::stringstream ss;
@@ -1318,7 +1384,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);
}
}
}
@@ -1551,7 +1617,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);
}
}
}
@@ -1576,7 +1642,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);
}
}
}
@@ -1802,7 +1868,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()
@@ -1832,7 +1899,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);
@@ -1888,7 +1956,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();
@@ -1901,6 +1973,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);
}
@@ -2113,8 +2189,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,
@@ -2180,7 +2260,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);
@@ -2462,6 +2542,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;
@@ -2547,6 +2631,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 69c79c50e52..89c03129b18 100644
--- a/be/src/io/cache/block_file_cache.h
+++ b/be/src/io/cache/block_file_cache.h
@@ -33,6 +33,7 @@
#include <optional>
#include <thread>
#include <unordered_map>
+#include <unordered_set>
#include <vector>
#include "io/cache/block_file_cache_ttl_mgr.h"
@@ -425,7 +426,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);
@@ -542,6 +544,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;
@@ -586,6 +599,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;
@@ -653,6 +678,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 85935068f9c..549efd6b67e 100644
--- a/be/test/io/cache/block_file_cache_test.cpp
+++ b/be/test/io/cache/block_file_cache_test.cpp
@@ -3700,6 +3700,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]
