freemandealer commented on code in PR #63602:
URL: https://github.com/apache/doris/pull/63602#discussion_r3341299130
##########
be/src/io/cache/block_file_cache.cpp:
##########
@@ -679,27 +715,201 @@ std::string BlockFileCache::clear_file_cache_async() {
LOG(INFO) << "cell is not releasable, hash="
<< " offset=" << cell->file_block->offset();
cell->file_block->set_deleting();
- ++num_cells_wait_recycle;
+ ++result.num_cells_wait_recycle;
continue;
}
FileBlockSPtr file_block = cell->file_block;
if (file_block) {
std::lock_guard block_lock(file_block->_mutex);
remove(file_block, cache_lock, block_lock, false);
- ++num_cells_to_delete;
+ ++result.num_cells_to_delete;
}
}
clear_need_update_lru_blocks();
}
- std::stringstream ss;
- ss << "finish clear_file_cache_async, path=" << _cache_base_path
- << " num_files_all=" << num_files_all << " num_cells_all=" <<
num_cells_all
- << " num_cells_to_delete=" << num_cells_to_delete
- << " num_cells_wait_recycle=" << num_cells_wait_recycle;
- auto msg = ss.str();
- LOG(INFO) << msg;
_lru_dumper->remove_lru_dump_files();
+ return result;
+}
+
+size_t BlockFileCache::count_deleting_blocks_unlocked(
+ std::lock_guard<std::mutex>& /* cache_lock */) const {
+ size_t deleting_blocks = 0;
+ for (const auto& [_, offset_to_cell] : _files) {
+ for (const auto& [_1, cell] : offset_to_cell) {
+ if (cell.file_block->is_deleting()) {
+ ++deleting_blocks;
+ }
+ }
+ }
+ return deleting_blocks;
+}
+
+bool BlockFileCache::try_dequeue_recycle_key(FileCacheKey* key) {
+ std::lock_guard lock(_recycle_keys_mutex);
+ if (!_recycle_keys.try_dequeue(*key)) {
+ return false;
+ }
+ ++_recycle_remove_inflight;
+ return true;
+}
+
+Status BlockFileCache::remove_dequeued_recycle_key(const FileCacheKey& key) {
+ FileCacheStorageRemoveContextPtr context;
+ Status st;
+ int64_t duration_ns = 0;
+ {
+ SCOPED_RAW_TIMER(&duration_ns);
+ st = _storage->remove(key, &context);
+ }
+ *_storage_async_remove_latency_us << (duration_ns / 1000);
+ if (context) {
+ Status fence_st = context->wait();
+ if (st.ok() && !fence_st.ok()) {
+ st = fence_st;
+ }
+ }
+ {
+ std::lock_guard lock(_recycle_keys_mutex);
+ DCHECK_GT(_recycle_remove_inflight, 0);
+ --_recycle_remove_inflight;
+ }
+ _recycle_keys_cv.notify_all();
+ return st;
+}
+
+BlockFileCache::ClearFileCacheResult BlockFileCache::drain_recycle_keys(
+ const std::shared_ptr<ClearFileCacheCancelToken>& cancel_token) {
+ ClearFileCacheResult result;
+ auto is_cancelled = [&]() {
+ return cancel_token != nullptr &&
cancel_token->cancelled.load(std::memory_order_acquire);
+ };
+ FileCacheKey key;
+ while (!is_cancelled() && try_dequeue_recycle_key(&key)) {
+ Status st = remove_dequeued_recycle_key(key);
+ ++result.num_recycle_drained;
+ if (!st.ok()) {
+ LOG_WARNING("").error(st);
+ if (result.status.ok()) {
+ result.status = st;
+ }
+ }
+ if (is_cancelled()) {
+ result.cancelled = true;
+ break;
+ }
+ }
+ if (is_cancelled()) {
+ result.cancelled = true;
+ }
+ *_recycle_keys_length_recorder << _recycle_keys.size_approx();
+ return result;
+}
+
+bool BlockFileCache::recycle_keys_idle() {
+ std::unique_lock lock(_recycle_keys_mutex);
+ return _recycle_keys.size_approx() == 0 && _recycle_remove_inflight == 0;
+}
+
+void BlockFileCache::refresh_metrics_unlocked(std::lock_guard<std::mutex>&
cache_lock) {
+ _cur_cache_size_metrics->set_value(_cur_cache_size);
+ _cur_ttl_cache_size_metrics->set_value(_cur_ttl_size);
+
_cur_ttl_cache_lru_queue_cache_size_metrics->set_value(_ttl_queue.get_capacity(cache_lock));
+ _cur_ttl_cache_lru_queue_element_count_metrics->set_value(
+ _ttl_queue.get_elements_num(cache_lock));
+
_cur_normal_queue_cache_size_metrics->set_value(_normal_queue.get_capacity(cache_lock));
+
_cur_normal_queue_element_count_metrics->set_value(_normal_queue.get_elements_num(cache_lock));
+
_cur_index_queue_cache_size_metrics->set_value(_index_queue.get_capacity(cache_lock));
+
_cur_index_queue_element_count_metrics->set_value(_index_queue.get_elements_num(cache_lock));
+
_cur_disposable_queue_cache_size_metrics->set_value(_disposable_queue.get_capacity(cache_lock));
+ _cur_disposable_queue_element_count_metrics->set_value(
+ _disposable_queue.get_elements_num(cache_lock));
+}
+
+std::string BlockFileCache::clear_file_cache_sync(
+ std::shared_ptr<ClearFileCacheCancelToken> cancel_token) {
+ LOG(INFO) << "start clear_file_cache_sync, path=" << _cache_base_path;
+ std::lock_guard clear_lock(_clear_mutex);
+ using namespace std::chrono;
+ auto start = steady_clock::now();
+ ClearFileCacheResult result;
+
+ class TtlManagerPauseGuard {
+ public:
+ explicit TtlManagerPauseGuard(BlockFileCache* cache) : _cache(cache) {
+ _cache->pause_ttl_manager();
+ }
+ ~TtlManagerPauseGuard() { _cache->resume_ttl_manager(); }
+
+ private:
+ BlockFileCache* _cache;
+ } ttl_pause_guard(this);
+
+ append_clear_result(result, clear_file_cache_async_impl());
+
Review Comment:
yes, I'll fix it.
##########
be/src/service/http/action/file_cache_action.cpp:
##########
@@ -205,6 +219,27 @@ Status FileCacheAction::_handle_header(HttpRequest* req,
std::string* json_metri
}
void FileCacheAction::handle(HttpRequest* req) {
+ if (_is_sync_clear_request(req)) {
+ const std::string header_json(HEADER_JSON);
+ req->add_output_header(HttpHeaders::CONTENT_TYPE, header_json.c_str());
+ req->mark_send_reply();
+ auto cancel_token =
std::make_shared<io::BlockFileCache::ClearFileCacheCancelToken>();
+ req->set_cancel_callback(
+ [cancel_token] { cancel_token->cancelled.store(true,
std::memory_order_release); });
+ std::thread([this, req, cancel_token] {
+ std::string json_metrics;
Review Comment:
yes, I'll fix it.
--
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]
