github-actions[bot] commented on code in PR #65658: URL: https://github.com/apache/doris/pull/65658#discussion_r3600759872
########## be/src/io/cache/async_cache_write_service.cpp: ########## @@ -0,0 +1,495 @@ +// 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/async_cache_write_service.h" + +#include <algorithm> +#include <chrono> +#include <thread> +#include <utility> + +#include "common/logging.h" +#include "core/allocator.h" +#include "cpp/sync_point.h" +#include "io/cache/block_file_cache.h" +#include "runtime/thread_context.h" +#include "util/defer_op.h" +#include "util/time.h" + +namespace doris::io { + +using AsyncCacheWriteAllocator = Allocator<false, false, false, DefaultMemoryAllocator, true>; + +AsyncCacheWriteBuffer::AsyncCacheWriteBuffer(size_t size, + std::shared_ptr<MemTrackerLimiter> tracker) + : _size(size), _tracker(std::move(tracker)) { + AsyncCacheWriteAllocator allocator; + _data = reinterpret_cast<char*>(allocator.alloc(_size)); +} + +AsyncCacheWriteBuffer::~AsyncCacheWriteBuffer() { + SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_tracker); + AsyncCacheWriteAllocator allocator; + allocator.free(_data, _size); +} + +AsyncCacheWriteService::AsyncCacheWriteService(BlockFileCache* cache, + AsyncCacheWriteServiceOptions options) + : _cache(cache), + _options(std::make_shared<const AsyncCacheWriteServiceOptions>(options)), + _desired_worker_count(options.worker_count) { + DORIS_CHECK(_cache != nullptr); + DORIS_CHECK(options.worker_count > 0); + DORIS_CHECK(options.max_pending_tasks > 0); + DORIS_CHECK(options.batch_size > 0); + DORIS_CHECK(options.watchdog_warn_secs >= 0); + DORIS_CHECK(options.watchdog_drop_secs > options.watchdog_warn_secs); + + const char* prefix = _cache->get_base_path().c_str(); + _mem_tracker = MemTrackerLimiter::create_shared( + MemTrackerLimiter::Type::CACHE, + fmt::format("AsyncFileCacheWrite:{}", _cache->get_base_path())); + _pending_count_metric = std::make_shared<bvar::PassiveStatus<size_t>>( + prefix, "async_cache_write_pending_count", + [](void* service) { + return static_cast<AsyncCacheWriteService*>(service)->pending_count(); + }, + this); + _buffer_memory_metric = std::make_shared<bvar::PassiveStatus<int64_t>>( + prefix, "async_cache_write_buffer_memory_bytes", + [](void* service) { + return static_cast<AsyncCacheWriteService*>(service)->buffer_memory_bytes(); + }, + this); + _submitted_metric = + std::make_shared<bvar::Adder<uint64_t>>(prefix, "async_cache_write_submitted_total"); + _rejected_metric = + std::make_shared<bvar::Adder<uint64_t>>(prefix, "async_cache_write_rejected_total"); + _buffer_alloc_fail_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_buffer_alloc_fail_total"); + _latency_metric = + std::make_shared<bvar::LatencyRecorder>(prefix, "async_cache_write_latency_us"); + _skip_downloaded_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_skip_downloaded_total"); + _skip_downloading_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_skip_downloading_total"); + _skip_partial_overlap_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_skip_partial_overlap_total"); + _drop_stale_epoch_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_drop_stale_epoch_total"); + _skip_deleting_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_skip_deleting_total"); + _append_fail_metric = + std::make_shared<bvar::Adder<uint64_t>>(prefix, "async_cache_write_append_fail_total"); + _watchdog_timeout_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_watchdog_timeout_total"); +} + +AsyncCacheWriteService::~AsyncCacheWriteService() { + shutdown(); +} + +Status AsyncCacheWriteService::start() { + std::lock_guard resize_lock(_resize_mutex); + if (_shutdown_requested.load(std::memory_order_acquire) || + !_accepting.load(std::memory_order_acquire)) { + return Status::InternalError("async file cache write service is shutting down"); + } + if (_started.load(std::memory_order_acquire)) { + return Status::OK(); + } + + const size_t worker_count = _desired_worker_count.load(std::memory_order_acquire); + if (_worker_pool == nullptr) { + RETURN_IF_ERROR( + ThreadPoolBuilder(fmt::format("AsyncFileCacheWrite-{}", + std::hash<std::string> {}(_cache->get_base_path()))) + .set_min_threads(0) + .set_max_threads(static_cast<int>(worker_count)) + .set_max_queue_size(128) + .build(&_worker_pool)); + } else { + // A previous start may have created only part of the requested workers before returning an + // error. Reuse that pool and apply the latest desired size before filling the missing ids. + RETURN_IF_ERROR(_worker_pool->set_max_threads(static_cast<int>(worker_count))); + } + { + std::lock_guard state_lock(_worker_state_mutex); + if (_worker_scheduled.size() < worker_count) { + _worker_scheduled.resize(worker_count, false); + } + } + for (size_t worker_id = 0; worker_id < worker_count; ++worker_id) { + bool scheduled = false; + { + std::lock_guard state_lock(_worker_state_mutex); + scheduled = _worker_scheduled[worker_id]; + } + if (!scheduled) { + RETURN_IF_ERROR(_schedule_worker(worker_id)); + } + } + // Publish readiness only after every configured worker loop has been accepted by the pool. + _started.store(true, std::memory_order_release); + return Status::OK(); +} + +bool AsyncCacheWriteService::try_submit(AsyncCacheWriteTask task) { + _active_submitters.fetch_add(1, std::memory_order_acq_rel); + Defer submitter_done {[&]() { _active_submitters.fetch_sub(1, std::memory_order_acq_rel); }}; + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::try_submit:after_register", &task); + if (!_started.load(std::memory_order_acquire) || !_accepting.load(std::memory_order_acquire)) { + *_rejected_metric << 1; + return false; + } + + const auto options = _options.load(std::memory_order_acquire); + const size_t max_pending = options->max_pending_tasks; + size_t current = _pending_count.load(std::memory_order_relaxed); + while (current < max_pending) { + if (_pending_count.compare_exchange_weak(current, current + 1, std::memory_order_acq_rel, + std::memory_order_relaxed)) { + if (_queue.enqueue(std::move(task))) { + *_submitted_metric << 1; + _cv.notify_one(); + return true; + } + _pending_count.fetch_sub(1, std::memory_order_acq_rel); + *_rejected_metric << 1; + return false; + } + } + *_rejected_metric << 1; + return false; +} + +Status AsyncCacheWriteService::allocate_tracked_buffer(size_t size, + AsyncCacheWriteBufferPtr* buffer) { + DORIS_CHECK(buffer != nullptr); + DORIS_CHECK(size > 0); + Status injected_status; + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::allocate_tracked_buffer:inject_failure", + &injected_status); + if (!injected_status.ok()) { + *_buffer_alloc_fail_metric << 1; + return injected_status; + } + SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker); + Status status = Status::OK(); + try { + *buffer = AsyncCacheWriteBufferPtr(new AsyncCacheWriteBuffer(size, _mem_tracker)); + } catch (const std::exception& e) { + status = Status::MemoryAllocFailed("allocate async file cache write buffer failed: {}", + e.what()); + } + if (!status.ok()) { + *_buffer_alloc_fail_metric << 1; + } + return status; +} + +Status AsyncCacheWriteService::_schedule_worker(size_t worker_id) { + { + std::lock_guard lock(_worker_state_mutex); + if (_worker_scheduled.size() <= worker_id) { + _worker_scheduled.resize(worker_id + 1, false); + } + DORIS_CHECK(!_worker_scheduled[worker_id]); + _worker_scheduled[worker_id] = true; + } + Status status = _worker_pool->submit_func([this, worker_id]() { _worker_loop(worker_id); }); + if (!status.ok()) { + std::lock_guard lock(_worker_state_mutex); + _worker_scheduled[worker_id] = false; + _worker_state_cv.notify_all(); + } + return status; +} + +void AsyncCacheWriteService::_worker_loop(size_t worker_id) { + Defer mark_stopped {[&]() { + std::lock_guard lock(_worker_state_mutex); + _worker_scheduled[worker_id] = false; + _worker_state_cv.notify_all(); + }}; + + // Keep one consumer cursor per worker so concurrent consumers rotate across the queue's + // producer streams instead of rescanning them from scratch for every disk-write task. + moodycamel::ConsumerToken consumer_token(_queue); + + while (true) { + if (!_shutdown_requested.load(std::memory_order_acquire) && + worker_id >= _desired_worker_count.load(std::memory_order_acquire)) { + return; + } + + size_t processed = 0; + const auto options = _options.load(std::memory_order_acquire); + AsyncCacheWriteTask task; + while (processed < options->batch_size && _queue.try_dequeue(consumer_token, task)) { + ++processed; + Defer finish {[&]() { + _finish_task(task); + task = AsyncCacheWriteTask {}; + }}; + + const int64_t age_us = MonotonicMicros() - task.submit_ts_us; + const int64_t warn_us = options->watchdog_warn_secs * 1000 * 1000; + const int64_t drop_us = options->watchdog_drop_secs * 1000 * 1000; + if (age_us > warn_us) { + *_watchdog_timeout_metric << 1; + LOG(WARNING) << "Async file cache write task waited " << age_us + << " us, cache=" << _cache->get_base_path() + << ", hash=" << task.cache_hash.to_string() + << ", offset=" << task.file_offset; + } + if (age_us > drop_us) { + continue; + } + if (!is_current_write_epoch(task.write_epoch)) { + *_drop_stale_epoch_metric << 1; + continue; + } + + const int64_t start_us = MonotonicMicros(); + Status status = _write_one(task); + *_latency_metric << (MonotonicMicros() - start_us); + if (!status.ok()) { + LOG(WARNING) << "Async file cache write failed, cache=" << _cache->get_base_path() + << ", hash=" << task.cache_hash.to_string() + << ", offset=" << task.file_offset << ", size=" << task.file_size + << ", status=" << status; + } + } + + if (_shutdown_requested.load(std::memory_order_acquire) && + _pending_count.load(std::memory_order_acquire) == 0) { + return; + } + if (processed == 0) { + std::unique_lock lock(_cv_mutex); + _cv.wait_for(lock, std::chrono::milliseconds(1), [&]() { + return _queue.size_approx() != 0 || + _shutdown_requested.load(std::memory_order_acquire) || + worker_id >= _desired_worker_count.load(std::memory_order_acquire); + }); + } + } +} + +Status AsyncCacheWriteService::_write_one(const AsyncCacheWriteTask& task) { + DORIS_CHECK(task.buffer != nullptr); + DORIS_CHECK(task.file_size == task.buffer->size()); + if (!is_current_write_epoch(task.write_epoch)) { + *_drop_stale_epoch_metric << 1; + return Status::OK(); + } + + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::_write_one:before_get_or_set", &task); + ReadStatistics dummy_stats; + CacheContext context; + context.query_id = task.admission_ctx.query_id; + context.cache_type = task.admission_ctx.cache_type; + context.expiration_time = task.admission_ctx.expiration_time; + context.tablet_id = task.admission_ctx.tablet_id; + context.is_warmup = task.admission_ctx.is_warmup; + context.stats = &dummy_stats; + auto holder = _cache->get_or_set(task.cache_hash, task.file_offset, task.file_size, context); + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::_write_one:after_get_or_set", &task); + + if (!is_current_write_epoch(task.write_epoch)) { + *_drop_stale_epoch_metric << 1; + return Status::OK(); + } + + const size_t task_end = task.file_offset + task.file_size; + for (const auto& block : holder.file_blocks) { + if (block->range().left < task.file_offset || block->range().right >= task_end) { + *_skip_partial_overlap_metric << 1; + continue; + } + if (!is_current_write_epoch(task.write_epoch)) { + *_drop_stale_epoch_metric << 1; + return Status::OK(); + } + if (_cache->is_block_deleting(block)) { + *_skip_deleting_metric << 1; + continue; + } + + switch (block->state()) { + case FileBlock::State::DOWNLOADED: + *_skip_downloaded_metric << 1; + continue; + case FileBlock::State::DOWNLOADING: + *_skip_downloading_metric << 1; + continue; + case FileBlock::State::SKIP_CACHE: + continue; + case FileBlock::State::EMPTY: + break; + } + + if (block->get_or_set_downloader() != FileBlock::get_caller_id()) { + *_skip_downloading_metric << 1; + continue; + } + const size_t buffer_offset = block->range().left - task.file_offset; + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::_write_one:before_append", &task); + Status status = + block->append(Slice(task.buffer->data() + buffer_offset, block->range().size())); + if (!status.ok()) { + *_append_fail_metric << 1; + LOG(WARNING) << "Append async file cache block failed, cache=" + << _cache->get_base_path() << ", hash=" << task.cache_hash.to_string() + << ", offset=" << block->offset() << ", size=" << block->range().size() + << ", status=" << status; + continue; + } + status = block->finalize(); + if (!status.ok()) { + *_append_fail_metric << 1; + LOG(WARNING) << "Finalize async file cache block failed, cache=" + << _cache->get_base_path() << ", hash=" << task.cache_hash.to_string() + << ", offset=" << block->offset() << ", size=" << block->range().size() + << ", status=" << status; + } + } + return Status::OK(); +} + +void AsyncCacheWriteService::_finish_task(const AsyncCacheWriteTask& task) { + const size_t old_pending = _pending_count.fetch_sub(1, std::memory_order_acq_rel); + DORIS_CHECK(old_pending > 0); + if (task.on_finalized) { + task.on_finalized(task); + } +} + +Status AsyncCacheWriteService::resize_workers(size_t worker_count) { + if (worker_count == 0) { + return Status::InvalidArgument("async file cache write worker count must be positive"); + } + std::lock_guard resize_lock(_resize_mutex); + if (!_started.load(std::memory_order_acquire)) { + _desired_worker_count.store(worker_count, std::memory_order_release); + return Status::OK(); + } + if (_shutdown_requested.load(std::memory_order_acquire)) { + return Status::InternalError("async file cache write service is shutting down"); + } + + const size_t old_count = + _desired_worker_count.exchange(worker_count, std::memory_order_acq_rel); + if (old_count == worker_count) { + return Status::OK(); + } + _cv.notify_all(); + + if (worker_count < old_count) { + std::unique_lock state_lock(_worker_state_mutex); Review Comment: This wait can be permanent after a partial OS-thread creation failure. Doris `ThreadPool::do_submit()` enqueues a task and returns OK when `create_thread()` fails but another pool thread exists; `_schedule_worker()` then leaves that queued id marked scheduled. Here the existing thread is occupied indefinitely by worker 0, so queued higher-id loops can never start and clear their flags, while a shrink to 1 waits for exactly those flags. Please distinguish accepted from actually started loops and provide a cancellation/retirement path for queued persistent workers. ########## be/src/io/cache/cached_remote_file_reader_async_write.cpp: ########## @@ -0,0 +1,545 @@ +// 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 <bvar/bvar.h> +#include <glog/logging.h> + +#include <algorithm> +#include <cstring> +#include <memory> +#include <optional> +#include <utility> +#include <vector> + +#include "common/compiler_util.h" // IWYU pragma: keep +#include "common/config.h" +#include "cpp/sync_point.h" +#include "io/cache/async_cache_write_service.h" +#include "io/cache/block_file_cache.h" +#include "io/cache/cached_remote_file_reader.h" +#include "io/cache/inflight_write_buffer_index.h" +#include "io/io_common.h" +#include "runtime/runtime_profile.h" +#include "util/time.h" + +namespace doris::io { + +// These counters are shared by the synchronous and asynchronous indirect-read paths, so their +// definitions remain in cached_remote_file_reader.cpp. +extern bvar::Adder<uint64_t> g_read_cache_indirect_num; +extern bvar::Adder<uint64_t> g_read_cache_indirect_bytes; +extern bvar::Adder<uint64_t> g_read_cache_indirect_total_bytes; +extern bvar::Adder<uint64_t> g_read_cache_self_heal_on_not_found; + +namespace { + +bvar::Adder<uint64_t> g_cached_remote_reader_probe_total("cached_remote_file_reader_probe_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_probe_downloaded( + "cached_remote_file_reader_probe_hit_downloaded_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_probe_downloading( + "cached_remote_file_reader_probe_hit_downloading_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_probe_miss( + "cached_remote_file_reader_probe_miss_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_inflight_hit( + "cached_remote_file_reader_inflight_write_buffer_hit_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_async_skip_existing( + "cached_remote_file_reader_async_write_skip_inflight_existing_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_block_wait( + "cached_remote_file_reader_block_wait_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_block_wait_timeout( + "cached_remote_file_reader_block_wait_timeout_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_remote_after_dedup_miss( + "cached_remote_file_reader_remote_read_after_all_dedup_miss_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_middle_span_read_bytes( + "cached_remote_file_reader_middle_span_read_bytes"); +bvar::Adder<uint64_t> g_cached_remote_reader_middle_span_miss_bytes( + "cached_remote_file_reader_middle_span_miss_bytes"); + +} // namespace + +// One aligned cache block in the simplified read plan. REMOTE blocks delimit the single remote +// range; submit_write distinguishes real cache misses from blocks already being downloaded. +struct CachedRemoteFileReader::AsyncReadBlock { + enum class Source { + INFLIGHT, + CACHE, + DOWNLOADING, + REMOTE, + }; + + explicit AsyncReadBlock(FileBlock::Range range_) : range(range_) {} + + FileBlock::Range range; + Source source {Source::REMOTE}; + bool submit_write {false}; + std::shared_ptr<InflightWriteBufferEntry> inflight_entry; +}; + +// A read first checks inflight buffers and owns a cache probe result only when that lookup leaves +// uncovered blocks. first_remote_block and remote_block_count identify the inclusive first-to-last +// uncovered span, including any cache hits between those boundaries. +struct CachedRemoteFileReader::AsyncReadPlan { + AsyncReadPlan(uint64_t write_epoch_, size_t user_left_, size_t user_right_) + : write_epoch(write_epoch_), user_left(user_left_), user_right(user_right_) {} + + uint64_t write_epoch {0}; + size_t user_left {0}; + size_t user_right {0}; + std::optional<FileBlocksProbeResult> probe_result; + std::vector<AsyncReadBlock> blocks; + size_t first_remote_block {0}; + size_t remote_block_count {0}; +}; + +// Resolve mode on every read so online configuration changes affect existing readers. +CacheWriteMode CachedRemoteFileReader::_resolve_cache_write_mode(const IOContext* io_ctx) const { + if (io_ctx->is_dryrun || io_ctx->is_warmup || _should_read_from_peer(io_ctx)) { + return CacheWriteMode::SYNC_WRITE; + } + if (io_ctx->cache_write_mode_override.has_value()) { + return *io_ctx->cache_write_mode_override; + } + if (_cache_write_mode != CacheWriteMode::DEFAULT) { + return _cache_write_mode; + } + return config::enable_async_file_cache_write ? CacheWriteMode::ASYNC_WRITE + : CacheWriteMode::SYNC_WRITE; +} + +// Build a deliberately small plan. The inflight index is checked first so a fully covered read +// avoids BlockFileCache::probe and its cache mutex. Only an incomplete inflight lookup performs one +// whole-range probe whose result entries map directly to the logical plan blocks. +CachedRemoteFileReader::AsyncReadPlan CachedRemoteFileReader::_build_async_read_plan( + size_t remaining_offset, size_t remaining_size, uint64_t write_epoch, + const IOContext* io_ctx, ReadStatistics& stats) { + DORIS_CHECK(remaining_size > 0); + const auto [align_left, align_size] = s_align_size(remaining_offset, remaining_size, size()); + const size_t cache_block_size = static_cast<size_t>(config::file_cache_each_block_size); + DORIS_CHECK(cache_block_size > 0); + + AsyncReadPlan plan(write_epoch, remaining_offset, remaining_offset + remaining_size - 1); + + std::vector<size_t> block_offsets; + const size_t align_end = align_left + align_size; + for (size_t block_offset = align_left; block_offset < align_end; + block_offset += cache_block_size) { + const size_t block_size = std::min(cache_block_size, size() - block_offset); + DORIS_CHECK(block_size > 0); + const FileBlock::Range block_range(block_offset, block_offset + block_size - 1); + plan.blocks.emplace_back(block_range); + block_offsets.emplace_back(block_offset); + } + DORIS_CHECK(!plan.blocks.empty()); + + const bool inflight_index_enabled = + config::enable_async_file_cache_write_inflight_write_buffer_index; + bool all_blocks_inflight = inflight_index_enabled; + if (inflight_index_enabled) { + auto* inflight_index = _cache->inflight_write_buffer_index(); + DORIS_CHECK(inflight_index != nullptr); + auto inflight_results = inflight_index->lookup_all(_cache_hash, block_offsets, write_epoch); + DORIS_CHECK(inflight_results.size() == plan.blocks.size()); + for (size_t index = 0; index < plan.blocks.size(); ++index) { + auto& entry = inflight_results[index].entry; + if (!entry) { + all_blocks_inflight = false; + ++stats.inflight_write_buffer_index_miss; + continue; + } + + auto& read_block = plan.blocks[index]; + DORIS_CHECK(entry->buffer != nullptr); + DORIS_CHECK(entry->buffer_offset <= read_block.range.left); + DORIS_CHECK(entry->buffer_offset + entry->buffer_size > read_block.range.right); + read_block.source = AsyncReadBlock::Source::INFLIGHT; + read_block.inflight_entry = std::move(entry); + ++stats.inflight_write_buffer_index_hit; + g_cached_remote_reader_inflight_hit << 1; + } + } + if (all_blocks_inflight) { + return plan; + } + + CacheContext cache_context(io_ctx); + cache_context.stats = &stats; + cache_context.tablet_id = _tablet_id; + plan.probe_result.emplace(_cache->probe(_cache_hash, align_left, align_size, cache_context)); + g_cached_remote_reader_probe_total << 1; + const auto& probe_result = *plan.probe_result; + DORIS_CHECK(probe_result.file_blocks.size() == plan.blocks.size()); + + for (size_t index = 0; index < plan.blocks.size(); ++index) { + auto& read_block = plan.blocks[index]; + if (read_block.source == AsyncReadBlock::Source::INFLIGHT) { + continue; + } + + const auto& file_block = probe_result.file_blocks[index]; + bool is_miss = file_block == nullptr; + bool is_downloading = false; + if (file_block != nullptr) { + DORIS_CHECK(file_block->range().left == read_block.range.left); + DORIS_CHECK(file_block->range().right == read_block.range.right); + if (_cache->is_block_deleting(file_block)) { + is_miss = true; + } else { + switch (file_block->state()) { + case FileBlock::State::DOWNLOADED: + break; + case FileBlock::State::DOWNLOADING: + is_downloading = true; + break; + case FileBlock::State::EMPTY: + case FileBlock::State::SKIP_CACHE: + is_miss = true; + break; + } + } + } + + if (is_miss) { + read_block.submit_write = true; + ++stats.probe_miss; + g_cached_remote_reader_probe_miss << 1; + } else if (is_downloading) { + read_block.source = AsyncReadBlock::Source::DOWNLOADING; + ++stats.probe_downloading_hit; + g_cached_remote_reader_probe_downloading << 1; + continue; + } else { + read_block.source = AsyncReadBlock::Source::CACHE; + ++stats.probe_downloaded_hit; + g_cached_remote_reader_probe_downloaded << 1; + continue; + } + + if (plan.remote_block_count == 0) { + plan.first_remote_block = index; + } + plan.remote_block_count = index - plan.first_remote_block + 1; + } + return plan; +} + +// Copy one block available from inflight memory or cache. DOWNLOADING blocks outside the remote +// span retain the existing wait semantics. Any race or read failure asks the caller to replace the +// whole planned request with one remote read instead of incrementally repairing the range. +bool CachedRemoteFileReader::_materialize_async_block(const AsyncReadPlan& plan, size_t block_index, + size_t user_offset, Slice result, + const CacheContext& cache_context, + ReadStatistics& stats, + size_t* materialized_bytes, + bool* need_self_heal) { + DORIS_CHECK(block_index < plan.blocks.size()); + const auto& read_block = plan.blocks[block_index]; + DORIS_CHECK(read_block.source != AsyncReadBlock::Source::REMOTE); + DORIS_CHECK(materialized_bytes != nullptr); + DORIS_CHECK(need_self_heal != nullptr); + + const size_t copy_left = std::max(read_block.range.left, plan.user_left); + const size_t copy_right = std::min(read_block.range.right, plan.user_right); + if (copy_left > copy_right) { + return true; + } + const size_t copy_size = copy_right - copy_left + 1; + + if (read_block.source == AsyncReadBlock::Source::INFLIGHT) { + const auto& entry = read_block.inflight_entry; + DORIS_CHECK(entry != nullptr); + const size_t entry_offset = copy_left - entry->buffer_offset; + DORIS_CHECK(entry_offset + copy_size <= entry->buffer_size); + memcpy(result.data + (copy_left - user_offset), entry->buffer->data() + entry_offset, + copy_size); + *materialized_bytes += copy_size; + return true; + } + + DORIS_CHECK(plan.probe_result.has_value()); + DORIS_CHECK(block_index < plan.probe_result->file_blocks.size()); + const auto& file_block = plan.probe_result->file_blocks[block_index]; + DORIS_CHECK(file_block != nullptr); + DORIS_CHECK(file_block->range().left == read_block.range.left); + DORIS_CHECK(file_block->range().right == read_block.range.right); + if (_cache->is_block_deleting(file_block)) { + return false; + } + + FileBlock::State state = file_block->state(); + if (state == FileBlock::State::DOWNLOADING) { + DORIS_CHECK(read_block.source == AsyncReadBlock::Source::DOWNLOADING); + { + SCOPED_RAW_TIMER(&stats.remote_wait_timer); + state = file_block->wait(); + } + if (state != FileBlock::State::DOWNLOADED) { + ++stats.block_wait_timeout; + g_cached_remote_reader_block_wait_timeout << 1; + return false; + } + ++stats.block_wait_success; + g_cached_remote_reader_block_wait << 1; + } + if (state != FileBlock::State::DOWNLOADED) { + return false; + } + + Status status; + { + SCOPED_RAW_TIMER(&stats.local_read_timer); + status = file_block->read(Slice(result.data + (copy_left - user_offset), copy_size), + copy_left - file_block->range().left); + } + if (!status.ok()) { + if (status.is<ErrorCode::NOT_FOUND>()) { + *need_self_heal = true; + g_read_cache_self_heal_on_not_found << 1; + } + LOG_EVERY_N(WARNING, 100) + << "Read probed file cache block failed, falling back to remote. path=" + << path().native() << ", hash=" << _cache_hash.to_string() + << ", offset=" << file_block->offset() << ", status=" << status; + return false; + } + + _cache->touch_probe_block_if_cached(file_block, cache_context); + *materialized_bytes += copy_size; + return true; +} + +// Read cache/inflight blocks outside the first-to-last remote span. A side failure returns false; +// the caller then performs one remote read for the whole aligned request. +bool CachedRemoteFileReader::_materialize_async_cached_sides( + const AsyncReadPlan& plan, size_t user_offset, Slice result, + const CacheContext& cache_context, ReadStatistics& stats, + SourceReadBreakdown& source_read_breakdown, size_t* indirect_read_bytes, + bool* need_self_heal) { + DORIS_CHECK(indirect_read_bytes != nullptr); + DORIS_CHECK(need_self_heal != nullptr); + + size_t materialized_bytes = 0; + const auto materialize_range = [&](size_t begin, size_t end) { + for (size_t index = begin; index < end; ++index) { + if (!_materialize_async_block(plan, index, user_offset, result, cache_context, stats, + &materialized_bytes, need_self_heal)) { + return false; + } + } + return true; + }; + + const size_t prefix_end = + plan.remote_block_count == 0 ? plan.blocks.size() : plan.first_remote_block; + if (!materialize_range(0, prefix_end)) { + return false; + } + if (plan.remote_block_count != 0) { + const size_t suffix_begin = plan.first_remote_block + plan.remote_block_count; + DORIS_CHECK(suffix_begin <= plan.blocks.size()); + if (!materialize_range(suffix_begin, plan.blocks.size())) { + return false; + } + } + + *indirect_read_bytes += materialized_bytes; + source_read_breakdown.local_bytes += materialized_bytes; + return true; +} + +// Read the single aligned range from the first REMOTE block through the last REMOTE block. Cache +// hits inside that span are intentionally reread to keep one straightforward remote operation. +Status CachedRemoteFileReader::_read_async_remote_range( + const AsyncReadPlan& plan, size_t user_offset, Slice result, bool need_self_heal, + const IOContext* io_ctx, ReadStatistics& stats, SourceReadBreakdown& source_read_breakdown, + size_t* indirect_read_bytes, std::unique_ptr<char[]>* remote_buffer) { + DORIS_CHECK(indirect_read_bytes != nullptr); + DORIS_CHECK(remote_buffer != nullptr); + DORIS_CHECK(plan.remote_block_count > 0); + const size_t remote_end = plan.first_remote_block + plan.remote_block_count; + DORIS_CHECK(remote_end <= plan.blocks.size()); + const size_t remote_left = plan.blocks[plan.first_remote_block].range.left; + const size_t remote_right = plan.blocks[remote_end - 1].range.right; + const size_t remote_size = remote_right - remote_left + 1; + + stats.hit_cache = false; + stats.from_peer_cache = false; + if (need_self_heal) { + _cache->remove_if_cached_async(_cache_hash); + } + + const std::vector<FileBlockSPtr> no_peer_blocks; + RETURN_IF_ERROR(_execute_remote_read(no_peer_blocks, remote_left, remote_size, *remote_buffer, Review Comment: This call re-evaluates mutable peer-read configuration after `_resolve_cache_write_mode()` already selected the async/S3-only path. If peer reads are enabled in that window, `no_peer_blocks` makes the peer fetch return OK without producing `remote_buffer` (and the peer race can likewise win), so the next `DORIS_CHECK(*remote_buffer != nullptr)` aborts the BE process. Preserve the dispatch-time peer decision or call an explicitly S3-only helper here. ########## be/src/io/cache/cached_remote_file_reader_async_write.cpp: ########## @@ -0,0 +1,545 @@ +// 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 <bvar/bvar.h> +#include <glog/logging.h> + +#include <algorithm> +#include <cstring> +#include <memory> +#include <optional> +#include <utility> +#include <vector> + +#include "common/compiler_util.h" // IWYU pragma: keep +#include "common/config.h" +#include "cpp/sync_point.h" +#include "io/cache/async_cache_write_service.h" +#include "io/cache/block_file_cache.h" +#include "io/cache/cached_remote_file_reader.h" +#include "io/cache/inflight_write_buffer_index.h" +#include "io/io_common.h" +#include "runtime/runtime_profile.h" +#include "util/time.h" + +namespace doris::io { + +// These counters are shared by the synchronous and asynchronous indirect-read paths, so their +// definitions remain in cached_remote_file_reader.cpp. +extern bvar::Adder<uint64_t> g_read_cache_indirect_num; +extern bvar::Adder<uint64_t> g_read_cache_indirect_bytes; +extern bvar::Adder<uint64_t> g_read_cache_indirect_total_bytes; +extern bvar::Adder<uint64_t> g_read_cache_self_heal_on_not_found; + +namespace { + +bvar::Adder<uint64_t> g_cached_remote_reader_probe_total("cached_remote_file_reader_probe_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_probe_downloaded( + "cached_remote_file_reader_probe_hit_downloaded_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_probe_downloading( + "cached_remote_file_reader_probe_hit_downloading_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_probe_miss( + "cached_remote_file_reader_probe_miss_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_inflight_hit( + "cached_remote_file_reader_inflight_write_buffer_hit_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_async_skip_existing( + "cached_remote_file_reader_async_write_skip_inflight_existing_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_block_wait( + "cached_remote_file_reader_block_wait_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_block_wait_timeout( + "cached_remote_file_reader_block_wait_timeout_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_remote_after_dedup_miss( + "cached_remote_file_reader_remote_read_after_all_dedup_miss_total"); +bvar::Adder<uint64_t> g_cached_remote_reader_middle_span_read_bytes( + "cached_remote_file_reader_middle_span_read_bytes"); +bvar::Adder<uint64_t> g_cached_remote_reader_middle_span_miss_bytes( + "cached_remote_file_reader_middle_span_miss_bytes"); + +} // namespace + +// One aligned cache block in the simplified read plan. REMOTE blocks delimit the single remote +// range; submit_write distinguishes real cache misses from blocks already being downloaded. +struct CachedRemoteFileReader::AsyncReadBlock { + enum class Source { + INFLIGHT, + CACHE, + DOWNLOADING, + REMOTE, + }; + + explicit AsyncReadBlock(FileBlock::Range range_) : range(range_) {} + + FileBlock::Range range; + Source source {Source::REMOTE}; + bool submit_write {false}; + std::shared_ptr<InflightWriteBufferEntry> inflight_entry; +}; + +// A read first checks inflight buffers and owns a cache probe result only when that lookup leaves +// uncovered blocks. first_remote_block and remote_block_count identify the inclusive first-to-last +// uncovered span, including any cache hits between those boundaries. +struct CachedRemoteFileReader::AsyncReadPlan { + AsyncReadPlan(uint64_t write_epoch_, size_t user_left_, size_t user_right_) + : write_epoch(write_epoch_), user_left(user_left_), user_right(user_right_) {} + + uint64_t write_epoch {0}; + size_t user_left {0}; + size_t user_right {0}; + std::optional<FileBlocksProbeResult> probe_result; + std::vector<AsyncReadBlock> blocks; + size_t first_remote_block {0}; + size_t remote_block_count {0}; +}; + +// Resolve mode on every read so online configuration changes affect existing readers. +CacheWriteMode CachedRemoteFileReader::_resolve_cache_write_mode(const IOContext* io_ctx) const { + if (io_ctx->is_dryrun || io_ctx->is_warmup || _should_read_from_peer(io_ctx)) { + return CacheWriteMode::SYNC_WRITE; + } + if (io_ctx->cache_write_mode_override.has_value()) { + return *io_ctx->cache_write_mode_override; + } + if (_cache_write_mode != CacheWriteMode::DEFAULT) { + return _cache_write_mode; + } + return config::enable_async_file_cache_write ? CacheWriteMode::ASYNC_WRITE + : CacheWriteMode::SYNC_WRITE; +} + +// Build a deliberately small plan. The inflight index is checked first so a fully covered read +// avoids BlockFileCache::probe and its cache mutex. Only an incomplete inflight lookup performs one +// whole-range probe whose result entries map directly to the logical plan blocks. +CachedRemoteFileReader::AsyncReadPlan CachedRemoteFileReader::_build_async_read_plan( + size_t remaining_offset, size_t remaining_size, uint64_t write_epoch, + const IOContext* io_ctx, ReadStatistics& stats) { + DORIS_CHECK(remaining_size > 0); + const auto [align_left, align_size] = s_align_size(remaining_offset, remaining_size, size()); + const size_t cache_block_size = static_cast<size_t>(config::file_cache_each_block_size); + DORIS_CHECK(cache_block_size > 0); + + AsyncReadPlan plan(write_epoch, remaining_offset, remaining_offset + remaining_size - 1); + + std::vector<size_t> block_offsets; + const size_t align_end = align_left + align_size; + for (size_t block_offset = align_left; block_offset < align_end; + block_offset += cache_block_size) { + const size_t block_size = std::min(cache_block_size, size() - block_offset); + DORIS_CHECK(block_size > 0); + const FileBlock::Range block_range(block_offset, block_offset + block_size - 1); + plan.blocks.emplace_back(block_range); + block_offsets.emplace_back(block_offset); + } + DORIS_CHECK(!plan.blocks.empty()); + + const bool inflight_index_enabled = + config::enable_async_file_cache_write_inflight_write_buffer_index; + bool all_blocks_inflight = inflight_index_enabled; + if (inflight_index_enabled) { + auto* inflight_index = _cache->inflight_write_buffer_index(); + DORIS_CHECK(inflight_index != nullptr); + auto inflight_results = inflight_index->lookup_all(_cache_hash, block_offsets, write_epoch); + DORIS_CHECK(inflight_results.size() == plan.blocks.size()); + for (size_t index = 0; index < plan.blocks.size(); ++index) { + auto& entry = inflight_results[index].entry; + if (!entry) { + all_blocks_inflight = false; + ++stats.inflight_write_buffer_index_miss; + continue; + } + + auto& read_block = plan.blocks[index]; + DORIS_CHECK(entry->buffer != nullptr); + DORIS_CHECK(entry->buffer_offset <= read_block.range.left); + DORIS_CHECK(entry->buffer_offset + entry->buffer_size > read_block.range.right); + read_block.source = AsyncReadBlock::Source::INFLIGHT; + read_block.inflight_entry = std::move(entry); + ++stats.inflight_write_buffer_index_hit; + g_cached_remote_reader_inflight_hit << 1; + } + } + if (all_blocks_inflight) { + return plan; + } + + CacheContext cache_context(io_ctx); + cache_context.stats = &stats; + cache_context.tablet_id = _tablet_id; + plan.probe_result.emplace(_cache->probe(_cache_hash, align_left, align_size, cache_context)); + g_cached_remote_reader_probe_total << 1; + const auto& probe_result = *plan.probe_result; + DORIS_CHECK(probe_result.file_blocks.size() == plan.blocks.size()); + + for (size_t index = 0; index < plan.blocks.size(); ++index) { + auto& read_block = plan.blocks[index]; + if (read_block.source == AsyncReadBlock::Source::INFLIGHT) { + continue; + } + + const auto& file_block = probe_result.file_blocks[index]; + bool is_miss = file_block == nullptr; + bool is_downloading = false; + if (file_block != nullptr) { + DORIS_CHECK(file_block->range().left == read_block.range.left); + DORIS_CHECK(file_block->range().right == read_block.range.right); + if (_cache->is_block_deleting(file_block)) { + is_miss = true; + } else { + switch (file_block->state()) { + case FileBlock::State::DOWNLOADED: + break; + case FileBlock::State::DOWNLOADING: + is_downloading = true; + break; + case FileBlock::State::EMPTY: + case FileBlock::State::SKIP_CACHE: + is_miss = true; + break; + } + } + } + + if (is_miss) { + read_block.submit_write = true; + ++stats.probe_miss; + g_cached_remote_reader_probe_miss << 1; + } else if (is_downloading) { + read_block.source = AsyncReadBlock::Source::DOWNLOADING; + ++stats.probe_downloading_hit; + g_cached_remote_reader_probe_downloading << 1; + continue; + } else { + read_block.source = AsyncReadBlock::Source::CACHE; + ++stats.probe_downloaded_hit; + g_cached_remote_reader_probe_downloaded << 1; + continue; + } + + if (plan.remote_block_count == 0) { + plan.first_remote_block = index; + } + plan.remote_block_count = index - plan.first_remote_block + 1; + } + return plan; +} + +// Copy one block available from inflight memory or cache. DOWNLOADING blocks outside the remote +// span retain the existing wait semantics. Any race or read failure asks the caller to replace the +// whole planned request with one remote read instead of incrementally repairing the range. +bool CachedRemoteFileReader::_materialize_async_block(const AsyncReadPlan& plan, size_t block_index, + size_t user_offset, Slice result, + const CacheContext& cache_context, + ReadStatistics& stats, + size_t* materialized_bytes, + bool* need_self_heal) { + DORIS_CHECK(block_index < plan.blocks.size()); + const auto& read_block = plan.blocks[block_index]; + DORIS_CHECK(read_block.source != AsyncReadBlock::Source::REMOTE); + DORIS_CHECK(materialized_bytes != nullptr); + DORIS_CHECK(need_self_heal != nullptr); + + const size_t copy_left = std::max(read_block.range.left, plan.user_left); + const size_t copy_right = std::min(read_block.range.right, plan.user_right); + if (copy_left > copy_right) { + return true; + } + const size_t copy_size = copy_right - copy_left + 1; + + if (read_block.source == AsyncReadBlock::Source::INFLIGHT) { + const auto& entry = read_block.inflight_entry; + DORIS_CHECK(entry != nullptr); + const size_t entry_offset = copy_left - entry->buffer_offset; + DORIS_CHECK(entry_offset + copy_size <= entry->buffer_size); + memcpy(result.data + (copy_left - user_offset), entry->buffer->data() + entry_offset, + copy_size); + *materialized_bytes += copy_size; + return true; + } + + DORIS_CHECK(plan.probe_result.has_value()); + DORIS_CHECK(block_index < plan.probe_result->file_blocks.size()); + const auto& file_block = plan.probe_result->file_blocks[block_index]; + DORIS_CHECK(file_block != nullptr); + DORIS_CHECK(file_block->range().left == read_block.range.left); + DORIS_CHECK(file_block->range().right == read_block.range.right); + if (_cache->is_block_deleting(file_block)) { + return false; + } + + FileBlock::State state = file_block->state(); + if (state == FileBlock::State::DOWNLOADING) { + DORIS_CHECK(read_block.source == AsyncReadBlock::Source::DOWNLOADING); + { + SCOPED_RAW_TIMER(&stats.remote_wait_timer); + state = file_block->wait(); + } + if (state != FileBlock::State::DOWNLOADED) { + ++stats.block_wait_timeout; + g_cached_remote_reader_block_wait_timeout << 1; + return false; + } + ++stats.block_wait_success; + g_cached_remote_reader_block_wait << 1; + } + if (state != FileBlock::State::DOWNLOADED) { + return false; + } + + Status status; + { + SCOPED_RAW_TIMER(&stats.local_read_timer); + status = file_block->read(Slice(result.data + (copy_left - user_offset), copy_size), + copy_left - file_block->range().left); + } + if (!status.ok()) { + if (status.is<ErrorCode::NOT_FOUND>()) { + *need_self_heal = true; + g_read_cache_self_heal_on_not_found << 1; + } + LOG_EVERY_N(WARNING, 100) + << "Read probed file cache block failed, falling back to remote. path=" + << path().native() << ", hash=" << _cache_hash.to_string() + << ", offset=" << file_block->offset() << ", status=" << status; + return false; + } + + _cache->touch_probe_block_if_cached(file_block, cache_context); + *materialized_bytes += copy_size; + return true; +} + +// Read cache/inflight blocks outside the first-to-last remote span. A side failure returns false; +// the caller then performs one remote read for the whole aligned request. +bool CachedRemoteFileReader::_materialize_async_cached_sides( + const AsyncReadPlan& plan, size_t user_offset, Slice result, + const CacheContext& cache_context, ReadStatistics& stats, + SourceReadBreakdown& source_read_breakdown, size_t* indirect_read_bytes, + bool* need_self_heal) { + DORIS_CHECK(indirect_read_bytes != nullptr); + DORIS_CHECK(need_self_heal != nullptr); + + size_t materialized_bytes = 0; + const auto materialize_range = [&](size_t begin, size_t end) { + for (size_t index = begin; index < end; ++index) { + if (!_materialize_async_block(plan, index, user_offset, result, cache_context, stats, + &materialized_bytes, need_self_heal)) { + return false; + } + } + return true; + }; + + const size_t prefix_end = + plan.remote_block_count == 0 ? plan.blocks.size() : plan.first_remote_block; + if (!materialize_range(0, prefix_end)) { + return false; + } + if (plan.remote_block_count != 0) { + const size_t suffix_begin = plan.first_remote_block + plan.remote_block_count; + DORIS_CHECK(suffix_begin <= plan.blocks.size()); + if (!materialize_range(suffix_begin, plan.blocks.size())) { + return false; + } + } + + *indirect_read_bytes += materialized_bytes; + source_read_breakdown.local_bytes += materialized_bytes; + return true; +} + +// Read the single aligned range from the first REMOTE block through the last REMOTE block. Cache +// hits inside that span are intentionally reread to keep one straightforward remote operation. +Status CachedRemoteFileReader::_read_async_remote_range( + const AsyncReadPlan& plan, size_t user_offset, Slice result, bool need_self_heal, + const IOContext* io_ctx, ReadStatistics& stats, SourceReadBreakdown& source_read_breakdown, + size_t* indirect_read_bytes, std::unique_ptr<char[]>* remote_buffer) { + DORIS_CHECK(indirect_read_bytes != nullptr); + DORIS_CHECK(remote_buffer != nullptr); + DORIS_CHECK(plan.remote_block_count > 0); + const size_t remote_end = plan.first_remote_block + plan.remote_block_count; + DORIS_CHECK(remote_end <= plan.blocks.size()); + const size_t remote_left = plan.blocks[plan.first_remote_block].range.left; + const size_t remote_right = plan.blocks[remote_end - 1].range.right; + const size_t remote_size = remote_right - remote_left + 1; + + stats.hit_cache = false; + stats.from_peer_cache = false; + if (need_self_heal) { + _cache->remove_if_cached_async(_cache_hash); + } + + const std::vector<FileBlockSPtr> no_peer_blocks; + RETURN_IF_ERROR(_execute_remote_read(no_peer_blocks, remote_left, remote_size, *remote_buffer, + nullptr, stats, io_ctx)); + DORIS_CHECK(*remote_buffer != nullptr); + + const size_t copy_left = std::max(remote_left, plan.user_left); + const size_t copy_right = std::min(remote_right, plan.user_right); + if (copy_left <= copy_right) { + const size_t copy_size = copy_right - copy_left + 1; + memcpy(result.data + (copy_left - user_offset), + remote_buffer->get() + (copy_left - remote_left), copy_size); + *indirect_read_bytes += copy_size; + source_read_breakdown.remote_bytes += copy_size; + } + + size_t miss_bytes = 0; + for (size_t index = plan.first_remote_block; index < remote_end; ++index) { + if (plan.blocks[index].submit_write) { + miss_bytes += plan.blocks[index].range.size(); + } + } + if (miss_bytes > 0) { + g_cached_remote_reader_remote_after_dedup_miss << 1; + } + g_cached_remote_reader_middle_span_read_bytes << remote_size; + g_cached_remote_reader_middle_span_miss_bytes << miss_bytes; + return Status::OK(); +} + +// Submit exactly the real cache misses contained in the remote span. Inflight insertion happens after +// remote IO and immediately before enqueueing, so a concurrent owner wins without duplicate work. +void CachedRemoteFileReader::_submit_async_write_tasks(const AsyncReadPlan& plan, + const std::unique_ptr<char[]>& remote_buffer, + const IOContext* io_ctx, + ReadStatistics& stats) { + auto* service = _cache->async_write_service(); + auto* inflight_index = _cache->inflight_write_buffer_index(); + DORIS_CHECK(service != nullptr); + DORIS_CHECK(inflight_index != nullptr); + DORIS_CHECK(remote_buffer != nullptr); + + CacheContext cache_context(io_ctx); + CacheAdmissionContext admission_context { Review Comment: This handoff loses both query-scoped admission contracts. The worker reconstructs a context without `RemoteScanCacheWriteLimiter`, so positive `file_cache_query_limit_bytes` never accumulates and never switches the query to remote-only-on-miss. Retaining only `query_id` also fails the older percentage limit: once query teardown removes its `QueryFileCacheContext`, a queued task reserves from global LRU without `file_cache_query_limit_percent`. Admission needs to happen while the query state is alive (or the task must retain lifetime-safe admission state); please add async tests for both limits. ########## be/src/common/config.cpp: ########## @@ -1279,6 +1279,28 @@ DEFINE_mBool(file_cache_enable_only_warm_up_idx, "false"); DEFINE_Int32(file_cache_downloader_thread_num_min, "32"); DEFINE_Int32(file_cache_downloader_thread_num_max, "32"); +// async file cache write +DEFINE_mBool(enable_async_file_cache_write, "false"); +DEFINE_mInt32(async_file_cache_write_workers_per_disk, "16"); +DEFINE_mInt64(async_file_cache_write_max_pending_tasks_per_disk, "256"); +DEFINE_mInt32(async_file_cache_write_batch_size, "16"); +DEFINE_mInt64(async_file_cache_write_watchdog_warn_secs, "30"); +DEFINE_mInt64(async_file_cache_write_watchdog_drop_secs, "120"); +DEFINE_mBool(enable_async_file_cache_write_inflight_write_buffer_index, "true"); +DEFINE_Int32(async_file_cache_write_inflight_write_buffer_index_shard_count, "64"); +DEFINE_Validator(async_file_cache_write_workers_per_disk, + [](int32_t value) { return value > 0 && value <= 128; }); +DEFINE_Validator(async_file_cache_write_max_pending_tasks_per_disk, + [](int64_t value) { return value > 0; }); +DEFINE_Validator(async_file_cache_write_batch_size, [](int32_t value) { return value > 0; }); Review Comment: These new validators cannot enforce the advertised runtime/startup contract through the current config framework. Runtime `UPDATE_FIELD` calls the zero-argument validator before assigning `new_value`, so e.g. `batch_size=0` validates the old 16, commits 0, returns OK, and only logs when the service rejects it; a later cache construction hits its `DORIS_CHECK`. At startup the field map is alphabetical, so `drop_secs` is loaded before `warn_secs`: a valid final pair `warn=5, drop=10` is rejected against the default warn=30. Please validate the incoming coherent option set before committing it and add invalid-runtime plus lowered-pair startup tests. ########## be/src/io/cache/block_file_cache.cpp: ########## @@ -822,6 +844,108 @@ Status BlockFileCache::get_downloaded_blocks_if_fully_covered(const UInt128Wrapp return Status::OK(); } +FileBlocksProbeResult BlockFileCache::probe(const UInt128Wrapper& hash, size_t offset, size_t size, + const CacheContext& context) { + DORIS_CHECK(size > 0); + DORIS_CHECK(_max_file_block_size > 0); + const size_t end = offset + size; + DORIS_CHECK(end > offset); + std::lock_guard cache_lock(_mutex); + + auto file_iterator = _files.find(hash); + if (file_iterator == _files.end() && !_async_open_done) { + FileCacheKey key; + key.hash = hash; + key.meta.type = context.cache_type; + key.meta.expiration_time = context.expiration_time; + key.meta.tablet_id = context.tablet_id; + _storage->load_blocks_directly_unlocked(this, key, cache_lock); + file_iterator = _files.find(hash); + } + + FileBlocksByOffset* cached_blocks = nullptr; + FileBlocksByOffset::iterator cached_block; + if (file_iterator != _files.end()) { + DORIS_CHECK(!file_iterator->second.empty()); + cached_blocks = &file_iterator->second; + cached_block = cached_blocks->lower_bound(offset); + if (cached_block != cached_blocks->begin()) { + const auto& previous_range = std::prev(cached_block)->second.file_block->range(); + DORIS_CHECK(previous_range.right < offset); + } + } + + std::vector<FileBlockSPtr> result; + result.reserve(size / _max_file_block_size + (size % _max_file_block_size != 0)); + for (size_t block_offset = offset; block_offset < end;) { + const size_t block_size = std::min(_max_file_block_size, end - block_offset); + const FileBlock::Range expected_range(block_offset, block_offset + block_size - 1); + FileBlockSPtr file_block; + if (cached_blocks != nullptr && cached_block != cached_blocks->end() && + cached_block->second.file_block->range().left <= expected_range.right) { + file_block = cached_block->second.file_block; + DORIS_CHECK(file_block->range().left == expected_range.left); + DORIS_CHECK(file_block->range().right == expected_range.right); Review Comment: `probe()` cannot require existing cells to have the new planner's exact slot boundaries. `get_or_set()` accepts arbitrary ranges, and the S3 write-through path allocates each mutable `s3_write_buffer_size` part without requiring divisibility by `file_cache_each_block_size` (persisted cells can also retain an older block size). For example, 4 MiB cache blocks plus a 5 MiB upload leave a legal `[4,5 MiB)` cell; an async query probes `[4,8 MiB)` and aborts here. Please plan from actual intersecting ranges or fall back to the established path for noncanonical layouts, and cover a non-divisible writer/restored layout. ########## be/src/io/cache/block_file_cache_factory.cpp: ########## @@ -394,3 +410,78 @@ void FileCacheFactory::get_cache_stats_block(Block* block) { } // namespace io } // namespace doris + +namespace doris::config { + +namespace { + +/// Capture all mutable async-write fields after a config update. Returning one complete value +/// keeps every per-disk service on a coherent set of queue, worker, batch, and watchdog settings. +io::AsyncCacheWriteServiceOptions load_async_write_options_from_config() { + return io::AsyncCacheWriteServiceOptions { + .worker_count = static_cast<size_t>(async_file_cache_write_workers_per_disk), + .max_pending_tasks = + static_cast<size_t>(async_file_cache_write_max_pending_tasks_per_disk), + .batch_size = static_cast<size_t>(async_file_cache_write_batch_size), + .watchdog_warn_secs = async_file_cache_write_watchdog_warn_secs, + .watchdog_drop_secs = async_file_cache_write_watchdog_drop_secs, + }; +} + +/// Forward one changed config field through the explicit factory/service update interface. +/// @param config_name Name used only to identify failures in the log. +/// @param old_value Previous config value; equal values require no service update. +/// @param new_value Newly accepted config value. +template <typename T> +void update_async_write_options(const char* config_name, T old_value, T new_value) { + if (old_value == new_value) { + return; + } + auto* factory = ExecEnv::GetInstance()->file_cache_factory(); + if (factory == nullptr) { + return; + } + Status status = factory->update_async_write_options(load_async_write_options_from_config()); + if (!status.ok()) { + LOG(WARNING) << "Failed to apply async file cache write option " << config_name << " from " + << old_value << " to " << new_value << ": " << status.to_string(); + } +} + +} // namespace + +DEFINE_ON_UPDATE(enable_async_file_cache_write, [](bool old_value, bool new_value) { + if (old_value == new_value || !new_value) { + return; + } + auto* factory = io::FileCacheFactory::instance(); + if (factory == nullptr) { + return; + } + Status status = factory->start_async_write_services(); Review Comment: A service-start failure is swallowed after the global flag has already been committed. The HTTP update still reports OK, `_resolve_cache_write_mode()` selects async solely from that flag, and disks whose first worker failed to start reject every write submission; setting `true` again is also a callback no-op because old==new. Make enablement transactional/error-reporting, or dispatch per cache only after that service is ready and fall back to synchronous writeback on failed disks. ########## be/src/io/cache/block_file_cache_factory.cpp: ########## @@ -394,3 +410,78 @@ void FileCacheFactory::get_cache_stats_block(Block* block) { } // namespace io } // namespace doris + +namespace doris::config { + +namespace { + +/// Capture all mutable async-write fields after a config update. Returning one complete value +/// keeps every per-disk service on a coherent set of queue, worker, batch, and watchdog settings. +io::AsyncCacheWriteServiceOptions load_async_write_options_from_config() { + return io::AsyncCacheWriteServiceOptions { + .worker_count = static_cast<size_t>(async_file_cache_write_workers_per_disk), + .max_pending_tasks = + static_cast<size_t>(async_file_cache_write_max_pending_tasks_per_disk), + .batch_size = static_cast<size_t>(async_file_cache_write_batch_size), + .watchdog_warn_secs = async_file_cache_write_watchdog_warn_secs, + .watchdog_drop_secs = async_file_cache_write_watchdog_drop_secs, + }; +} + +/// Forward one changed config field through the explicit factory/service update interface. +/// @param config_name Name used only to identify failures in the log. +/// @param old_value Previous config value; equal values require no service update. +/// @param new_value Newly accepted config value. +template <typename T> +void update_async_write_options(const char* config_name, T old_value, T new_value) { + if (old_value == new_value) { + return; + } + auto* factory = ExecEnv::GetInstance()->file_cache_factory(); + if (factory == nullptr) { + return; + } + Status status = factory->update_async_write_options(load_async_write_options_from_config()); Review Comment: The factory mutex orders service application, but it does not order config mutation with this full-snapshot read. Two concurrent update requests can therefore apply stale state last: A commits batch=32 and captures pending=256, B commits pending=512 and applies `{32,512}`, then A acquires `_mtx` and reapplies `{32,256}`. Both requests report OK while services no longer match the final globals. Serialize snapshot/version creation with application, or reject older generations. ########## be/src/io/cache/async_cache_write_service.cpp: ########## @@ -0,0 +1,495 @@ +// 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/async_cache_write_service.h" + +#include <algorithm> +#include <chrono> +#include <thread> +#include <utility> + +#include "common/logging.h" +#include "core/allocator.h" +#include "cpp/sync_point.h" +#include "io/cache/block_file_cache.h" +#include "runtime/thread_context.h" +#include "util/defer_op.h" +#include "util/time.h" + +namespace doris::io { + +using AsyncCacheWriteAllocator = Allocator<false, false, false, DefaultMemoryAllocator, true>; + +AsyncCacheWriteBuffer::AsyncCacheWriteBuffer(size_t size, + std::shared_ptr<MemTrackerLimiter> tracker) + : _size(size), _tracker(std::move(tracker)) { + AsyncCacheWriteAllocator allocator; + _data = reinterpret_cast<char*>(allocator.alloc(_size)); +} + +AsyncCacheWriteBuffer::~AsyncCacheWriteBuffer() { + SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_tracker); + AsyncCacheWriteAllocator allocator; + allocator.free(_data, _size); +} + +AsyncCacheWriteService::AsyncCacheWriteService(BlockFileCache* cache, + AsyncCacheWriteServiceOptions options) + : _cache(cache), + _options(std::make_shared<const AsyncCacheWriteServiceOptions>(options)), + _desired_worker_count(options.worker_count) { + DORIS_CHECK(_cache != nullptr); + DORIS_CHECK(options.worker_count > 0); + DORIS_CHECK(options.max_pending_tasks > 0); + DORIS_CHECK(options.batch_size > 0); + DORIS_CHECK(options.watchdog_warn_secs >= 0); + DORIS_CHECK(options.watchdog_drop_secs > options.watchdog_warn_secs); + + const char* prefix = _cache->get_base_path().c_str(); + _mem_tracker = MemTrackerLimiter::create_shared( + MemTrackerLimiter::Type::CACHE, + fmt::format("AsyncFileCacheWrite:{}", _cache->get_base_path())); + _pending_count_metric = std::make_shared<bvar::PassiveStatus<size_t>>( + prefix, "async_cache_write_pending_count", + [](void* service) { + return static_cast<AsyncCacheWriteService*>(service)->pending_count(); + }, + this); + _buffer_memory_metric = std::make_shared<bvar::PassiveStatus<int64_t>>( + prefix, "async_cache_write_buffer_memory_bytes", + [](void* service) { + return static_cast<AsyncCacheWriteService*>(service)->buffer_memory_bytes(); + }, + this); + _submitted_metric = + std::make_shared<bvar::Adder<uint64_t>>(prefix, "async_cache_write_submitted_total"); + _rejected_metric = + std::make_shared<bvar::Adder<uint64_t>>(prefix, "async_cache_write_rejected_total"); + _buffer_alloc_fail_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_buffer_alloc_fail_total"); + _latency_metric = + std::make_shared<bvar::LatencyRecorder>(prefix, "async_cache_write_latency_us"); + _skip_downloaded_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_skip_downloaded_total"); + _skip_downloading_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_skip_downloading_total"); + _skip_partial_overlap_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_skip_partial_overlap_total"); + _drop_stale_epoch_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_drop_stale_epoch_total"); + _skip_deleting_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_skip_deleting_total"); + _append_fail_metric = + std::make_shared<bvar::Adder<uint64_t>>(prefix, "async_cache_write_append_fail_total"); + _watchdog_timeout_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_watchdog_timeout_total"); +} + +AsyncCacheWriteService::~AsyncCacheWriteService() { + shutdown(); +} + +Status AsyncCacheWriteService::start() { + std::lock_guard resize_lock(_resize_mutex); + if (_shutdown_requested.load(std::memory_order_acquire) || + !_accepting.load(std::memory_order_acquire)) { + return Status::InternalError("async file cache write service is shutting down"); + } + if (_started.load(std::memory_order_acquire)) { + return Status::OK(); + } + + const size_t worker_count = _desired_worker_count.load(std::memory_order_acquire); + if (_worker_pool == nullptr) { + RETURN_IF_ERROR( + ThreadPoolBuilder(fmt::format("AsyncFileCacheWrite-{}", + std::hash<std::string> {}(_cache->get_base_path()))) + .set_min_threads(0) + .set_max_threads(static_cast<int>(worker_count)) + .set_max_queue_size(128) + .build(&_worker_pool)); + } else { + // A previous start may have created only part of the requested workers before returning an + // error. Reuse that pool and apply the latest desired size before filling the missing ids. + RETURN_IF_ERROR(_worker_pool->set_max_threads(static_cast<int>(worker_count))); + } + { + std::lock_guard state_lock(_worker_state_mutex); + if (_worker_scheduled.size() < worker_count) { + _worker_scheduled.resize(worker_count, false); + } + } + for (size_t worker_id = 0; worker_id < worker_count; ++worker_id) { + bool scheduled = false; + { + std::lock_guard state_lock(_worker_state_mutex); + scheduled = _worker_scheduled[worker_id]; + } + if (!scheduled) { + RETURN_IF_ERROR(_schedule_worker(worker_id)); + } + } + // Publish readiness only after every configured worker loop has been accepted by the pool. + _started.store(true, std::memory_order_release); + return Status::OK(); +} + +bool AsyncCacheWriteService::try_submit(AsyncCacheWriteTask task) { + _active_submitters.fetch_add(1, std::memory_order_acq_rel); + Defer submitter_done {[&]() { _active_submitters.fetch_sub(1, std::memory_order_acq_rel); }}; + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::try_submit:after_register", &task); + if (!_started.load(std::memory_order_acquire) || !_accepting.load(std::memory_order_acquire)) { + *_rejected_metric << 1; + return false; + } + + const auto options = _options.load(std::memory_order_acquire); + const size_t max_pending = options->max_pending_tasks; + size_t current = _pending_count.load(std::memory_order_relaxed); + while (current < max_pending) { + if (_pending_count.compare_exchange_weak(current, current + 1, std::memory_order_acq_rel, + std::memory_order_relaxed)) { + if (_queue.enqueue(std::move(task))) { + *_submitted_metric << 1; + _cv.notify_one(); + return true; + } + _pending_count.fetch_sub(1, std::memory_order_acq_rel); + *_rejected_metric << 1; + return false; + } + } + *_rejected_metric << 1; + return false; +} + +Status AsyncCacheWriteService::allocate_tracked_buffer(size_t size, + AsyncCacheWriteBufferPtr* buffer) { + DORIS_CHECK(buffer != nullptr); + DORIS_CHECK(size > 0); + Status injected_status; + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::allocate_tracked_buffer:inject_failure", + &injected_status); + if (!injected_status.ok()) { + *_buffer_alloc_fail_metric << 1; + return injected_status; + } + SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker); + Status status = Status::OK(); + try { + *buffer = AsyncCacheWriteBufferPtr(new AsyncCacheWriteBuffer(size, _mem_tracker)); + } catch (const std::exception& e) { + status = Status::MemoryAllocFailed("allocate async file cache write buffer failed: {}", + e.what()); + } + if (!status.ok()) { + *_buffer_alloc_fail_metric << 1; + } + return status; +} + +Status AsyncCacheWriteService::_schedule_worker(size_t worker_id) { + { + std::lock_guard lock(_worker_state_mutex); + if (_worker_scheduled.size() <= worker_id) { + _worker_scheduled.resize(worker_id + 1, false); + } + DORIS_CHECK(!_worker_scheduled[worker_id]); + _worker_scheduled[worker_id] = true; + } + Status status = _worker_pool->submit_func([this, worker_id]() { _worker_loop(worker_id); }); + if (!status.ok()) { + std::lock_guard lock(_worker_state_mutex); + _worker_scheduled[worker_id] = false; + _worker_state_cv.notify_all(); + } + return status; +} + +void AsyncCacheWriteService::_worker_loop(size_t worker_id) { + Defer mark_stopped {[&]() { + std::lock_guard lock(_worker_state_mutex); + _worker_scheduled[worker_id] = false; + _worker_state_cv.notify_all(); + }}; + + // Keep one consumer cursor per worker so concurrent consumers rotate across the queue's + // producer streams instead of rescanning them from scratch for every disk-write task. + moodycamel::ConsumerToken consumer_token(_queue); + + while (true) { + if (!_shutdown_requested.load(std::memory_order_acquire) && + worker_id >= _desired_worker_count.load(std::memory_order_acquire)) { + return; + } + + size_t processed = 0; + const auto options = _options.load(std::memory_order_acquire); + AsyncCacheWriteTask task; + while (processed < options->batch_size && _queue.try_dequeue(consumer_token, task)) { + ++processed; + Defer finish {[&]() { + _finish_task(task); + task = AsyncCacheWriteTask {}; + }}; + + const int64_t age_us = MonotonicMicros() - task.submit_ts_us; + const int64_t warn_us = options->watchdog_warn_secs * 1000 * 1000; + const int64_t drop_us = options->watchdog_drop_secs * 1000 * 1000; + if (age_us > warn_us) { + *_watchdog_timeout_metric << 1; + LOG(WARNING) << "Async file cache write task waited " << age_us + << " us, cache=" << _cache->get_base_path() + << ", hash=" << task.cache_hash.to_string() + << ", offset=" << task.file_offset; + } + if (age_us > drop_us) { + continue; + } + if (!is_current_write_epoch(task.write_epoch)) { + *_drop_stale_epoch_metric << 1; + continue; + } + + const int64_t start_us = MonotonicMicros(); + Status status = _write_one(task); + *_latency_metric << (MonotonicMicros() - start_us); + if (!status.ok()) { + LOG(WARNING) << "Async file cache write failed, cache=" << _cache->get_base_path() + << ", hash=" << task.cache_hash.to_string() + << ", offset=" << task.file_offset << ", size=" << task.file_size + << ", status=" << status; + } + } + + if (_shutdown_requested.load(std::memory_order_acquire) && + _pending_count.load(std::memory_order_acquire) == 0) { + return; + } + if (processed == 0) { + std::unique_lock lock(_cv_mutex); + _cv.wait_for(lock, std::chrono::milliseconds(1), [&]() { Review Comment: With the default 16 workers per disk, this 1 ms timeout causes about 16,000 idle timed wakeups per second per cache disk (and scales linearly with disks), even when no cache write exists. It also appears to be masking a wakeup-protocol gap: enqueue notifies without synchronizing the queue predicate on `_cv_mutex`, so merely changing this to an indefinite wait could lose a notification between the predicate check and sleep. Please use a proper semaphore/CV sequence (or synchronize predicate updates) and block normally; at minimum, make any safety poll genuinely coarse. ########## be/src/io/cache/async_cache_write_service.cpp: ########## @@ -0,0 +1,495 @@ +// 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/async_cache_write_service.h" + +#include <algorithm> +#include <chrono> +#include <thread> +#include <utility> + +#include "common/logging.h" +#include "core/allocator.h" +#include "cpp/sync_point.h" +#include "io/cache/block_file_cache.h" +#include "runtime/thread_context.h" +#include "util/defer_op.h" +#include "util/time.h" + +namespace doris::io { + +using AsyncCacheWriteAllocator = Allocator<false, false, false, DefaultMemoryAllocator, true>; + +AsyncCacheWriteBuffer::AsyncCacheWriteBuffer(size_t size, + std::shared_ptr<MemTrackerLimiter> tracker) + : _size(size), _tracker(std::move(tracker)) { + AsyncCacheWriteAllocator allocator; + _data = reinterpret_cast<char*>(allocator.alloc(_size)); +} + +AsyncCacheWriteBuffer::~AsyncCacheWriteBuffer() { + SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_tracker); + AsyncCacheWriteAllocator allocator; + allocator.free(_data, _size); +} + +AsyncCacheWriteService::AsyncCacheWriteService(BlockFileCache* cache, + AsyncCacheWriteServiceOptions options) + : _cache(cache), + _options(std::make_shared<const AsyncCacheWriteServiceOptions>(options)), + _desired_worker_count(options.worker_count) { + DORIS_CHECK(_cache != nullptr); + DORIS_CHECK(options.worker_count > 0); + DORIS_CHECK(options.max_pending_tasks > 0); + DORIS_CHECK(options.batch_size > 0); + DORIS_CHECK(options.watchdog_warn_secs >= 0); + DORIS_CHECK(options.watchdog_drop_secs > options.watchdog_warn_secs); + + const char* prefix = _cache->get_base_path().c_str(); + _mem_tracker = MemTrackerLimiter::create_shared( + MemTrackerLimiter::Type::CACHE, + fmt::format("AsyncFileCacheWrite:{}", _cache->get_base_path())); + _pending_count_metric = std::make_shared<bvar::PassiveStatus<size_t>>( + prefix, "async_cache_write_pending_count", + [](void* service) { + return static_cast<AsyncCacheWriteService*>(service)->pending_count(); + }, + this); + _buffer_memory_metric = std::make_shared<bvar::PassiveStatus<int64_t>>( + prefix, "async_cache_write_buffer_memory_bytes", + [](void* service) { + return static_cast<AsyncCacheWriteService*>(service)->buffer_memory_bytes(); + }, + this); + _submitted_metric = + std::make_shared<bvar::Adder<uint64_t>>(prefix, "async_cache_write_submitted_total"); + _rejected_metric = + std::make_shared<bvar::Adder<uint64_t>>(prefix, "async_cache_write_rejected_total"); + _buffer_alloc_fail_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_buffer_alloc_fail_total"); + _latency_metric = + std::make_shared<bvar::LatencyRecorder>(prefix, "async_cache_write_latency_us"); + _skip_downloaded_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_skip_downloaded_total"); + _skip_downloading_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_skip_downloading_total"); + _skip_partial_overlap_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_skip_partial_overlap_total"); + _drop_stale_epoch_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_drop_stale_epoch_total"); + _skip_deleting_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_skip_deleting_total"); + _append_fail_metric = + std::make_shared<bvar::Adder<uint64_t>>(prefix, "async_cache_write_append_fail_total"); + _watchdog_timeout_metric = std::make_shared<bvar::Adder<uint64_t>>( + prefix, "async_cache_write_watchdog_timeout_total"); +} + +AsyncCacheWriteService::~AsyncCacheWriteService() { + shutdown(); +} + +Status AsyncCacheWriteService::start() { + std::lock_guard resize_lock(_resize_mutex); + if (_shutdown_requested.load(std::memory_order_acquire) || + !_accepting.load(std::memory_order_acquire)) { + return Status::InternalError("async file cache write service is shutting down"); + } + if (_started.load(std::memory_order_acquire)) { + return Status::OK(); + } + + const size_t worker_count = _desired_worker_count.load(std::memory_order_acquire); + if (_worker_pool == nullptr) { + RETURN_IF_ERROR( + ThreadPoolBuilder(fmt::format("AsyncFileCacheWrite-{}", + std::hash<std::string> {}(_cache->get_base_path()))) + .set_min_threads(0) + .set_max_threads(static_cast<int>(worker_count)) + .set_max_queue_size(128) + .build(&_worker_pool)); + } else { + // A previous start may have created only part of the requested workers before returning an + // error. Reuse that pool and apply the latest desired size before filling the missing ids. + RETURN_IF_ERROR(_worker_pool->set_max_threads(static_cast<int>(worker_count))); + } + { + std::lock_guard state_lock(_worker_state_mutex); + if (_worker_scheduled.size() < worker_count) { + _worker_scheduled.resize(worker_count, false); + } + } + for (size_t worker_id = 0; worker_id < worker_count; ++worker_id) { + bool scheduled = false; + { + std::lock_guard state_lock(_worker_state_mutex); + scheduled = _worker_scheduled[worker_id]; + } + if (!scheduled) { + RETURN_IF_ERROR(_schedule_worker(worker_id)); + } + } + // Publish readiness only after every configured worker loop has been accepted by the pool. + _started.store(true, std::memory_order_release); + return Status::OK(); +} + +bool AsyncCacheWriteService::try_submit(AsyncCacheWriteTask task) { + _active_submitters.fetch_add(1, std::memory_order_acq_rel); + Defer submitter_done {[&]() { _active_submitters.fetch_sub(1, std::memory_order_acq_rel); }}; + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::try_submit:after_register", &task); + if (!_started.load(std::memory_order_acquire) || !_accepting.load(std::memory_order_acquire)) { + *_rejected_metric << 1; + return false; + } + + const auto options = _options.load(std::memory_order_acquire); + const size_t max_pending = options->max_pending_tasks; + size_t current = _pending_count.load(std::memory_order_relaxed); + while (current < max_pending) { + if (_pending_count.compare_exchange_weak(current, current + 1, std::memory_order_acq_rel, + std::memory_order_relaxed)) { + if (_queue.enqueue(std::move(task))) { + *_submitted_metric << 1; + _cv.notify_one(); + return true; + } + _pending_count.fetch_sub(1, std::memory_order_acq_rel); + *_rejected_metric << 1; + return false; + } + } + *_rejected_metric << 1; + return false; +} + +Status AsyncCacheWriteService::allocate_tracked_buffer(size_t size, + AsyncCacheWriteBufferPtr* buffer) { + DORIS_CHECK(buffer != nullptr); + DORIS_CHECK(size > 0); + Status injected_status; + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::allocate_tracked_buffer:inject_failure", + &injected_status); + if (!injected_status.ok()) { + *_buffer_alloc_fail_metric << 1; + return injected_status; + } + SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(_mem_tracker); + Status status = Status::OK(); + try { + *buffer = AsyncCacheWriteBufferPtr(new AsyncCacheWriteBuffer(size, _mem_tracker)); + } catch (const std::exception& e) { + status = Status::MemoryAllocFailed("allocate async file cache write buffer failed: {}", + e.what()); + } + if (!status.ok()) { + *_buffer_alloc_fail_metric << 1; + } + return status; +} + +Status AsyncCacheWriteService::_schedule_worker(size_t worker_id) { + { + std::lock_guard lock(_worker_state_mutex); + if (_worker_scheduled.size() <= worker_id) { + _worker_scheduled.resize(worker_id + 1, false); + } + DORIS_CHECK(!_worker_scheduled[worker_id]); + _worker_scheduled[worker_id] = true; + } + Status status = _worker_pool->submit_func([this, worker_id]() { _worker_loop(worker_id); }); + if (!status.ok()) { + std::lock_guard lock(_worker_state_mutex); + _worker_scheduled[worker_id] = false; + _worker_state_cv.notify_all(); + } + return status; +} + +void AsyncCacheWriteService::_worker_loop(size_t worker_id) { + Defer mark_stopped {[&]() { + std::lock_guard lock(_worker_state_mutex); + _worker_scheduled[worker_id] = false; + _worker_state_cv.notify_all(); + }}; + + // Keep one consumer cursor per worker so concurrent consumers rotate across the queue's + // producer streams instead of rescanning them from scratch for every disk-write task. + moodycamel::ConsumerToken consumer_token(_queue); + + while (true) { + if (!_shutdown_requested.load(std::memory_order_acquire) && + worker_id >= _desired_worker_count.load(std::memory_order_acquire)) { + return; + } + + size_t processed = 0; + const auto options = _options.load(std::memory_order_acquire); + AsyncCacheWriteTask task; + while (processed < options->batch_size && _queue.try_dequeue(consumer_token, task)) { + ++processed; + Defer finish {[&]() { + _finish_task(task); + task = AsyncCacheWriteTask {}; + }}; + + const int64_t age_us = MonotonicMicros() - task.submit_ts_us; + const int64_t warn_us = options->watchdog_warn_secs * 1000 * 1000; + const int64_t drop_us = options->watchdog_drop_secs * 1000 * 1000; + if (age_us > warn_us) { + *_watchdog_timeout_metric << 1; + LOG(WARNING) << "Async file cache write task waited " << age_us + << " us, cache=" << _cache->get_base_path() + << ", hash=" << task.cache_hash.to_string() + << ", offset=" << task.file_offset; + } + if (age_us > drop_us) { + continue; + } + if (!is_current_write_epoch(task.write_epoch)) { + *_drop_stale_epoch_metric << 1; + continue; + } + + const int64_t start_us = MonotonicMicros(); + Status status = _write_one(task); + *_latency_metric << (MonotonicMicros() - start_us); + if (!status.ok()) { + LOG(WARNING) << "Async file cache write failed, cache=" << _cache->get_base_path() + << ", hash=" << task.cache_hash.to_string() + << ", offset=" << task.file_offset << ", size=" << task.file_size + << ", status=" << status; + } + } + + if (_shutdown_requested.load(std::memory_order_acquire) && + _pending_count.load(std::memory_order_acquire) == 0) { + return; + } + if (processed == 0) { + std::unique_lock lock(_cv_mutex); + _cv.wait_for(lock, std::chrono::milliseconds(1), [&]() { + return _queue.size_approx() != 0 || + _shutdown_requested.load(std::memory_order_acquire) || + worker_id >= _desired_worker_count.load(std::memory_order_acquire); + }); + } + } +} + +Status AsyncCacheWriteService::_write_one(const AsyncCacheWriteTask& task) { + DORIS_CHECK(task.buffer != nullptr); + DORIS_CHECK(task.file_size == task.buffer->size()); + if (!is_current_write_epoch(task.write_epoch)) { + *_drop_stale_epoch_metric << 1; + return Status::OK(); + } + + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::_write_one:before_get_or_set", &task); + ReadStatistics dummy_stats; + CacheContext context; + context.query_id = task.admission_ctx.query_id; + context.cache_type = task.admission_ctx.cache_type; + context.expiration_time = task.admission_ctx.expiration_time; + context.tablet_id = task.admission_ctx.tablet_id; + context.is_warmup = task.admission_ctx.is_warmup; + context.stats = &dummy_stats; + auto holder = _cache->get_or_set(task.cache_hash, task.file_offset, task.file_size, context); + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::_write_one:after_get_or_set", &task); + + if (!is_current_write_epoch(task.write_epoch)) { + *_drop_stale_epoch_metric << 1; + return Status::OK(); + } + + const size_t task_end = task.file_offset + task.file_size; + for (const auto& block : holder.file_blocks) { + if (block->range().left < task.file_offset || block->range().right >= task_end) { + *_skip_partial_overlap_metric << 1; + continue; + } + if (!is_current_write_epoch(task.write_epoch)) { + *_drop_stale_epoch_metric << 1; + return Status::OK(); + } + if (_cache->is_block_deleting(block)) { + *_skip_deleting_metric << 1; + continue; + } + + switch (block->state()) { + case FileBlock::State::DOWNLOADED: + *_skip_downloaded_metric << 1; + continue; + case FileBlock::State::DOWNLOADING: + *_skip_downloading_metric << 1; + continue; + case FileBlock::State::SKIP_CACHE: + continue; + case FileBlock::State::EMPTY: + break; + } + + if (block->get_or_set_downloader() != FileBlock::get_caller_id()) { + *_skip_downloading_metric << 1; + continue; + } + const size_t buffer_offset = block->range().left - task.file_offset; + TEST_SYNC_POINT_CALLBACK("AsyncCacheWriteService::_write_one:before_append", &task); Review Comment: There is still an invalidation window after these last epoch/deleting checks. An E worker can pause here, be marked deleting by E+1, then append/finalize; holder cleanup queues a delayed removal containing only the cache key. If an E+1 task refills the same `(hash, offset)` before background GC consumes that key, GC unconditionally deletes the shared file/meta for the new block, leaving a DOWNLOADED cell whose storage is gone. Make finalization/removal generation-aware (or share a linearization boundary), and test refill-before-recycle for this existing sync point. -- 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]
