freemandealer commented on code in PR #49456: URL: https://github.com/apache/doris/pull/49456#discussion_r2174639236
########## be/src/io/cache/cache_lru_dumper.cpp: ########## @@ -0,0 +1,375 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#include "io/cache/cache_lru_dumper.h" + +#include "io/cache/block_file_cache.h" +#include "io/cache/cache_lru_dumper.h" + +namespace doris::io { +Status CacheLRUDumper::check_ofstream_status(std::ofstream& out, std::string& filename) { + if (!out.good()) { + std::ios::iostate state = out.rdstate(); + std::stringstream err_msg; + if (state & std::ios::eofbit) { + err_msg << "End of file reached."; + } + if (state & std::ios::failbit) { + err_msg << "Input/output operation failed, err_code: " << strerror(errno); + } + if (state & std::ios::badbit) { + err_msg << "Serious I/O error occurred, err_code: " << strerror(errno); + } + out.close(); + std::string warn_msg = fmt::format("dump lru writing failed, file={}, {}", filename, + err_msg.str().c_str()); + LOG(WARNING) << warn_msg; + return Status::InternalError<false>(warn_msg); + } + + return Status::OK(); +} + +Status CacheLRUDumper::check_ifstream_status(std::ifstream& in, std::string& filename) { + if (!in.good()) { + std::ios::iostate state = in.rdstate(); + std::stringstream err_msg; + if (state & std::ios::eofbit) { + err_msg << "End of file reached."; + } + if (state & std::ios::failbit) { + err_msg << "Input/output operation failed, err_code: " << strerror(errno); + } + if (state & std::ios::badbit) { + err_msg << "Serious I/O error occurred, err_code: " << strerror(errno); + } + in.close(); + std::string warn_msg = std::string( + fmt::format("dump lru reading failed, file={}, {}", filename, err_msg.str())); + LOG(WARNING) << warn_msg; + return Status::InternalError<false>(warn_msg); + } + + return Status::OK(); +} + +Status CacheLRUDumper::dump_one_lru_entry(std::ofstream& out, std::string& filename, + const UInt128Wrapper& hash, size_t offset, size_t size) { + // Dump file format description: + // +-----------------------------------------------+ + // | LRUDumpEntryGroupPb_1 | + // +-----------------------------------------------+ + // | LRUDumpEntryGroupPb_2 | + // +-----------------------------------------------+ + // | LRUDumpEntryGroupPb_3 | + // +-----------------------------------------------+ + // | ... | + // +-----------------------------------------------+ + // | LRUDumpEntryGroupPb_n | + // +-----------------------------------------------+ + // | LRUDumpMetaPb (queueName, Group<off,size>List)| + // +-----------------------------------------------+ + // | FOOTER_OFFSET (8Bytes) | + // +-----------------------------------------------+ + // | CHECKSUM (4Bytes)｜VERSION (1Byte)｜ MAGIC | + // +-----------------------------------------------+ + // + // why we are not using protobuf as a whole? + // AFAIK, current protobuf version dose not support streaming mode, + // so that we need to store all the message in memory which will + // consume loads of RAMs. + // Instead, we use protobuf serialize each of the single entry + // and provide the version field in the footer for upgrade + + ::doris::io::cache::LRUDumpEntryPb* entry = _current_dump_group.add_entries(); + ::doris::io::cache::UInt128WrapperPb* hash_pb = entry->mutable_hash(); + hash_pb->set_high(hash.high()); + hash_pb->set_low(hash.low()); + entry->set_offset(offset); + entry->set_size(size); + + _current_dump_group_count++; + if (_current_dump_group_count >= 10000) { + RETURN_IF_ERROR(flush_current_group(out, filename)); + } + return Status::OK(); +} + +Status CacheLRUDumper::flush_current_group(std::ofstream& out, std::string& filename) { + if (_current_dump_group_count == 0) { + return Status::OK(); + } + + // Record current position as group start offset + size_t group_start = out.tellp(); + + // Serialize and write the group + std::string serialized; + LOG(INFO) << "Before serialization: " << _current_dump_group.DebugString(); + if (!_current_dump_group.SerializeToString(&serialized)) { + return Status::InternalError<false>("Failed to serialize LRUDumpEntryGroupPb"); + } + LOG(INFO) << "Serialized size: " << serialized.size(); + out.write(serialized.data(), serialized.size()); + RETURN_IF_ERROR(check_ofstream_status(out, filename)); + + // Record group metadata + ::doris::io::cache::EntryGroupOffsetSizePb* group_info = _dump_meta.add_group_offset_size(); + group_info->set_offset(group_start); + group_info->set_size(serialized.size()); + + // Reset for next group + _current_dump_group.Clear(); + _current_dump_group_count = 0; + return Status::OK(); +} + +Status CacheLRUDumper::finalize_dump(std::ofstream& out, size_t entry_num, + std::string& tmp_filename, std::string& final_filename, + size_t& file_size) { + // Flush any remaining entries + if (_current_dump_group_count > 0) { + RETURN_IF_ERROR(flush_current_group(out, tmp_filename)); + } + + // Write meta information + _dump_meta.set_entry_num(entry_num); + size_t meta_offset = out.tellp(); + LOG(INFO) << "dump meta: " << _dump_meta.DebugString(); + std::string meta_serialized; + if (!_dump_meta.SerializeToString(&meta_serialized)) { + std::string warn_msg = + fmt::format("Failed to serialize LRUDumpMetaPb, file={}", tmp_filename); + LOG(WARNING) << warn_msg; + return Status::InternalError<false>("warn_msg"); + } + out.write(meta_serialized.data(), meta_serialized.size()); + RETURN_IF_ERROR(check_ofstream_status(out, tmp_filename)); + + // Write footer + Footer footer; + footer.meta_offset = htole64(meta_offset); // Explicitly convert to little-endian + footer.checksum = 0; // TODO: Calculate checksum + footer.version = 1; + std::memcpy(footer.magic, "DOR", 3); + + out.write(reinterpret_cast<const char*>(&footer), sizeof(footer)); + RETURN_IF_ERROR(check_ofstream_status(out, tmp_filename)); + + out.close(); + + // Rename tmp to formal file + if (std::rename(tmp_filename.c_str(), final_filename.c_str()) != 0) { + std::remove(tmp_filename.c_str()); + file_size = std::filesystem::file_size(final_filename); + } + _dump_meta.Clear(); + _current_dump_group.Clear(); + _current_dump_group_count = 0; + + return Status::OK(); +} + +void CacheLRUDumper::dump_queue(LRUQueue& queue, const std::string& queue_name) { + Status st; + std::vector<std::tuple<UInt128Wrapper, size_t, size_t>> elements; + elements.reserve(config::file_cache_background_lru_dump_tail_record_num); + + { + std::lock_guard<std::mutex> lru_log_lock(_mgr->_mutex_lru_log); + size_t count = 0; + for (const auto& [hash, offset, size] : queue) { + if (count++ >= config::file_cache_background_lru_dump_tail_record_num) break; + elements.emplace_back(hash, offset, size); + } + } + + // Write to disk + int64_t duration_ns = 0; + std::uintmax_t file_size = 0; + { + SCOPED_RAW_TIMER(&duration_ns); + std::string tmp_filename = + fmt::format("{}/lru_dump_{}.bin.tmp", _mgr->_cache_base_path, queue_name); + std::string final_filename = + fmt::format("{}/lru_dump_{}.bin", _mgr->_cache_base_path, queue_name); + std::ofstream out(tmp_filename, std::ios::binary); + if (out) { + LOG(INFO) << "begin dump " << queue_name << "with " << elements.size() << " elements"; + for (const auto& [hash, offset, size] : elements) { + RETURN_IF_STATUS_ERROR(st, + dump_one_lru_entry(out, tmp_filename, hash, offset, size)); + } + RETURN_IF_STATUS_ERROR(st, finalize_dump(out, elements.size(), tmp_filename, + final_filename, file_size)); + } else { + LOG(WARNING) << "open lru dump file failed"; + } + } + *(_mgr->_lru_dump_latency_us) << (duration_ns / 1000); + LOG(INFO) << fmt::format("lru dump for {} size={} time={}us", queue_name, file_size, + duration_ns / 1000); +}; + +Status CacheLRUDumper::parse_dump_footer(std::ifstream& in, std::string& filename, + size_t& entry_num) { + size_t file_size = std::filesystem::file_size(filename); + + // Read footer + Footer footer; + size_t footer_size = sizeof(footer); + if (file_size < footer_size) { + std::string warn_msg = std::string(fmt::format( + "LRU dump file too small to contain footer, file={}, skip restore", filename)); + LOG(WARNING) << warn_msg; + return Status::InternalError<false>(warn_msg); + } + + in.seekg(-footer_size, std::ios::end); + in.read(reinterpret_cast<char*>(&footer), footer_size); + RETURN_IF_ERROR(check_ifstream_status(in, filename)); + + // Convert from little-endian to host byte order + footer.meta_offset = le64toh(footer.meta_offset); + + // Validate footer + if (footer.version != 1 || std::string(footer.magic, 3) != "DOR") { + std::string warn_msg = std::string(fmt::format( + "LRU dump file invalid footer format, file={}, skip restore", filename)); + LOG(WARNING) << warn_msg; + return Status::InternalError<false>(warn_msg); + } + + // Read meta + in.seekg(footer.meta_offset, std::ios::beg); + size_t meta_size = file_size - footer.meta_offset - footer_size; + if (meta_size <= 0) { + std::string warn_msg = std::string( + fmt::format("LRU dump file invalid meta size, file={}, skip restore", filename)); + LOG(WARNING) << warn_msg; + return Status::InternalError<false>(warn_msg); + } + std::string meta_serialized(meta_size, '\0'); + in.read(&meta_serialized[0], meta_serialized.size()); + RETURN_IF_ERROR(check_ifstream_status(in, filename)); + _parse_meta.Clear(); + _current_parse_group.Clear(); + if (!_parse_meta.ParseFromString(meta_serialized)) { + std::string warn_msg = std::string( + fmt::format("LRU dump file meta parse failed, file={}, skip restore", filename)); + LOG(WARNING) << warn_msg; + return Status::InternalError<false>(warn_msg); + } + LOG(INFO) << "parse meta: " << _parse_meta.DebugString(); + + entry_num = _parse_meta.entry_num(); + return Status::OK(); +} + +Status CacheLRUDumper::parse_one_lru_entry(std::ifstream& in, std::string& filename, + UInt128Wrapper& hash, size_t& offset, size_t& size) { + // Read next group if current is empty + if (_current_parse_group.entries_size() == 0) { + if (_parse_meta.group_offset_size_size() == 0) { + return Status::EndOfFile("No more entries"); + } + + auto group_info = _parse_meta.group_offset_size(0); + in.seekg(group_info.offset(), std::ios::beg); + std::string group_serialized(group_info.size(), '\0'); + in.read(&group_serialized[0], group_serialized.size()); + RETURN_IF_ERROR(check_ifstream_status(in, filename)); + + LOG(INFO) << "Deserializing group of size: " << group_serialized.size(); + if (!_current_parse_group.ParseFromString(group_serialized)) { + std::string warn_msg = + fmt::format("restore lru failed to parse group, file={}", filename); + LOG(WARNING) << warn_msg; + return Status::InternalError(warn_msg); + } + + // Remove processed group info + _parse_meta.mutable_group_offset_size()->erase(_parse_meta.group_offset_size().begin()); + } + + // Get next entry from current group + LOG(INFO) << "After deserialization: " << _current_parse_group.DebugString(); + auto entry = _current_parse_group.entries(0); + hash = UInt128Wrapper((static_cast<uint128_t>(entry.hash().high()) << 64) | entry.hash().low()); + offset = entry.offset(); + size = entry.size(); + + // Remove processed entry + _current_parse_group.mutable_entries()->erase(_current_parse_group.entries().begin()); + return Status::OK(); +} + +void CacheLRUDumper::restore_queue(LRUQueue& queue, const std::string& queue_name, + std::lock_guard<std::mutex>& cache_lock) { + Status st; + std::string filename = fmt::format("{}/lru_dump_{}.bin", _mgr->_cache_base_path, queue_name); + std::ifstream in(filename, std::ios::binary); + int64_t duration_ns = 0; + if (in) { + LOG(INFO) << "lru dump file is founded for " << queue_name << ". starting lru restore."; + + SCOPED_RAW_TIMER(&duration_ns); + size_t entry_num = 0; + RETURN_IF_STATUS_ERROR(st, parse_dump_footer(in, filename, entry_num)); + in.seekg(0, std::ios::beg); + UInt128Wrapper hash; + size_t offset, size; + for (int i = 0; i < entry_num; ++i) { + RETURN_IF_STATUS_ERROR(st, parse_one_lru_entry(in, filename, hash, offset, size)); + CacheContext ctx; + if (queue_name == "ttl") { + ctx.cache_type = FileCacheType::TTL; + // TODO(zhengyu): we haven't persist expiration time yet, use 3h default + // TODO(zhengyu): we don't use stats yet, see if this will cause any problem + ctx.expiration_time = 10800; + } else if (queue_name == "index") { + ctx.cache_type = FileCacheType::INDEX; + } else if (queue_name == "normal") { + ctx.cache_type = FileCacheType::NORMAL; + } else if (queue_name == "disposable") { + ctx.cache_type = FileCacheType::DISPOSABLE; + } else { + LOG_WARNING("unknown queue type"); + DCHECK(false); + return; + } + _mgr->add_cell(hash, ctx, offset, size, FileBlock::State::DOWNLOADED, cache_lock); Review Comment: no need -- 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: commits-unsubscr...@doris.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org --------------------------------------------------------------------- To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org For additional commands, e-mail: commits-h...@doris.apache.org
