liaoxin01 commented on code in PR #68348:
URL: https://github.com/apache/doris/pull/68348#discussion_r4272597777
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -1029,8 +1095,11 @@ 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="
- << " offset=" << cell->file_block->offset();
- cell->file_block->set_deleting();
+ << cell->file_block->get_hash_value().to_string()
+ << " offset=" << cell->file_block->offset()
+ << " use_count=" << cell->file_block.use_count() <<
" state="
+ <<
FileBlock::state_to_string(cell->file_block->state_unsafe());
Review Comment:
Dropped the use count and the state from this log line. It now only adds the
hash, which was missing from the original line, next to the offset; both are
fixed for the block's lifetime, so nothing is read without the block mutex here
any more.
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -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) {
Review Comment:
Agreed that examination is not bounded by `batch_limit`, but in practice the
set stays small. A key only exists for a block that was busy when a clear,
`try_release`, `reset_capacity` or `remove_if_cached` marked it, and it leaves
the set as soon as its cell is erased. Rowset deletion erases the segment cache
before it calls `remove_if_cached_async`, so readers are normally gone by then,
and query holders only live for one read. Long-lived busy entries need a manual
full clear on a loaded BE while many blocks are pinned by long-lived
references. Even then each examination is just a lookup and a `use_count` read,
and the scan shrinks once those references go away. I'd rather keep this fix
small and bound the scan with a FIFO plus rotation in a follow-up if it ever
shows up; the new `file_cache_deleting_block_count` gauge will make that
visible.
##########
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_gc_interval = config::file_cache_background_gc_interval_ms;
+ // Park the background GC thread: the explicit recycle_deleting_blocks()
calls below must be
+ // the only thing reclaiming blocks, otherwise the assertions race with it.
+ config::file_cache_background_gc_interval_ms = 3600 * 1000;
Review Comment:
Changed: the sweep tests now give the GC thread a zero removal budget
(`file_cache_remove_block_qps_limit = 0`) and keep the default interval, so
`~BlockFileCache()` can no longer wait an hour for it. The predicate-less wait
in the background threads is pre-existing; with production intervals a missed
notification only delays shutdown by one interval. All `BlockFileCacheTest`
cases pass under ASAN.
--
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]
